feat(run): arm + prime alerts, mid-session catchup, late-start guard
Notify on IDLE→ARMED and ARMED→PRIMED so the user gets staged warnings before FIRE. If atm run starts mid-cycle on dark_green/dark_red, synth a catchup arm so the prime alert still fires (audit tagged catchup:true). If it starts on light_green/light_red, emit one late_start warn and skip the FSM feed — FIRE already passed. Extracted _handle_tick() as a pure helper (takes fsm + duck-typed notifier/audit Protocols) so the dispatch is unit-testable without mocking FanoutNotifier. 9 new tests cover arm, prime, refresh silence, catchup synth-arm (+ audit), and late-start on both directions. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
220
tests/test_handle_tick.py
Normal file
220
tests/test_handle_tick.py
Normal file
@@ -0,0 +1,220 @@
|
||||
"""Unit tests for atm.main._handle_tick.
|
||||
|
||||
Covers the six cases from the arm+prime notification plan:
|
||||
1. fresh BUY arm (turquoise)
|
||||
2. fresh SELL arm (yellow)
|
||||
3. BUY prime (turquoise → dark_green)
|
||||
4. silent refresh (turquoise → turquoise)
|
||||
5. catchup synth-arm from dark_green + prime alert, audit tagged catchup:true
|
||||
6. late-start guard (light_green from IDLE with first_accepted=True)
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from atm.main import _handle_tick
|
||||
from atm.notifier import Alert
|
||||
from atm.state_machine import State, StateMachine
|
||||
|
||||
|
||||
class FakeNotifier:
|
||||
"""Captures alerts instead of fanning out. send() is synchronous."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.alerts: list[Alert] = []
|
||||
|
||||
def send(self, alert: Alert) -> None:
|
||||
self.alerts.append(alert)
|
||||
|
||||
|
||||
class FakeAudit:
|
||||
"""Captures audit events in-memory."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.events: list[dict] = []
|
||||
|
||||
def log(self, event: dict) -> None:
|
||||
self.events.append(event)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 1. fresh BUY arm
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_handle_tick_arm_buy():
|
||||
fsm = StateMachine(lockout_s=60)
|
||||
notif = FakeNotifier()
|
||||
audit = FakeAudit()
|
||||
|
||||
tr = _handle_tick(fsm, "turquoise", 1.0, notif, audit, first_accepted=False)
|
||||
|
||||
assert tr is not None
|
||||
assert tr.next == State.ARMED_BUY
|
||||
assert tr.reason == "arm"
|
||||
assert len(notif.alerts) == 1
|
||||
assert notif.alerts[0].kind == "arm"
|
||||
assert notif.alerts[0].direction == "BUY"
|
||||
assert "turquoise" in notif.alerts[0].title
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 2. fresh SELL arm
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_handle_tick_arm_sell():
|
||||
fsm = StateMachine(lockout_s=60)
|
||||
notif = FakeNotifier()
|
||||
audit = FakeAudit()
|
||||
|
||||
tr = _handle_tick(fsm, "yellow", 1.0, notif, audit, first_accepted=False)
|
||||
|
||||
assert tr is not None
|
||||
assert tr.next == State.ARMED_SELL
|
||||
assert len(notif.alerts) == 1
|
||||
assert notif.alerts[0].kind == "arm"
|
||||
assert notif.alerts[0].direction == "SELL"
|
||||
assert "yellow" in notif.alerts[0].title
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 3. BUY prime — turquoise then dark_green
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_handle_tick_prime_buy():
|
||||
fsm = StateMachine(lockout_s=60)
|
||||
notif = FakeNotifier()
|
||||
audit = FakeAudit()
|
||||
|
||||
_handle_tick(fsm, "turquoise", 1.0, notif, audit, first_accepted=False)
|
||||
tr2 = _handle_tick(fsm, "dark_green", 2.0, notif, audit, first_accepted=False)
|
||||
|
||||
assert tr2 is not None
|
||||
assert tr2.next == State.PRIMED_BUY
|
||||
assert len(notif.alerts) == 2
|
||||
assert notif.alerts[0].kind == "arm"
|
||||
assert notif.alerts[1].kind == "prime"
|
||||
assert notif.alerts[1].direction == "BUY"
|
||||
# Non-catchup prime alert must not mention catchup
|
||||
assert "catchup" not in notif.alerts[1].title.lower()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 4. refresh is silent — second turquoise in ARMED_BUY emits no alert
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_handle_tick_refresh_no_alert():
|
||||
fsm = StateMachine(lockout_s=60)
|
||||
notif = FakeNotifier()
|
||||
audit = FakeAudit()
|
||||
|
||||
_handle_tick(fsm, "turquoise", 1.0, notif, audit, first_accepted=False)
|
||||
tr2 = _handle_tick(fsm, "turquoise", 2.0, notif, audit, first_accepted=False)
|
||||
|
||||
assert tr2 is not None
|
||||
assert tr2.reason == "refresh"
|
||||
assert len(notif.alerts) == 1 # only the first (real) arm
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 5. catchup synth-arm from dark_green with first_accepted=True
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_handle_tick_catchup_dark_green():
|
||||
fsm = StateMachine(lockout_s=60)
|
||||
notif = FakeNotifier()
|
||||
audit = FakeAudit()
|
||||
|
||||
tr = _handle_tick(fsm, "dark_green", 1.0, notif, audit, first_accepted=True)
|
||||
|
||||
# Synth arm + real prime feed → FSM lands in PRIMED_BUY
|
||||
assert tr is not None
|
||||
assert tr.next == State.PRIMED_BUY
|
||||
assert tr.reason == "prime"
|
||||
|
||||
# Two alerts: synth-arm (catchup-tagged) + prime
|
||||
assert len(notif.alerts) == 2
|
||||
assert notif.alerts[0].kind == "arm"
|
||||
assert notif.alerts[0].direction == "BUY"
|
||||
assert "catchup" in notif.alerts[0].title.lower() or "catchup" in notif.alerts[0].body.lower()
|
||||
assert notif.alerts[1].kind == "prime"
|
||||
assert notif.alerts[1].direction == "BUY"
|
||||
assert "catchup" in notif.alerts[1].title.lower() or "catchup" in notif.alerts[1].body.lower()
|
||||
|
||||
# Audit: both tick entries tagged catchup:true
|
||||
catchup_events = [e for e in audit.events if e.get("catchup")]
|
||||
assert len(catchup_events) == 2
|
||||
synth = next(e for e in catchup_events if e.get("synthesized_from") == "dark_green")
|
||||
assert synth["color"] == "turquoise"
|
||||
assert synth["reason"] == "arm"
|
||||
|
||||
|
||||
def test_handle_tick_catchup_dark_red():
|
||||
"""Mirror of dark_green — SELL side."""
|
||||
fsm = StateMachine(lockout_s=60)
|
||||
notif = FakeNotifier()
|
||||
audit = FakeAudit()
|
||||
|
||||
tr = _handle_tick(fsm, "dark_red", 1.0, notif, audit, first_accepted=True)
|
||||
|
||||
assert tr is not None
|
||||
assert tr.next == State.PRIMED_SELL
|
||||
assert len(notif.alerts) == 2
|
||||
assert notif.alerts[0].direction == "SELL"
|
||||
assert notif.alerts[1].direction == "SELL"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 6. late-start: light_green from IDLE with first_accepted=True
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_handle_tick_late_start():
|
||||
fsm = StateMachine(lockout_s=60)
|
||||
notif = FakeNotifier()
|
||||
audit = FakeAudit()
|
||||
|
||||
tr = _handle_tick(fsm, "light_green", 1.0, notif, audit, first_accepted=True)
|
||||
|
||||
# FSM untouched
|
||||
assert tr is None
|
||||
assert fsm.state == State.IDLE
|
||||
|
||||
# Exactly one late_start alert
|
||||
assert len(notif.alerts) == 1
|
||||
assert notif.alerts[0].kind == "late_start"
|
||||
assert notif.alerts[0].direction == "BUY"
|
||||
|
||||
# Audit has a late_start event
|
||||
assert any(e.get("event") == "late_start" for e in audit.events)
|
||||
|
||||
|
||||
def test_handle_tick_late_start_sell():
|
||||
"""Mirror of light_green — light_red triggers SELL late_start."""
|
||||
fsm = StateMachine(lockout_s=60)
|
||||
notif = FakeNotifier()
|
||||
audit = FakeAudit()
|
||||
|
||||
tr = _handle_tick(fsm, "light_red", 1.0, notif, audit, first_accepted=True)
|
||||
|
||||
assert tr is None
|
||||
assert fsm.state == State.IDLE
|
||||
assert len(notif.alerts) == 1
|
||||
assert notif.alerts[0].kind == "late_start"
|
||||
assert notif.alerts[0].direction == "SELL"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Extra guard: first_accepted=True with turquoise is a normal arm (decision 3A)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_handle_tick_first_turquoise_no_catchup_label():
|
||||
fsm = StateMachine(lockout_s=60)
|
||||
notif = FakeNotifier()
|
||||
audit = FakeAudit()
|
||||
|
||||
tr = _handle_tick(fsm, "turquoise", 1.0, notif, audit, first_accepted=True)
|
||||
|
||||
assert tr is not None
|
||||
assert tr.next == State.ARMED_BUY
|
||||
assert len(notif.alerts) == 1
|
||||
assert notif.alerts[0].kind == "arm"
|
||||
# Decision 3A: cannot distinguish fresh arm from catchup-at-arm-phase
|
||||
assert "catchup" not in notif.alerts[0].title.lower()
|
||||
assert "catchup" not in notif.alerts[0].body.lower()
|
||||
Reference in New Issue
Block a user