diff --git a/cortex_viz/server/graph_event_stream.py b/cortex_viz/server/graph_event_stream.py index 645269c..fb2479e 100644 --- a/cortex_viz/server/graph_event_stream.py +++ b/cortex_viz/server/graph_event_stream.py @@ -185,12 +185,17 @@ def subscribe( # ── Process-wide singleton ────────────────────────────────────────── -# One stream per process. Callers reach it through get_stream() and use the -# GraphEventStream API directly (``activity_stream`` is the worked example); -# there are deliberately no module-level emit/close/reset forwarders. Three -# such wrappers existed and had never had a caller in this repository's -# history, and once ``emit`` gained ``event_meta`` the wrapper silently -# dropped it — a second, subtly weaker door onto the same stream. +# One stream per process. get_stream() returns it for callers that want the +# GraphEventStream API directly (``activity_stream`` is the worked example). +# graph_build_run.py and graph_build_merge.py instead import this module +# ONCE (as ``_events`` / ``events``) and call these module-level forwarders +# for the lifetime of a build — that is the real production binding, not a +# convenience wrapper, so each forwarder mirrors its GraphEventStream method +# parameter-for-parameter. (Issue #134: a prior revision deleted these three +# forwarders instead of fixing ``emit``'s forwarder, which had drifted out of +# sync with ``GraphEventStream.emit`` and silently dropped ``event_meta``; +# that took down every caller. Restored functional — the fix for a drifted +# forwarder is to fix the forwarder.) _stream = GraphEventStream() @@ -199,6 +204,28 @@ def get_stream() -> GraphEventStream: return _stream +def emit( + label: str, + nodes: list[dict[str, Any]], + edges: list[dict[str, Any]], + *, + chunk: int = 1000, + event_meta: dict[str, Any] | None = None, +) -> int: + """Forward to the singleton's ``emit`` — see ``GraphEventStream.emit``.""" + return _stream.emit(label, nodes, edges, chunk=chunk, event_meta=event_meta) + + +def close() -> None: + """Forward to the singleton's ``close`` — see ``GraphEventStream.close``.""" + _stream.close() + + +def reset() -> None: + """Forward to the singleton's ``reset`` — see ``GraphEventStream.reset``.""" + _stream.reset() + + # ── SSE wire helpers ──────────────────────────────────────────────── diff --git a/tests/test_graph_event_stream_forwarders.py b/tests/test_graph_event_stream_forwarders.py new file mode 100644 index 0000000..e604d90 --- /dev/null +++ b/tests/test_graph_event_stream_forwarders.py @@ -0,0 +1,114 @@ +"""Contract for the module-level ``emit``/``close``/``reset`` forwarders. + +Issue #134: a prior revision deleted these three forwarders on the premise +that they had no caller in this repository's history. They had four — +``graph_build_run.py`` and ``graph_build_merge.py`` import this module ONCE +(as ``_events`` / ``events``) and call the module-level functions for the +lifetime of a build. Every existing merge test (``test_graph_build_coverage_ +contracts.py``) substitutes a ``SimpleNamespace(emit=MagicMock())`` for that +import, which is exactly why the deletion went unnoticed: the defect was in +how the caller *obtains* the stream, not in what it does with it. The last +test below exercises the real binding instead of a fake, so a call site and +this module cannot drift apart the same way again. +""" + +from __future__ import annotations + +from cortex_viz.server import graph_event_stream as ges + + +def _fresh_singleton(): + """Leave the process-wide singleton clean for the next test.""" + ges.reset() + + +class _FakeStream: + """Stands in for the process singleton. ``GraphEventStream`` declares + ``__slots__`` (no ``__dict__``), so its methods cannot be monkeypatched + on a live instance — swap the module-level ``_stream`` binding instead, + which is exactly what each forwarder reads through ``get_stream()``.""" + + def __init__(self): + self.emit_calls = [] + self.close_calls = 0 + self.reset_calls = 0 + + def emit(self, *args, **kwargs): + self.emit_calls.append((args, kwargs)) + return 3 + + def close(self): + self.close_calls += 1 + + def reset(self): + self.reset_calls += 1 + + +def test_emit_forwarder_delegates_to_the_process_singleton(monkeypatch): + fake = _FakeStream() + monkeypatch.setattr(ges, "_stream", fake) + result = ges.emit( + "label", [{"id": "n1"}], [{"id": "e1"}], chunk=7, event_meta={"k": 1} + ) + assert result == 3 + assert fake.emit_calls == [ + ( + ("label", [{"id": "n1"}], [{"id": "e1"}]), + {"chunk": 7, "event_meta": {"k": 1}}, + ) + ] + assert ges.get_stream() is fake + + +def test_close_forwarder_delegates_to_the_process_singleton(monkeypatch): + fake = _FakeStream() + monkeypatch.setattr(ges, "_stream", fake) + ges.close() + assert fake.close_calls == 1 + + +def test_reset_forwarder_delegates_to_the_process_singleton(monkeypatch): + fake = _FakeStream() + monkeypatch.setattr(ges, "_stream", fake) + ges.reset() + assert fake.reset_calls == 1 + + +def test_emit_forwarder_carries_event_meta_through_to_the_subscriber(): + """The exact drift that started the incident: `emit` grew `event_meta` + and the forwarder silently dropped it. This must fail against the + pre-#134-fix forwarder — which had no `event_meta` parameter at all — + and pass against the restored one.""" + _fresh_singleton() + try: + ges.emit("activity", [{"id": "n1"}], [], event_meta={"activity_id": 42}) + + (_idx, event) = next(iter(ges.get_stream().subscribe(since=0, timeout=0.01))) + assert event["activity_id"] == 42 + assert event["label"] == "activity" + finally: + _fresh_singleton() + + +def test_production_build_binding_reaches_the_real_singleton(): + """Reproduces exactly what ``graph_build_run.py`` does: import this + module once, then drive it through ``reset`` -> ``emit`` -> ``close`` as + module-level calls (never touching ``get_stream()`` directly, the way a + fake-stream test would). If a future change breaks the module/singleton + binding, this is the test that catches it.""" + from cortex_viz.server import graph_event_stream as _events + + _events.reset() + emitted = _events.emit("baseline", [{"id": "n1"}], [], chunk=1000) + assert emitted == 1 + + stream = _events.get_stream() + events = [event for _idx, event in stream.subscribe(since=0, timeout=0.01)] + assert len(events) == 1 + assert events[0]["label"] == "baseline" + + _events.close() + assert stream.stats()["closed"] is True + + _events.reset() + assert stream.stats() == {"count": 0, "closed": False}