phase 5

tests/detect/test_graph.py

@@ -0,0 +1,79 @@
+"""Tests for the LangGraph detection pipeline."""
+
+import pytest
+
+from detect.graph import NODES, build_graph, get_pipeline
+from detect.models import PipelineStats
+from detect.state import DetectState
+
+VIDEO = "media/out/chunks/95043d50-4df6-4ac8-bbd5-2ba873117c6e/chunk_0000.mp4"
+
+
+def test_graph_compiles():
+    pipeline = get_pipeline()
+    assert pipeline is not None
+
+
+def test_graph_has_all_nodes():
+    graph = build_graph()
+    for node in NODES:
+        assert node in graph.nodes
+
+
+def test_graph_runs_end_to_end(monkeypatch):
+    """Run the full graph with mocked event emission."""
+    events = []
+    monkeypatch.setattr("detect.emit.push_detect_event",
+                        lambda job_id, etype, data: events.append((etype, data)))
+
+    pipeline = get_pipeline()
+    initial_state = DetectState(
+        video_path=VIDEO,
+        job_id="test-graph",
+        profile_name="soccer_broadcast",
+    )
+
+    result = pipeline.invoke(initial_state)
+
+    # All nodes should have transitioned
+    graph_events = [e for e in events if e[0] == "graph_update"]
+    assert len(graph_events) > 0
+
+    # Should have frames
+    assert len(result["frames"]) > 0
+    assert len(result["filtered_frames"]) > 0
+
+    # Report should exist
+    assert result["report"] is not None
+    assert result["report"].content_type == "soccer_broadcast"
+
+    # job_complete should have been emitted
+    complete_events = [e for e in events if e[0] == "job_complete"]
+    assert len(complete_events) == 1
+
+
+def test_graph_node_transitions(monkeypatch):
+    """Verify each node emits running → done transitions."""
+    events = []
+    monkeypatch.setattr("detect.emit.push_detect_event",
+                        lambda job_id, etype, data: events.append((etype, data)))
+
+    pipeline = get_pipeline()
+    initial_state = DetectState(
+        video_path=VIDEO,
+        job_id="test-transitions",
+        profile_name="soccer_broadcast",
+    )
+
+    pipeline.invoke(initial_state)
+
+    graph_events = [e[1] for e in events if e[0] == "graph_update"]
+
+    # Each node should appear as "running" then "done"
+    for node_name in NODES:
+        running = [e for e in graph_events
+                   if any(n["id"] == node_name and n["status"] == "running" for n in e["nodes"])]
+        done = [e for e in graph_events
+                if any(n["id"] == node_name and n["status"] == "done" for n in e["nodes"])]
+        assert len(running) >= 1, f"{node_name} never entered 'running'"
+        assert len(done) >= 1, f"{node_name} never reached 'done'"