import asyncio import pytest from app.analysis.alerts import Alert from app.analysis.confluence import Cluster, cluster_levels from app.analysis.levels import Level, LevelKind, Side from app.analysis.manual_lines import ManualLine from app.bars.models import Timeframe from app.config import Settings from app.runtime import Runtime def runtime(tmp_path, **overrides) -> Runtime: settings = Settings( manual_lines_path=tmp_path / "manual_lines.json", # Isolated per test: the default is relative to the working directory, # so without this every test shares one alert-suppression file and they # silence each other. alert_state_path=tmp_path / "alert_state.json", ntfy_topic=overrides.pop("ntfy_topic", ""), **overrides, ) return Runtime(settings) def alert() -> Alert: level = Level( "pd:high", LevelKind.HORIZONTAL, Timeframe.D1, Side.RESISTANCE, 16, 1, "PDH", 100, 5000, 0, None, 0, 100, 100, False, False, ) cluster = Cluster("cl_x", Side.RESISTANCE, 5000, 5000, 5000, 16, [level], 1.0) return Alert(cluster, "BEARISH ZONE /ES") def test_one_engine_serves_every_connection(tmp_path): # Previously each WebSocket built its own engine, so reloading the page # cleared the cooldown and the same zone alerted again immediately. instance = runtime(tmp_path) assert instance.alert_engine is not None assert instance.alert_engine.min_score == instance.settings.confluence_min_score def test_alerts_reach_subscribers(tmp_path): instance = runtime(tmp_path) queue: asyncio.Queue = asyncio.Queue(maxsize=10) instance.subscribers.add(queue) async def scenario(): instance.dispatch_alerts([alert()]) return queue.get_nowait() event = asyncio.run(scenario()) assert event["type"] == "alert" assert event["message"] == "BEARISH ZONE /ES" def test_ntfy_failure_does_not_propagate(tmp_path, monkeypatch): # send_ntfy used to be awaited inside the WebSocket loop, whose except # clause only caught disconnects — so a push outage killed the connection. instance = runtime(tmp_path, ntfy_topic="chart-test") async def explode(*args, **kwargs): raise RuntimeError("ntfy is down") monkeypatch.setattr("app.runtime.send_ntfy", explode) asyncio.run(instance.notify("anything")) def test_blank_topic_sends_nothing(tmp_path, monkeypatch): instance = runtime(tmp_path) calls = [] async def record(server, topic, message): calls.append(topic) monkeypatch.setattr("app.runtime.send_ntfy", record) asyncio.run(instance.notify("anything")) assert calls == [""] # send_ntfy itself is the one that short-circuits def test_tick_bars_update_higher_timeframes_without_doubling_volume(tmp_path): # The aggregator accumulates with `current.v += incoming.v`, so a forming # minute re-sent on every tick would add its volume to each higher # timeframe again and again. The provisional path must combine, not # accumulate: five ticks in one minute leave the hour's volume equal to the # closed minutes plus the live one, exactly once. from app.bars.models import Bar instance = runtime(tmp_path) def minute(t, close, volume, closed=True): return Bar(tf=Timeframe.M1, t=t, o=close, h=close, l=close, c=close, v=volume, closed=closed, symbol="/ES", source="test") base = 1786356000 # top of an hour asyncio.run(instance.on_bar(minute(base, 100.0, 10))) asyncio.run(instance.on_bar(minute(base + 60, 101.0, 20))) settled = [b for b in instance.store.get(Timeframe.H1) if b.t == base][-1].v assert settled == 30 for _ in range(5): asyncio.run(instance.on_bar(minute(base + 120, 102.0, 7, closed=False))) hour = [b for b in instance.store.get(Timeframe.H1) if b.t == base][-1] assert hour.v == 37, "the live minute's volume must be added once, not per tick" assert hour.c == 102.0 assert hour.closed is False def test_a_tripped_manual_alert_stays_disarmed_after_rebuild_and_restart(tmp_path): instance = runtime(tmp_path) instance.manual_lines.add( ManualLine( "ml_once", Timeframe.D1, Side.RESISTANCE, 1000, 5000, 0.0, 1000, 1000, note="one shot", ) ) instance.rebuild_levels() clusters = cluster_levels(instance.levels, 1000, 5000, 1) alerts = instance.alert_engine.evaluate(clusters, 5000, 1, 1000, "/ES") async def dispatch(): instance.dispatch_alerts(alerts) await asyncio.gather(*instance._notify_tasks) asyncio.run(dispatch()) assert len(alerts) == 1 assert alerts[0].tripped == ("ml_once",) assert instance.manual_lines.lines["ml_once"].armed is False assert next(level for level in instance.levels if level.id == "ml_once").armed is False restarted = runtime(tmp_path) assert restarted.manual_lines.lines["ml_once"].armed is False assert next(level for level in restarted.levels if level.id == "ml_once").armed is False