test(graph): Add comprehensive test suite for graph system

Add 17 tests covering all aspects of the graph-based pipeline system: - Graph creation and manipulation (7 tests) - Empty graph creation - Node addition with various formats - Connection handling with validation - Chain connection helper - Graph validation (3 tests) - Disconnected node detection - Cycle detection using DFS - Clean graph validation - Serialization/deserialization (2 tests) - to_dict() for basic graphs - from_dict() for loading from dictionaries - Pipeline conversion (5 tests) - Minimal pipeline conversion - Effect nodes with intensity - Positioning nodes - Camera nodes - Simple graph execution All tests pass successfully and verify the graph system works correctly with the existing pipeline architecture.
2026-03-21 19:26:53 -07:00
parent 1a7da400e3
commit 19fe87573d
1 changed files with 260 additions and 0 deletions
--- a/tests/test_graph_pipeline.py
+++ b/tests/test_graph_pipeline.py
@@ -0,0 +1,260 @@
+"""
+Tests for the graph-based pipeline configuration.
+"""
+
+import pytest
+
+from engine.effects.plugins import discover_plugins
+from engine.pipeline.graph import Graph, NodeType, Node
+from engine.pipeline.graph_adapter import dict_to_pipeline, graph_to_pipeline
+
+
+@pytest.fixture(autouse=True)
+def setup_effects():
+    """Ensure effects are discovered before each test."""
+    discover_plugins()
+
+
+class TestGraphCreation:
+    """Tests for Graph creation and manipulation."""
+
+    def test_create_empty_graph(self):
+        """Graph can be created empty."""
+        graph = Graph()
+        assert len(graph.nodes) == 0
+        assert len(graph.connections) == 0
+
+    def test_add_node(self):
+        """Graph.node adds a node."""
+        graph = Graph()
+        graph.node("source", NodeType.SOURCE, source="headlines")
+
+        assert "source" in graph.nodes
+        node = graph.nodes["source"]
+        assert node.name == "source"
+        assert node.type == NodeType.SOURCE
+        assert node.config["source"] == "headlines"
+
+    def test_add_node_string_type(self):
+        """Graph.node accepts string type."""
+        graph = Graph()
+        graph.node("camera", "camera", mode="scroll")
+
+        assert "camera" in graph.nodes
+        assert graph.nodes["camera"].type == NodeType.CAMERA
+
+    def test_connect_nodes(self):
+        """Graph.connect adds connection between nodes."""
+        graph = Graph()
+        graph.node("source", NodeType.SOURCE)
+        graph.node("display", NodeType.DISPLAY)
+        graph.connect("source", "display")
+
+        assert len(graph.connections) == 1
+        conn = graph.connections[0]
+        assert conn.source == "source"
+        assert conn.target == "display"
+
+    def test_connect_nonexistent_source(self):
+        """Graph.connect raises error for nonexistent source."""
+        graph = Graph()
+        graph.node("display", NodeType.DISPLAY)
+
+        with pytest.raises(ValueError, match="Source node 'missing' not found"):
+            graph.connect("missing", "display")
+
+    def test_connect_nonexistent_target(self):
+        """Graph.connect raises error for nonexistent target."""
+        graph = Graph()
+        graph.node("source", NodeType.SOURCE)
+
+        with pytest.raises(ValueError, match="Target node 'missing' not found"):
+            graph.connect("source", "missing")
+
+    def test_chain_connects_nodes(self):
+        """Graph.chain connects nodes in sequence."""
+        graph = Graph()
+        graph.node("source", NodeType.SOURCE)
+        graph.node("camera", NodeType.CAMERA)
+        graph.node("display", NodeType.DISPLAY)
+        graph.chain("source", "camera", "display")
+
+        assert len(graph.connections) == 2
+        assert graph.connections[0].source == "source"
+        assert graph.connections[0].target == "camera"
+        assert graph.connections[1].source == "camera"
+        assert graph.connections[1].target == "display"
+
+
+class TestGraphValidation:
+    """Tests for Graph validation."""
+
+    def test_validate_disconnected_node(self):
+        """Validation detects disconnected nodes."""
+        graph = Graph()
+        graph.node("source", NodeType.SOURCE)
+        graph.node("orphan", NodeType.EFFECT, effect="noise")
+
+        errors = graph.validate()
+        # Both source and orphan are disconnected
+        assert len(errors) == 2
+        assert any("orphan" in e and "not connected" in e for e in errors)
+        assert any("source" in e and "not connected" in e for e in errors)
+
+    def test_validate_cycle_detection(self):
+        """Validation detects cycles."""
+        graph = Graph()
+        graph.node("a", NodeType.SOURCE)
+        graph.node("b", NodeType.CAMERA)
+        graph.node("c", NodeType.DISPLAY)
+        graph.connect("a", "b")
+        graph.connect("b", "c")
+        graph.connect("c", "a")  # Creates cycle
+
+        errors = graph.validate()
+        assert len(errors) > 0
+        assert any("cycle" in e.lower() for e in errors)
+
+    def test_validate_clean_graph(self):
+        """Validation returns no errors for valid graph."""
+        graph = Graph()
+        graph.node("source", NodeType.SOURCE)
+        graph.node("display", NodeType.DISPLAY)
+        graph.connect("source", "display")
+
+        errors = graph.validate()
+        assert len(errors) == 0
+
+
+class TestGraphToDict:
+    """Tests for Graph serialization."""
+
+    def test_to_dict_basic(self):
+        """Graph.to_dict produces correct structure."""
+        graph = Graph()
+        graph.node("source", NodeType.SOURCE, source="headlines")
+        graph.node("display", NodeType.DISPLAY, backend="terminal")
+        graph.connect("source", "display")
+
+        data = graph.to_dict()
+
+        assert "nodes" in data
+        assert "connections" in data
+        assert "source" in data["nodes"]
+        assert "display" in data["nodes"]
+        assert data["nodes"]["source"]["type"] == "source"
+        assert data["nodes"]["display"]["type"] == "display"
+
+    def test_from_dict_simple(self):
+        """Graph.from_dict loads simple format."""
+        data = {
+            "nodes": {
+                "source": "headlines",
+                "display": {"type": "display", "backend": "terminal"},
+            },
+            "connections": ["source -> display"],
+        }
+
+        graph = Graph().from_dict(data)
+
+        assert "source" in graph.nodes
+        assert "display" in graph.nodes
+        assert len(graph.connections) == 1
+
+
+class TestDictToPipeline:
+    """Tests for dict_to_pipeline conversion."""
+
+    def test_convert_minimal_pipeline(self):
+        """dict_to_pipeline creates a working pipeline."""
+        data = {
+            "nodes": {
+                "source": "headlines",
+                "display": {"type": "display", "backend": "null"},
+            },
+            "connections": ["source -> display"],
+        }
+
+        pipeline = dict_to_pipeline(data, viewport_width=80, viewport_height=24)
+
+        assert pipeline is not None
+        assert "source" in pipeline._stages
+        assert "display" in pipeline._stages
+
+    def test_convert_with_effect(self):
+        """dict_to_pipeline handles effect nodes."""
+        data = {
+            "nodes": {
+                "source": "headlines",
+                "noise": {"type": "effect", "effect": "noise", "intensity": 0.5},
+                "display": {"type": "display", "backend": "null"},
+            },
+            "connections": ["source -> noise -> display"],
+        }
+
+        pipeline = dict_to_pipeline(data)
+
+        assert "noise" in pipeline._stages
+        # Check that intensity was set (this is a global state check)
+        from engine.effects import get_registry
+
+        noise_effect = get_registry().get("noise")
+        assert noise_effect.config.intensity == 0.5
+
+    def test_convert_with_positioning(self):
+        """dict_to_pipeline handles positioning nodes."""
+        data = {
+            "nodes": {
+                "source": "headlines",
+                "position": {"type": "position", "mode": "absolute"},
+                "display": {"type": "display", "backend": "null"},
+            },
+            "connections": ["source -> position -> display"],
+        }
+
+        pipeline = dict_to_pipeline(data)
+
+        assert "position" in pipeline._stages
+        pos_stage = pipeline._stages["position"]
+        assert pos_stage.mode.value == "absolute"
+
+
+class TestGraphToPipeline:
+    """Tests for graph_to_pipeline conversion."""
+
+    def test_convert_simple_graph(self):
+        """graph_to_pipeline converts a simple graph."""
+        graph = Graph()
+        graph.node("source", NodeType.SOURCE, source="headlines")
+        graph.node("display", NodeType.DISPLAY, backend="null")
+        graph.connect("source", "display")
+
+        pipeline = graph_to_pipeline(graph)
+
+        assert pipeline is not None
+        # Pipeline auto-injects missing capabilities (camera, render, etc.)
+        # So we have more stages than just source and display
+        assert "source" in pipeline._stages
+        assert "display" in pipeline._stages
+        # Auto-injected stages include camera, camera_update, render
+        assert "camera" in pipeline._stages
+        assert "camera_update" in pipeline._stages
+        assert "render" in pipeline._stages
+
+    def test_convert_with_camera(self):
+        """graph_to_pipeline handles camera nodes."""
+        graph = Graph()
+        graph.node("source", NodeType.SOURCE, source="headlines")
+        graph.node("camera", NodeType.CAMERA, mode="scroll")
+        graph.node("display", NodeType.DISPLAY, backend="null")
+        graph.chain("source", "camera", "display")
+
+        pipeline = graph_to_pipeline(graph)
+
+        assert "camera" in pipeline._stages
+        camera_stage = pipeline._stages["camera"]
+        assert hasattr(camera_stage, "_camera")
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])