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", user_prefs_path=tmp_path / "user_prefs.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 def test_a_broadcast_from_a_worker_thread_reaches_subscribers(tmp_path): # The mutating routes are sync `def`, so FastAPI runs them in a threadpool, # and they reach broadcast through rebuild_levels. asyncio.Queue is not # thread-safe — it wakes a consumer by resolving a Future, which only the # loop thread may do — so writing it from there can drop the wakeup and # leave one browser's drawing invisible to another until the next tick. instance = runtime(tmp_path) async def exercise(): instance._loop = asyncio.get_running_loop() queue: asyncio.Queue = asyncio.Queue(maxsize=10) instance.subscribers.add(queue) waiting = asyncio.ensure_future(queue.get()) await asyncio.sleep(0) # park the consumer on the Future await asyncio.to_thread(instance.broadcast, {"type": "bar", "bar": "sentinel"}) return await asyncio.wait_for(waiting, timeout=2) assert asyncio.run(exercise())["bar"] == "sentinel" def test_a_cross_thread_broadcast_is_posted_through_the_loop(tmp_path): """The contract, asserted directly. The race itself is timing-dependent and usually masked — a foreign-thread put_nowait often lands in the loop's ready queue before it sleeps, and the stream ticking once a second papers over the times it does not. So this asserts the rule rather than trying to provoke the failure: a broadcast from off the loop must go through call_soon_threadsafe, never touch the queue. """ instance = runtime(tmp_path) async def exercise(): loop = asyncio.get_running_loop() instance._loop = loop posted = [] original = loop.call_soon_threadsafe def spy(callback, *args): posted.append(callback) return original(callback, *args) loop.call_soon_threadsafe = spy try: await asyncio.to_thread(instance.broadcast, {"type": "bar", "bar": "x"}) finally: loop.call_soon_threadsafe = original return posted assert asyncio.run(exercise()), "a worker-thread broadcast bypassed the loop" def test_broadcasting_on_the_loop_still_delivers_synchronously(tmp_path): # The stream's own path must not pay for a hop it does not need. instance = runtime(tmp_path) async def exercise(): instance._loop = asyncio.get_running_loop() queue: asyncio.Queue = asyncio.Queue(maxsize=10) instance.subscribers.add(queue) instance.broadcast({"type": "bar", "bar": "direct"}) return queue.get_nowait() # already there, no await needed assert asyncio.run(exercise())["bar"] == "direct"