initial: scaffold atm trading monitor (Faza 1)
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Agent
0
tests/__init__.py
Normal file
0
tests/__init__.py
Normal file
107
tests/test_audit.py
Normal file
107
tests/test_audit.py
Normal file
@@ -0,0 +1,107 @@
|
||||
"""Tests for AuditLog."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from atm.audit import AuditLog
|
||||
|
||||
|
||||
def _dt(s: str) -> datetime:
|
||||
return datetime.fromisoformat(s)
|
||||
|
||||
|
||||
def test_writes_jsonl(tmp_path: Path) -> None:
|
||||
clock_dt = _dt("2026-04-15T10:00:00")
|
||||
log = AuditLog(tmp_path / "logs", clock=lambda: clock_dt)
|
||||
|
||||
events = [
|
||||
{"msg": "first", "ts": "2026-04-15T10:00:00"},
|
||||
{"msg": "second", "ts": "2026-04-15T10:01:00"},
|
||||
{"msg": "third", "ts": "2026-04-15T10:02:00"},
|
||||
]
|
||||
for e in events:
|
||||
log.log(e)
|
||||
log.close()
|
||||
|
||||
lines = (tmp_path / "logs" / "2026-04-15.jsonl").read_text().splitlines()
|
||||
assert len(lines) == 3
|
||||
for original, line in zip(events, lines):
|
||||
parsed = json.loads(line)
|
||||
assert parsed["msg"] == original["msg"]
|
||||
assert parsed["ts"] == original["ts"]
|
||||
|
||||
|
||||
def test_daily_rotation(tmp_path: Path) -> None:
|
||||
base = tmp_path / "logs"
|
||||
times = [
|
||||
_dt("2026-04-15T23:59:59"), # just before midnight
|
||||
_dt("2026-04-16T00:00:01"), # just after midnight
|
||||
]
|
||||
idx = 0
|
||||
|
||||
def clock() -> datetime:
|
||||
return times[min(idx, len(times) - 1)]
|
||||
|
||||
log = AuditLog(base, clock=clock)
|
||||
|
||||
log.log({"msg": "before"})
|
||||
idx = 1
|
||||
log.log({"msg": "after"})
|
||||
log.close()
|
||||
|
||||
file_15 = base / "2026-04-15.jsonl"
|
||||
file_16 = base / "2026-04-16.jsonl"
|
||||
assert file_15.exists(), "File for Apr 15 should exist"
|
||||
assert file_16.exists(), "File for Apr 16 should exist"
|
||||
|
||||
lines_15 = [json.loads(l) for l in file_15.read_text().splitlines()]
|
||||
lines_16 = [json.loads(l) for l in file_16.read_text().splitlines()]
|
||||
assert lines_15[0]["msg"] == "before"
|
||||
assert lines_16[0]["msg"] == "after"
|
||||
|
||||
|
||||
def test_line_buffered(tmp_path: Path) -> None:
|
||||
base = tmp_path / "logs"
|
||||
clock_dt = _dt("2026-04-15T12:00:00")
|
||||
log = AuditLog(base, clock=lambda: clock_dt)
|
||||
log.log({"msg": "hello", "ts": "2026-04-15T12:00:00"})
|
||||
# Do NOT call close() — file should already have content due to line buffering
|
||||
path = base / "2026-04-15.jsonl"
|
||||
assert os.stat(path).st_size > 0
|
||||
log.close()
|
||||
|
||||
|
||||
def test_adds_ts_when_missing(tmp_path: Path) -> None:
|
||||
clock_dt = _dt("2026-04-15T09:30:00")
|
||||
log = AuditLog(tmp_path / "logs", clock=lambda: clock_dt)
|
||||
log.log({"msg": "hi"})
|
||||
log.close()
|
||||
|
||||
lines = (tmp_path / "logs" / "2026-04-15.jsonl").read_text().splitlines()
|
||||
parsed = json.loads(lines[0])
|
||||
assert "ts" in parsed
|
||||
assert parsed["ts"] == "2026-04-15T09:30:00"
|
||||
|
||||
|
||||
def test_preserves_existing_ts(tmp_path: Path) -> None:
|
||||
clock_dt = _dt("2026-04-15T09:30:00")
|
||||
log = AuditLog(tmp_path / "logs", clock=lambda: clock_dt)
|
||||
log.log({"ts": "2026-01-01T00:00:00", "msg": "hi"})
|
||||
log.close()
|
||||
|
||||
lines = (tmp_path / "logs" / "2026-04-15.jsonl").read_text().splitlines()
|
||||
parsed = json.loads(lines[0])
|
||||
assert parsed["ts"] == "2026-01-01T00:00:00"
|
||||
|
||||
|
||||
def test_close_idempotent(tmp_path: Path) -> None:
|
||||
clock_dt = _dt("2026-04-15T10:00:00")
|
||||
log = AuditLog(tmp_path / "logs", clock=lambda: clock_dt)
|
||||
log.log({"msg": "x", "ts": "2026-04-15T10:00:00"})
|
||||
log.close()
|
||||
log.close() # should not raise
|
||||
221
tests/test_notifier.py
Normal file
221
tests/test_notifier.py
Normal file
@@ -0,0 +1,221 @@
|
||||
"""Tests for notifier module: FanoutNotifier, DiscordNotifier, TelegramNotifier."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from atm.notifier import Alert
|
||||
from atm.notifier.fanout import FanoutNotifier
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fake backends
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class FakeBackend:
|
||||
"""Configurable fake backend for testing."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
name: str = "fake",
|
||||
always_fail: bool = False,
|
||||
fail_first_n: int = 0,
|
||||
sleep_s: float = 0.0,
|
||||
) -> None:
|
||||
self.name = name
|
||||
self._always_fail = always_fail
|
||||
self._fail_first_n = fail_first_n
|
||||
self._sleep_s = sleep_s
|
||||
self._call_count = 0
|
||||
|
||||
def send(self, alert: Alert) -> None:
|
||||
self._call_count += 1
|
||||
if self._sleep_s:
|
||||
time.sleep(self._sleep_s)
|
||||
if self._always_fail:
|
||||
raise RuntimeError(f"{self.name}: simulated failure")
|
||||
if self._call_count <= self._fail_first_n:
|
||||
raise RuntimeError(f"{self.name}: simulated failure #{self._call_count}")
|
||||
|
||||
|
||||
def _alert(title: str = "test", kind: str = "trigger") -> Alert:
|
||||
return Alert(kind=kind, title=title, body="body text")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# FanoutNotifier tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_fanout_both_delivered(tmp_path: Path) -> None:
|
||||
dl = tmp_path / "dead.jsonl"
|
||||
b1 = FakeBackend("b1")
|
||||
b2 = FakeBackend("b2")
|
||||
fan = FanoutNotifier([b1, b2], dl, backoff_base=0.01)
|
||||
for i in range(3):
|
||||
fan.send(_alert(f"alert-{i}"))
|
||||
fan.stop(timeout=5.0)
|
||||
|
||||
s = fan.stats()
|
||||
assert s["b1"]["sent"] == 3
|
||||
assert s["b2"]["sent"] == 3
|
||||
assert s["b1"]["failed"] == 0
|
||||
assert s["b2"]["failed"] == 0
|
||||
|
||||
|
||||
def test_one_backend_down_other_delivers(tmp_path: Path) -> None:
|
||||
dl = tmp_path / "dead.jsonl"
|
||||
ok_backend = FakeBackend("ok")
|
||||
bad_backend = FakeBackend("bad", always_fail=True)
|
||||
fan = FanoutNotifier(
|
||||
[ok_backend, bad_backend], dl, max_retries=1, backoff_base=0.01
|
||||
)
|
||||
for i in range(2):
|
||||
fan.send(_alert(f"a{i}"))
|
||||
fan.stop(timeout=5.0)
|
||||
|
||||
s = fan.stats()
|
||||
assert s["ok"]["sent"] == 2
|
||||
assert s["bad"]["failed"] == 2
|
||||
|
||||
# dead letter file should have entries for the bad backend
|
||||
assert dl.exists()
|
||||
lines = [json.loads(l) for l in dl.read_text().splitlines()]
|
||||
assert all(e["backend"] == "bad" for e in lines)
|
||||
assert len(lines) == 2
|
||||
|
||||
|
||||
def test_dead_letter_on_exhausted_retries(tmp_path: Path) -> None:
|
||||
dl = tmp_path / "dead.jsonl"
|
||||
bad = FakeBackend("bad", always_fail=True)
|
||||
fan = FanoutNotifier([bad], dl, max_retries=3, backoff_base=0.01)
|
||||
fan.send(_alert("my-alert"))
|
||||
fan.stop(timeout=5.0)
|
||||
|
||||
s = fan.stats()
|
||||
assert s["bad"]["failed"] == 1
|
||||
# retries = max_retries (3 extra attempts after first)
|
||||
assert s["bad"]["retries"] == 3
|
||||
|
||||
assert dl.exists()
|
||||
lines = [json.loads(l) for l in dl.read_text().splitlines()]
|
||||
assert len(lines) == 1
|
||||
entry = lines[0]
|
||||
assert entry["backend"] == "bad"
|
||||
assert entry["alert_title"] == "my-alert"
|
||||
assert "error_str" in entry
|
||||
assert "timestamp" in entry
|
||||
|
||||
|
||||
def test_queue_drop_oldest(tmp_path: Path) -> None:
|
||||
dl = tmp_path / "dead.jsonl"
|
||||
# slow backend: each send takes 0.5s so queue fills fast
|
||||
slow = FakeBackend("slow", sleep_s=0.5)
|
||||
fan = FanoutNotifier([slow], dl, queue_size=2, backoff_base=0.01)
|
||||
|
||||
# Pump 10 alerts rapidly; worker can't keep up
|
||||
for i in range(10):
|
||||
fan.send(_alert(f"a{i}"))
|
||||
|
||||
fan.stop(timeout=10.0)
|
||||
s = fan.stats()
|
||||
assert s["slow"]["dropped"] > 0
|
||||
assert s["slow"]["sent"] <= 2 + 1 # queue_size + possibly 1 in-flight
|
||||
|
||||
|
||||
def test_retry_backoff_recovers(tmp_path: Path) -> None:
|
||||
dl = tmp_path / "dead.jsonl"
|
||||
# Fails only the very first call, succeeds after
|
||||
b = FakeBackend("b", fail_first_n=1)
|
||||
fan = FanoutNotifier([b], dl, max_retries=3, backoff_base=0.01)
|
||||
fan.send(_alert("recover"))
|
||||
fan.stop(timeout=5.0)
|
||||
|
||||
s = fan.stats()
|
||||
assert s["b"]["sent"] == 1
|
||||
assert s["b"]["retries"] == 1
|
||||
assert s["b"]["failed"] == 0
|
||||
assert not dl.exists()
|
||||
|
||||
|
||||
def test_stop_drains(tmp_path: Path) -> None:
|
||||
dl = tmp_path / "dead.jsonl"
|
||||
b = FakeBackend("b")
|
||||
fan = FanoutNotifier([b], dl, backoff_base=0.01)
|
||||
for i in range(5):
|
||||
fan.send(_alert(f"a{i}"))
|
||||
fan.stop(timeout=5.0)
|
||||
# All items should have been processed before stop returned
|
||||
assert fan.stats()["b"]["sent"] == 5
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# DiscordNotifier unit tests (no real HTTP)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class _MockResponse:
|
||||
def __init__(self, status_code: int, text: str = "") -> None:
|
||||
self.status_code = status_code
|
||||
self.text = text
|
||||
|
||||
|
||||
class _MockSession:
|
||||
def __init__(self, status_code: int = 204) -> None:
|
||||
self.status_code = status_code
|
||||
self.calls: list[dict] = []
|
||||
|
||||
def post(self, url: str, **kwargs):
|
||||
self.calls.append({"url": url, **kwargs})
|
||||
return _MockResponse(self.status_code)
|
||||
|
||||
|
||||
def test_discord_send_ok() -> None:
|
||||
from atm.notifier.discord import DiscordNotifier
|
||||
session = _MockSession(204)
|
||||
n = DiscordNotifier("https://discord.example/hook", session=session)
|
||||
n.send(_alert("Hello"))
|
||||
assert len(session.calls) == 1
|
||||
assert "**Hello**" in session.calls[0]["json"]["content"]
|
||||
|
||||
|
||||
def test_discord_429_raises() -> None:
|
||||
from atm.notifier.discord import DiscordNotifier
|
||||
n = DiscordNotifier("https://discord.example/hook", session=_MockSession(429))
|
||||
with pytest.raises(RuntimeError, match="429"):
|
||||
n.send(_alert("x"))
|
||||
|
||||
|
||||
def test_discord_5xx_raises() -> None:
|
||||
from atm.notifier.discord import DiscordNotifier
|
||||
n = DiscordNotifier("https://discord.example/hook", session=_MockSession(500))
|
||||
with pytest.raises(RuntimeError, match="500"):
|
||||
n.send(_alert("x"))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TelegramNotifier unit tests (no real HTTP)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_telegram_send_ok() -> None:
|
||||
from atm.notifier.telegram import TelegramNotifier
|
||||
session = _MockSession(200)
|
||||
n = TelegramNotifier("token", "chat123", session=session)
|
||||
n.send(_alert("Hi"))
|
||||
assert len(session.calls) == 1
|
||||
assert "*Hi*" in session.calls[0]["json"]["text"]
|
||||
|
||||
|
||||
def test_telegram_429_raises() -> None:
|
||||
from atm.notifier.telegram import TelegramNotifier
|
||||
n = TelegramNotifier("token", "chat123", session=_MockSession(429))
|
||||
with pytest.raises(RuntimeError, match="429"):
|
||||
n.send(_alert("x"))
|
||||
|
||||
|
||||
def test_telegram_5xx_raises() -> None:
|
||||
from atm.notifier.telegram import TelegramNotifier
|
||||
n = TelegramNotifier("token", "chat123", session=_MockSession(500))
|
||||
with pytest.raises(RuntimeError, match="500"):
|
||||
n.send(_alert("x"))
|
||||
327
tests/test_state_machine.py
Normal file
327
tests/test_state_machine.py
Normal file
@@ -0,0 +1,327 @@
|
||||
"""Tests for atm.state_machine."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from atm.state_machine import DotColor, State, StateMachine, Transition
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _drive(sm: StateMachine, events: list[tuple[DotColor, float]]) -> list[Transition]:
|
||||
return [sm.feed(color, ts) for color, ts in events]
|
||||
|
||||
|
||||
def _buy_sequence_to_primed(sm: StateMachine, start_ts: float = 1.0) -> None:
|
||||
"""Drive sm from IDLE → PRIMED_BUY (does not fire)."""
|
||||
sm.feed("turquoise", start_ts) # → ARMED_BUY
|
||||
sm.feed("dark_green", start_ts + 1) # → PRIMED_BUY
|
||||
|
||||
|
||||
def _sell_sequence_to_primed(sm: StateMachine, start_ts: float = 1.0) -> None:
|
||||
"""Drive sm from IDLE → PRIMED_SELL (does not fire)."""
|
||||
sm.feed("yellow", start_ts) # → ARMED_SELL
|
||||
sm.feed("dark_red", start_ts + 1) # → PRIMED_SELL
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 1. clean_buy
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_clean_buy() -> None:
|
||||
sm = StateMachine()
|
||||
|
||||
t1 = sm.feed("turquoise", 1.0)
|
||||
assert t1.prev == State.IDLE
|
||||
assert t1.next == State.ARMED_BUY
|
||||
assert t1.reason == "arm"
|
||||
assert sm.state == State.ARMED_BUY
|
||||
|
||||
t2 = sm.feed("gray", 2.0)
|
||||
assert t2.prev == State.ARMED_BUY
|
||||
assert t2.next == State.ARMED_BUY
|
||||
assert t2.reason == "persist"
|
||||
assert sm.state == State.ARMED_BUY
|
||||
|
||||
t3 = sm.feed("dark_green", 3.0)
|
||||
assert t3.prev == State.ARMED_BUY
|
||||
assert t3.next == State.PRIMED_BUY
|
||||
assert t3.reason == "prime"
|
||||
assert sm.state == State.PRIMED_BUY
|
||||
|
||||
t4 = sm.feed("light_green", 4.0)
|
||||
assert t4.prev == State.PRIMED_BUY
|
||||
assert t4.next == State.IDLE
|
||||
assert t4.reason == "fire"
|
||||
assert t4.trigger == "BUY"
|
||||
assert t4.locked is False
|
||||
assert sm.state == State.IDLE
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 2. clean_sell (mirror of clean_buy)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_clean_sell() -> None:
|
||||
sm = StateMachine()
|
||||
|
||||
t1 = sm.feed("yellow", 1.0)
|
||||
assert t1.prev == State.IDLE
|
||||
assert t1.next == State.ARMED_SELL
|
||||
assert t1.reason == "arm"
|
||||
|
||||
t2 = sm.feed("gray", 2.0)
|
||||
assert t2.prev == State.ARMED_SELL
|
||||
assert t2.next == State.ARMED_SELL
|
||||
assert t2.reason == "persist"
|
||||
|
||||
t3 = sm.feed("dark_red", 3.0)
|
||||
assert t3.prev == State.ARMED_SELL
|
||||
assert t3.next == State.PRIMED_SELL
|
||||
assert t3.reason == "prime"
|
||||
|
||||
t4 = sm.feed("light_red", 4.0)
|
||||
assert t4.prev == State.PRIMED_SELL
|
||||
assert t4.next == State.IDLE
|
||||
assert t4.reason == "fire"
|
||||
assert t4.trigger == "SELL"
|
||||
assert t4.locked is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 3. cooled
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_cooled() -> None:
|
||||
sm = StateMachine()
|
||||
_buy_sequence_to_primed(sm, start_ts=1.0)
|
||||
assert sm.state == State.PRIMED_BUY
|
||||
|
||||
t = sm.feed("gray", 10.0)
|
||||
assert t.prev == State.PRIMED_BUY
|
||||
assert t.next == State.IDLE
|
||||
assert t.reason == "cooled"
|
||||
assert t.trigger is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 4. opposite_rearm from ARMED_BUY
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_opposite_rearm_from_armed_buy() -> None:
|
||||
sm = StateMachine()
|
||||
sm.feed("turquoise", 1.0) # → ARMED_BUY
|
||||
assert sm.state == State.ARMED_BUY
|
||||
|
||||
t = sm.feed("yellow", 2.0)
|
||||
assert t.prev == State.ARMED_BUY
|
||||
assert t.next == State.ARMED_SELL
|
||||
assert t.reason == "opposite_rearm"
|
||||
assert sm.state == State.ARMED_SELL
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 5. opposite_rearm from PRIMED_BUY
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_opposite_rearm_from_primed_buy() -> None:
|
||||
sm = StateMachine()
|
||||
_buy_sequence_to_primed(sm, start_ts=1.0)
|
||||
assert sm.state == State.PRIMED_BUY
|
||||
|
||||
t = sm.feed("yellow", 5.0)
|
||||
assert t.prev == State.PRIMED_BUY
|
||||
assert t.next == State.ARMED_SELL
|
||||
assert t.reason == "opposite_rearm"
|
||||
assert sm.state == State.ARMED_SELL
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 6. lockout_same_direction
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_lockout_same_direction() -> None:
|
||||
sm = StateMachine(lockout_s=240)
|
||||
|
||||
# First fire at t=100
|
||||
_buy_sequence_to_primed(sm, start_ts=90.0)
|
||||
t_fire1 = sm.feed("light_green", 100.0)
|
||||
assert t_fire1.trigger == "BUY"
|
||||
assert t_fire1.locked is False
|
||||
|
||||
# Re-prime
|
||||
_buy_sequence_to_primed(sm, start_ts=110.0)
|
||||
|
||||
# Second fire at t=200 — inside lockout window (200 - 100 = 100 < 240)
|
||||
t_fire2 = sm.feed("light_green", 200.0)
|
||||
assert t_fire2.trigger == "BUY"
|
||||
assert t_fire2.locked is True
|
||||
assert t_fire2.next == State.IDLE
|
||||
|
||||
# Re-prime again
|
||||
_buy_sequence_to_primed(sm, start_ts=300.0)
|
||||
|
||||
# Third fire at t=341 — outside lockout (341 - 200 = 141 < 240, still locked)
|
||||
# Actually 341 - 100 would be outside but last fire is at t=200:
|
||||
# 341 - 200 = 141 < 240 → still locked
|
||||
# We need t > 200 + 240 = 440
|
||||
_buy_sequence_to_primed(sm, start_ts=430.0)
|
||||
t_fire3 = sm.feed("light_green", 441.0)
|
||||
assert t_fire3.trigger == "BUY"
|
||||
assert t_fire3.locked is False
|
||||
|
||||
|
||||
def test_lockout_same_direction_boundary() -> None:
|
||||
"""Spec requirement: fire BUY @ t=100; fire again @ t=341 → locked=False (241 >= 240)."""
|
||||
sm = StateMachine(lockout_s=240)
|
||||
|
||||
_buy_sequence_to_primed(sm, start_ts=90.0)
|
||||
t_fire1 = sm.feed("light_green", 100.0)
|
||||
assert t_fire1.locked is False
|
||||
|
||||
# Re-prime and fire just inside window: 339-100=239 < 240 → locked
|
||||
_buy_sequence_to_primed(sm, start_ts=110.0)
|
||||
t_locked = sm.feed("light_green", 339.0)
|
||||
assert t_locked.locked is True
|
||||
|
||||
# last_fire is now 339. Re-prime and fire just outside: 580-339=241 >= 240 → unlocked
|
||||
_buy_sequence_to_primed(sm, start_ts=340.0)
|
||||
t_free = sm.feed("light_green", 580.0)
|
||||
assert t_free.locked is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 7. lockout_does_not_block_opposite
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_lockout_does_not_block_opposite() -> None:
|
||||
sm = StateMachine(lockout_s=240)
|
||||
|
||||
# Fire BUY at t=100
|
||||
_buy_sequence_to_primed(sm, start_ts=90.0)
|
||||
sm.feed("light_green", 100.0)
|
||||
|
||||
# Drive SELL sequence — opposite direction must not be locked
|
||||
_sell_sequence_to_primed(sm, start_ts=110.0)
|
||||
t_sell = sm.feed("light_red", 200.0)
|
||||
assert t_sell.trigger == "SELL"
|
||||
assert t_sell.locked is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 8. phase_skip from ARMED_BUY (light_green without priming)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_phase_skip_armed_buy() -> None:
|
||||
sm = StateMachine()
|
||||
sm.feed("turquoise", 1.0) # → ARMED_BUY
|
||||
assert sm.state == State.ARMED_BUY
|
||||
|
||||
t = sm.feed("light_green", 2.0)
|
||||
assert t.prev == State.ARMED_BUY
|
||||
assert t.next == State.IDLE
|
||||
assert t.reason == "phase_skip"
|
||||
assert t.trigger is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 9. noise_from_idle
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_noise_from_idle() -> None:
|
||||
sm = StateMachine()
|
||||
t = sm.feed("dark_green", 1.0)
|
||||
assert t.prev == State.IDLE
|
||||
assert t.next == State.IDLE
|
||||
assert t.reason == "noise"
|
||||
assert t.trigger is None
|
||||
assert sm.state == State.IDLE
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 10. refresh_arm_ts
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_refresh_arm_ts() -> None:
|
||||
sm = StateMachine()
|
||||
sm.feed("turquoise", 1.0) # arm at t=1
|
||||
|
||||
t1 = sm.feed("turquoise", 5.0) # refresh at t=5
|
||||
assert t1.prev == State.ARMED_BUY
|
||||
assert t1.next == State.ARMED_BUY
|
||||
assert t1.reason == "refresh"
|
||||
assert t1.arm_ts == 5.0
|
||||
|
||||
t2 = sm.feed("turquoise", 9.0) # refresh again at t=9
|
||||
assert t2.arm_ts == 9.0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 11. exhaustive — parameterize over every (state, color) pair
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
ALL_STATES = list(State)
|
||||
ALL_COLORS: list[DotColor] = [
|
||||
"turquoise", "yellow", "dark_green", "dark_red",
|
||||
"light_green", "light_red", "gray",
|
||||
]
|
||||
|
||||
FIRE_DIRECTIONS: dict[str, str] = {
|
||||
State.PRIMED_BUY.value: "BUY",
|
||||
State.PRIMED_SELL.value: "SELL",
|
||||
}
|
||||
|
||||
VALID_STATES = set(State)
|
||||
|
||||
|
||||
def _sm_in_state(target: State) -> StateMachine:
|
||||
"""Return a fresh StateMachine already in the given state."""
|
||||
sm = StateMachine()
|
||||
match target:
|
||||
case State.IDLE:
|
||||
pass
|
||||
case State.ARMED_BUY:
|
||||
sm.feed("turquoise", 1.0)
|
||||
case State.ARMED_SELL:
|
||||
sm.feed("yellow", 1.0)
|
||||
case State.PRIMED_BUY:
|
||||
sm.feed("turquoise", 1.0)
|
||||
sm.feed("dark_green", 2.0)
|
||||
case State.PRIMED_SELL:
|
||||
sm.feed("yellow", 1.0)
|
||||
sm.feed("dark_red", 2.0)
|
||||
assert sm.state == target, f"Setup failed: wanted {target}, got {sm.state}"
|
||||
return sm
|
||||
|
||||
|
||||
@pytest.mark.parametrize("state", ALL_STATES)
|
||||
@pytest.mark.parametrize("color", ALL_COLORS)
|
||||
def test_exhaustive(state: State, color: DotColor) -> None:
|
||||
sm = _sm_in_state(state)
|
||||
t = sm.feed(color, 10.0)
|
||||
|
||||
# (a) resulting state is valid
|
||||
assert t.next in VALID_STATES, f"Invalid next state: {t.next}"
|
||||
|
||||
# (b) reason is non-empty
|
||||
assert t.reason, f"Empty reason for ({state}, {color})"
|
||||
|
||||
# (c) if fire, trigger matches direction
|
||||
if t.reason == "fire":
|
||||
expected_dir = FIRE_DIRECTIONS.get(state.value)
|
||||
assert t.trigger == expected_dir, (
|
||||
f"Wrong trigger for fire from {state}: got {t.trigger}, expected {expected_dir}"
|
||||
)
|
||||
Reference in New Issue
Block a user