feat(multi-chart): refactor _run_multi_tick + fix alert spam pe oscilație strip
Bug critic: _strips_match(tol=10) trip pe pulsații naturale de lățime ~18px între ticks (ex. 792↔810px). Fiecare trip → _commit_layout_change → reset FSM + alert Telegram + scheduler stop. Logul 2026-05-04.jsonl arăta 576 evenimente layout_change/zi, plus prime alerts repetate la dark_red/dark_green (FSM resetat înghite lockout-ul) și sincronizare cross-chart pe ambele FSM-uri simultan. Fix: - main.py:1511 — gate doar pe count change (len(new) != len(current)); count stabil → silent update sub_roi indiferent de jitter - main.py:1438 — silent=True pe alert layout_change (Telegram fără sunet) - 2 teste regresie noi: width oscillation 792↔810 + silent assertion - 2 teste async reparate: bootstrap _detect_strips_for_ctx pentru ScriptedDetector (regresie după ce _run_multi_tick a devenit unica cale de detecție) Plus refactor multi-chart pre-existent: layout.py modul nou, _detect_strips_for_ctx, ChartState per-chart FSM/Detector, ROI per-strip pe screenshots, scripts/diag_*. Verificat: 292 passed, 2 skipped în 10s. Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -257,6 +257,14 @@ def test_run_live_catchup_sell_from_gray_then_dark_red(monkeypatch, tmp_path):
|
||||
monkeypatch.setattr("atm.commands.TelegramPoller", _StubPoller)
|
||||
monkeypatch.setattr("atm.scheduler.ScreenshotScheduler", _StubScheduler)
|
||||
|
||||
# Bootstrap a single chart so _run_multi_tick populates ctx.charts on tick 1.
|
||||
# Frame is zeros → real detect_strips returns [] → without this, charts stays
|
||||
# empty and the ScriptedDetector loop never advances (regression after the
|
||||
# multi-chart refactor made _run_multi_tick the single detection path).
|
||||
from atm.config import ROI
|
||||
_bootstrap_strip = [ROI(x=0, y=0, w=10, h=10)]
|
||||
monkeypatch.setattr("atm.main._detect_strips_for_ctx", lambda c, f: _bootstrap_strip)
|
||||
|
||||
with pytest.raises(_StopLoop):
|
||||
_main.run_live(cfg, duration_s=None)
|
||||
|
||||
@@ -384,6 +392,11 @@ async def test_lifecycle_idle_armed_primed_autopoll_fire_stop(monkeypatch, tmp_p
|
||||
monkeypatch.setattr("atm.commands.TelegramPoller", FakePoller)
|
||||
monkeypatch.setattr("atm.scheduler.ScreenshotScheduler", lambda *a, **kw: fake_sched)
|
||||
|
||||
# Bootstrap ctx.charts on tick 1 — see test_run_live_catchup_sell for context.
|
||||
from atm.config import ROI
|
||||
_bootstrap_strip = [ROI(x=0, y=0, w=10, h=10)]
|
||||
monkeypatch.setattr("atm.main._detect_strips_for_ctx", lambda c, f: _bootstrap_strip)
|
||||
|
||||
with pytest.raises(_StopLoop):
|
||||
await _main.run_live_async(cfg, duration_s=None)
|
||||
|
||||
@@ -1628,3 +1641,451 @@ async def test_status_window_line_when_oh_enabled():
|
||||
status = [a for a in ctx.notifier.alerts if a.kind == "status"]
|
||||
body = status[0].body
|
||||
assert "fereastră: deschisă" in body
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Multi-chart split workspace tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_chart_state_defaults():
|
||||
"""ChartState's first_accepted defaults to True; last_saved_color to None."""
|
||||
import atm.main as _main
|
||||
from atm.config import ROI
|
||||
|
||||
roi = ROI(x=0, y=0, w=200, h=35)
|
||||
chart = _main.ChartState(
|
||||
chart_id="", sub_roi=roi, detector=MagicMock(), fsm=MagicMock(),
|
||||
)
|
||||
assert chart.first_accepted is True
|
||||
assert chart.last_saved_color is None
|
||||
assert chart.chart_id == ""
|
||||
assert chart.sub_roi is roi
|
||||
|
||||
|
||||
def test_alert_prefix_import():
|
||||
"""_alert_prefix lives in atm.notifier and produces the expected prefixes."""
|
||||
from atm.notifier import _alert_prefix
|
||||
assert _alert_prefix("") == ""
|
||||
assert _alert_prefix("left") == "[stânga] "
|
||||
assert _alert_prefix("right") == "[dreapta] "
|
||||
assert _alert_prefix("chart_0") == "[chart 1] "
|
||||
assert _alert_prefix("chart_2") == "[chart 3] "
|
||||
|
||||
|
||||
def test_save_annotated_frame_no_price_overlay(tmp_path):
|
||||
"""_save_annotated_frame must NOT draw the price overlay anymore.
|
||||
|
||||
Top-right area must contain no white pixels (text was rendered there before).
|
||||
"""
|
||||
import atm.main as _main
|
||||
from atm.config import ROI, YAxisCalib, ColorSpec
|
||||
|
||||
frame = np.zeros((200, 400, 3), dtype=np.uint8)
|
||||
cfg = types.SimpleNamespace(
|
||||
dot_roi=ROI(x=10, y=10, w=380, h=180),
|
||||
y_axis=YAxisCalib(p1_y=10, p1_price=100.0, p2_y=190, p2_price=50.0),
|
||||
colors={"background": ColorSpec(rgb=(0, 0, 0), tolerance=15.0)},
|
||||
)
|
||||
|
||||
fpath = _main._save_annotated_frame(
|
||||
frame, cfg, tmp_path, "test", now=123.0,
|
||||
dot_pos_abs=(200, 100), canary_ok=True,
|
||||
)
|
||||
assert fpath is not None
|
||||
saved = cv2.imread(str(fpath))
|
||||
assert saved is not None
|
||||
# Top-right corner where the "$..." text used to live (rows 0..40, cols 300..390).
|
||||
# The cyan ROI rect is on top edge but only 2px thick → vast majority of that
|
||||
# corner is still pure black. White text would average a much higher value.
|
||||
corner = saved[0:40, 300:390]
|
||||
# No bright white pixels (text was 255,255,255).
|
||||
white_pixels = ((corner[:, :, 0] > 200) & (corner[:, :, 1] > 200) & (corner[:, :, 2] > 200)).sum()
|
||||
assert white_pixels == 0, f"expected 0 white pixels (no price text), got {white_pixels}"
|
||||
|
||||
|
||||
def test_save_annotated_frame_uses_roi_param(tmp_path):
|
||||
"""When roi= is passed, the cyan rect is drawn there (not at cfg.dot_roi)."""
|
||||
import atm.main as _main
|
||||
from atm.config import ROI, ColorSpec
|
||||
|
||||
frame = np.zeros((200, 400, 3), dtype=np.uint8)
|
||||
cfg = types.SimpleNamespace(
|
||||
dot_roi=ROI(x=0, y=0, w=10, h=10), # tiny cfg ROI
|
||||
colors={"background": ColorSpec(rgb=(0, 0, 0), tolerance=15.0)},
|
||||
)
|
||||
custom_roi = ROI(x=100, y=50, w=200, h=80)
|
||||
fpath = _main._save_annotated_frame(
|
||||
frame, cfg, tmp_path, "rtest", now=123.0, roi=custom_roi,
|
||||
)
|
||||
assert fpath is not None
|
||||
saved = cv2.imread(str(fpath))
|
||||
# ROI rect color is (0, 255, 255) BGR. Find any pixel where G+R are saturated and B=0.
|
||||
edge = saved[49:52, 100:300]
|
||||
rect_present = ((edge[:, :, 0] < 50) & (edge[:, :, 1] > 200) & (edge[:, :, 2] > 200)).any()
|
||||
assert rect_present, "Expected ROI rect along custom roi top edge"
|
||||
# Also check the cfg.dot_roi rect (x=0..10, y=0..10) is NOT drawn — proves we used roi=custom.
|
||||
cfg_corner = saved[0:10, 0:10]
|
||||
rect_at_cfg = ((cfg_corner[:, :, 0] < 50) & (cfg_corner[:, :, 1] > 200) & (cfg_corner[:, :, 2] > 200)).any()
|
||||
assert not rect_at_cfg, "cfg.dot_roi rect must not appear when roi= override is used"
|
||||
|
||||
|
||||
def test_commit_layout_change_resets_fsm(monkeypatch):
|
||||
"""_commit_layout_change rebuilds ctx.charts, resets FSM, zeroes n_primed_global."""
|
||||
import atm.main as _main
|
||||
from atm.config import ROI
|
||||
from atm.state_machine import StateMachine, State
|
||||
|
||||
# Prebuild a ctx with one chart in ARMED_BUY.
|
||||
cfg = MagicMock()
|
||||
cfg.lockout_s = 60
|
||||
cfg.colors = {}
|
||||
cfg.debounce_depth = 3
|
||||
fsm_old = StateMachine(lockout_s=60)
|
||||
fsm_old.feed("turquoise", 1.0) # IDLE -> ARMED_BUY
|
||||
assert fsm_old.state == State.ARMED_BUY
|
||||
initial_chart = _main.ChartState(
|
||||
chart_id="", sub_roi=ROI(x=0, y=0, w=400, h=35),
|
||||
detector=MagicMock(), fsm=fsm_old,
|
||||
)
|
||||
|
||||
state = _main._LoopState()
|
||||
state.n_primed_global = 5 # any non-zero value
|
||||
|
||||
notifier_alerts: list = []
|
||||
audit_events: list = []
|
||||
|
||||
class _N:
|
||||
def send(self, a): notifier_alerts.append(a)
|
||||
|
||||
class _A:
|
||||
def log(self, e): audit_events.append(e)
|
||||
|
||||
class _S:
|
||||
is_running = True
|
||||
def stop(self):
|
||||
type(self).is_running = False
|
||||
|
||||
ctx = _main.RunContext(
|
||||
cfg=cfg, capture=lambda: None, canary=MagicMock(),
|
||||
detector=MagicMock(), fsm=fsm_old,
|
||||
notifier=_N(), audit=_A(), detection_log=_A(),
|
||||
scheduler=_S(), samples_dir=Path("."), fires_dir=Path("."),
|
||||
cmd_queue=MagicMock(), state=state,
|
||||
levels_extractor_factory=lambda *a, **kw: None,
|
||||
)
|
||||
ctx.charts = [initial_chart]
|
||||
|
||||
new_strips = [
|
||||
ROI(x=0, y=0, w=200, h=35),
|
||||
ROI(x=250, y=0, w=200, h=35),
|
||||
]
|
||||
_main._commit_layout_change(ctx, new_strips, now=100.0)
|
||||
|
||||
assert len(ctx.charts) == 2
|
||||
assert ctx.charts[0].chart_id == "left"
|
||||
assert ctx.charts[1].chart_id == "right"
|
||||
# New FSMs must be IDLE (fresh StateMachine)
|
||||
assert ctx.charts[0].fsm.state == State.IDLE
|
||||
assert ctx.charts[1].fsm.state == State.IDLE
|
||||
# Old fsm not reused
|
||||
assert ctx.charts[0].fsm is not fsm_old
|
||||
# n_primed_global reset
|
||||
assert state.n_primed_global == 0
|
||||
# layout_change audit + status alert
|
||||
assert any(e.get("event") == "layout_change" and e["new_n"] == 2 for e in audit_events)
|
||||
assert any("Layout TS schimbat" in a.title for a in notifier_alerts)
|
||||
|
||||
|
||||
def test_commit_layout_change_chart_ids_for_n3(monkeypatch):
|
||||
"""For n>=3 charts use 'chart_0', 'chart_1', ..."""
|
||||
import atm.main as _main
|
||||
from atm.config import ROI
|
||||
|
||||
cfg = MagicMock()
|
||||
cfg.lockout_s = 60
|
||||
cfg.colors = {}
|
||||
cfg.debounce_depth = 3
|
||||
state = _main._LoopState()
|
||||
|
||||
class _N:
|
||||
def send(self, a): pass
|
||||
class _A:
|
||||
def log(self, e): pass
|
||||
class _S:
|
||||
is_running = False
|
||||
def stop(self): pass
|
||||
|
||||
ctx = _main.RunContext(
|
||||
cfg=cfg, capture=lambda: None, canary=MagicMock(),
|
||||
detector=MagicMock(), fsm=MagicMock(),
|
||||
notifier=_N(), audit=_A(), detection_log=_A(),
|
||||
scheduler=_S(), samples_dir=Path("."), fires_dir=Path("."),
|
||||
cmd_queue=MagicMock(), state=state,
|
||||
levels_extractor_factory=lambda *a, **kw: None,
|
||||
)
|
||||
ctx.charts = []
|
||||
|
||||
strips = [ROI(x=i * 100, y=0, w=80, h=35) for i in range(3)]
|
||||
_main._commit_layout_change(ctx, strips, now=100.0)
|
||||
assert [c.chart_id for c in ctx.charts] == ["chart_0", "chart_1", "chart_2"]
|
||||
|
||||
|
||||
def test_strips_match_silent_update_jitter():
|
||||
"""_strips_match returns True for <=10px jitter — used to trigger silent update."""
|
||||
from atm.layout import _strips_match
|
||||
from atm.config import ROI
|
||||
a = [ROI(x=0, y=0, w=200, h=35), ROI(x=250, y=0, w=200, h=35)]
|
||||
b = [ROI(x=3, y=0, w=200, h=35), ROI(x=253, y=0, w=200, h=35)]
|
||||
assert _strips_match(a, b, tol=10) is True
|
||||
# Drift 12px breaks the match
|
||||
c = [ROI(x=12, y=0, w=200, h=35), ROI(x=250, y=0, w=200, h=35)]
|
||||
assert _strips_match(a, c, tol=10) is False
|
||||
|
||||
|
||||
def test_build_heartbeat_alert_single_compat():
|
||||
"""Without charts arg, _build_heartbeat_alert returns the legacy single-chart body."""
|
||||
import atm.main as _main
|
||||
a = _main._build_heartbeat_alert(
|
||||
fsm_state="IDLE", fire_count=0, uptime_h=1.0, canary_paused=False,
|
||||
)
|
||||
assert a.kind == "heartbeat"
|
||||
assert a.title == "activ"
|
||||
assert a.body == "IDLE | semnale: 0 | 1.0h"
|
||||
|
||||
|
||||
def test_build_heartbeat_alert_multi_chart():
|
||||
"""charts=[left,right] body lines reference [stânga] / [dreapta]."""
|
||||
import atm.main as _main
|
||||
from atm.config import ROI
|
||||
|
||||
fsm_left = types.SimpleNamespace(state=types.SimpleNamespace(value="ARMED_BUY"))
|
||||
fsm_right = types.SimpleNamespace(state=types.SimpleNamespace(value="IDLE"))
|
||||
cs_left = _main.ChartState(
|
||||
chart_id="left", sub_roi=ROI(x=0, y=0, w=10, h=10),
|
||||
detector=MagicMock(), fsm=fsm_left,
|
||||
)
|
||||
cs_right = _main.ChartState(
|
||||
chart_id="right", sub_roi=ROI(x=0, y=0, w=10, h=10),
|
||||
detector=MagicMock(), fsm=fsm_right,
|
||||
)
|
||||
a = _main._build_heartbeat_alert(
|
||||
fsm_state="ignored", fire_count=2, uptime_h=1.5, canary_paused=False,
|
||||
charts=[cs_left, cs_right],
|
||||
)
|
||||
assert a.kind == "heartbeat"
|
||||
assert "[stânga]" in a.body
|
||||
assert "[dreapta]" in a.body
|
||||
assert "ARMED_BUY" in a.body
|
||||
assert "IDLE" in a.body
|
||||
assert "semnale: 2" in a.body
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_multi_tick_silent_jitter_update(monkeypatch, tmp_path):
|
||||
"""When detect_strips returns positions with <=10px jitter, sub_roi updates
|
||||
silently (no _commit_layout_change → no extra status alert)."""
|
||||
import atm.main as _main
|
||||
from atm.config import ROI
|
||||
|
||||
monkeypatch.chdir(tmp_path)
|
||||
|
||||
initial = [ROI(x=0, y=0, w=200, h=35), ROI(x=250, y=0, w=200, h=35)]
|
||||
jittered = [ROI(x=3, y=0, w=200, h=35), ROI(x=253, y=0, w=200, h=35)]
|
||||
|
||||
class _Det:
|
||||
def __init__(self): self.last_roi = None
|
||||
def step(self, ts, frame=None):
|
||||
from atm.detector import DetectionResult
|
||||
return DetectionResult(
|
||||
ts=ts, window_found=True, dot_found=False,
|
||||
rgb=None, match=None, accepted=False, color=None,
|
||||
)
|
||||
def update_dot_roi(self, roi):
|
||||
self.last_roi = roi
|
||||
|
||||
class _FSM:
|
||||
state = types.SimpleNamespace(value="IDLE")
|
||||
_last_fire: dict = {}
|
||||
|
||||
cfg = MagicMock()
|
||||
cfg.lockout_s = 60
|
||||
cfg.attach_screenshots = types.SimpleNamespace(arm=False, prime=False, trigger=False, late_start=False, catchup=False, opposite_rearm=False, rearm=False, phase_skip=False)
|
||||
cfg.window_title = None
|
||||
state_obj = _main._LoopState()
|
||||
|
||||
notifier_alerts: list = []
|
||||
audit_events: list = []
|
||||
|
||||
class _N:
|
||||
def send(self, a): notifier_alerts.append(a)
|
||||
class _A:
|
||||
def log(self, e): audit_events.append(e)
|
||||
class _S:
|
||||
is_running = False
|
||||
def start(self, s): pass
|
||||
def stop(self): pass
|
||||
|
||||
canary = types.SimpleNamespace(is_paused=False, check=lambda f: types.SimpleNamespace(distance=0, drifted=False))
|
||||
|
||||
ctx = _main.RunContext(
|
||||
cfg=cfg, capture=lambda: np.zeros((200, 600, 3), dtype=np.uint8),
|
||||
canary=canary, detector=MagicMock(), fsm=_FSM(),
|
||||
notifier=_N(), audit=_A(), detection_log=_A(),
|
||||
scheduler=_S(), samples_dir=tmp_path, fires_dir=tmp_path,
|
||||
cmd_queue=MagicMock(), state=state_obj,
|
||||
levels_extractor_factory=lambda *a, **kw: None,
|
||||
lifecycle=None,
|
||||
)
|
||||
det_l, det_r = _Det(), _Det()
|
||||
ctx.charts = [
|
||||
_main.ChartState("left", initial[0], det_l, _FSM()),
|
||||
_main.ChartState("right", initial[1], det_r, _FSM()),
|
||||
]
|
||||
|
||||
# Stub strip detection to return jittered strips.
|
||||
monkeypatch.setattr(_main, "_detect_strips_for_ctx", lambda c, f: jittered)
|
||||
|
||||
results = await _main._run_multi_tick(ctx)
|
||||
assert len(results) == 2
|
||||
# Sub-ROI silently updated, but no layout_change event.
|
||||
assert ctx.charts[0].sub_roi == jittered[0]
|
||||
assert ctx.charts[1].sub_roi == jittered[1]
|
||||
assert det_l.last_roi == jittered[0]
|
||||
assert det_r.last_roi == jittered[1]
|
||||
assert not any(e.get("event") == "layout_change" for e in audit_events)
|
||||
# No status "Layout TS schimbat" alert.
|
||||
assert not any("Layout TS schimbat" in a.title for a in notifier_alerts)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_multi_tick_silent_width_oscillation(monkeypatch, tmp_path):
|
||||
"""Regression: width pulsations (e.g. 792↔810px) on a stable chart count
|
||||
must NOT trigger _commit_layout_change. Reproduces the production bug
|
||||
visible in logs/2026-05-04.jsonl (576 layout_change events/day).
|
||||
"""
|
||||
import atm.main as _main
|
||||
from atm.config import ROI
|
||||
|
||||
monkeypatch.chdir(tmp_path)
|
||||
|
||||
initial = [ROI(x=0, y=0, w=792, h=35), ROI(x=912, y=0, w=845, h=35)]
|
||||
pulsed = [ROI(x=0, y=0, w=810, h=35), ROI(x=912, y=0, w=863, h=35)]
|
||||
|
||||
class _Det:
|
||||
def __init__(self): self.last_roi = None
|
||||
def step(self, ts, frame=None):
|
||||
from atm.detector import DetectionResult
|
||||
return DetectionResult(
|
||||
ts=ts, window_found=True, dot_found=False,
|
||||
rgb=None, match=None, accepted=False, color=None,
|
||||
)
|
||||
def update_dot_roi(self, roi):
|
||||
self.last_roi = roi
|
||||
|
||||
class _FSM:
|
||||
state = types.SimpleNamespace(value="PRIMED_BUY")
|
||||
_last_fire: dict = {}
|
||||
|
||||
cfg = MagicMock()
|
||||
cfg.lockout_s = 60
|
||||
cfg.attach_screenshots = types.SimpleNamespace(arm=False, prime=False, trigger=False, late_start=False, catchup=False, opposite_rearm=False, rearm=False, phase_skip=False)
|
||||
cfg.window_title = None
|
||||
state_obj = _main._LoopState()
|
||||
|
||||
notifier_alerts: list = []
|
||||
audit_events: list = []
|
||||
|
||||
class _N:
|
||||
def send(self, a): notifier_alerts.append(a)
|
||||
class _A:
|
||||
def log(self, e): audit_events.append(e)
|
||||
class _S:
|
||||
is_running = False
|
||||
def start(self, s): pass
|
||||
def stop(self): pass
|
||||
|
||||
canary = types.SimpleNamespace(is_paused=False, check=lambda f: types.SimpleNamespace(distance=0, drifted=False))
|
||||
|
||||
fsm_left = _FSM()
|
||||
fsm_right = _FSM()
|
||||
ctx = _main.RunContext(
|
||||
cfg=cfg, capture=lambda: np.zeros((200, 1800, 3), dtype=np.uint8),
|
||||
canary=canary, detector=MagicMock(), fsm=fsm_left,
|
||||
notifier=_N(), audit=_A(), detection_log=_A(),
|
||||
scheduler=_S(), samples_dir=tmp_path, fires_dir=tmp_path,
|
||||
cmd_queue=MagicMock(), state=state_obj,
|
||||
levels_extractor_factory=lambda *a, **kw: None,
|
||||
lifecycle=None,
|
||||
)
|
||||
det_l, det_r = _Det(), _Det()
|
||||
ctx.charts = [
|
||||
_main.ChartState("left", initial[0], det_l, fsm_left),
|
||||
_main.ChartState("right", initial[1], det_r, fsm_right),
|
||||
]
|
||||
|
||||
monkeypatch.setattr(_main, "_detect_strips_for_ctx", lambda c, f: pulsed)
|
||||
|
||||
await _main._run_multi_tick(ctx)
|
||||
|
||||
# Sub-ROI silently updated.
|
||||
assert ctx.charts[0].sub_roi == pulsed[0]
|
||||
assert ctx.charts[1].sub_roi == pulsed[1]
|
||||
assert det_l.last_roi == pulsed[0]
|
||||
assert det_r.last_roi == pulsed[1]
|
||||
# FSM identity preserved (not rebuilt).
|
||||
assert ctx.charts[0].fsm is fsm_left
|
||||
assert ctx.charts[1].fsm is fsm_right
|
||||
# Critical: no layout_change event, no status alert.
|
||||
assert not any(e.get("event") == "layout_change" for e in audit_events)
|
||||
assert not any("Layout TS schimbat" in a.title for a in notifier_alerts)
|
||||
|
||||
|
||||
def test_commit_layout_change_alert_is_silent(monkeypatch):
|
||||
"""Layout-change alert on real count change uses silent=True so Telegram
|
||||
doesn't ping the user. Re-anchor info still visible in chat history.
|
||||
"""
|
||||
import atm.main as _main
|
||||
from atm.config import ROI
|
||||
from atm.state_machine import StateMachine
|
||||
|
||||
cfg = MagicMock()
|
||||
cfg.lockout_s = 60
|
||||
cfg.colors = {}
|
||||
cfg.debounce_depth = 3
|
||||
cfg.confidence_min = 0.5
|
||||
cfg.dot_roi = ROI(x=0, y=0, w=600, h=35)
|
||||
|
||||
notifier_alerts: list = []
|
||||
audit_events: list = []
|
||||
|
||||
class _N:
|
||||
def send(self, a): notifier_alerts.append(a)
|
||||
class _A:
|
||||
def log(self, e): audit_events.append(e)
|
||||
class _S:
|
||||
is_running = False
|
||||
def stop(self): pass
|
||||
|
||||
state_obj = _main._LoopState()
|
||||
ctx = _main.RunContext(
|
||||
cfg=cfg, capture=lambda: None, canary=MagicMock(),
|
||||
detector=MagicMock(), fsm=StateMachine(60),
|
||||
notifier=_N(), audit=_A(), detection_log=_A(),
|
||||
scheduler=_S(), samples_dir=Path("."), fires_dir=Path("."),
|
||||
cmd_queue=MagicMock(), state=state_obj,
|
||||
levels_extractor_factory=lambda *a, **kw: None,
|
||||
lifecycle=None,
|
||||
)
|
||||
ctx.charts = [
|
||||
_main.ChartState("only", ROI(x=0, y=0, w=400, h=35), MagicMock(), StateMachine(60)),
|
||||
]
|
||||
|
||||
new_strips = [
|
||||
ROI(x=0, y=0, w=200, h=35),
|
||||
ROI(x=250, y=0, w=200, h=35),
|
||||
]
|
||||
_main._commit_layout_change(ctx, new_strips, now=100.0)
|
||||
|
||||
layout_alerts = [a for a in notifier_alerts if "Layout TS schimbat" in a.title]
|
||||
assert len(layout_alerts) == 1
|
||||
assert layout_alerts[0].silent is True
|
||||
|
||||
Reference in New Issue
Block a user