From 19fe87573d209e879926daade71f913b0eebb67d Mon Sep 17 00:00:00 2001 From: David Gwilliam Date: Sat, 21 Mar 2026 19:26:53 -0700 Subject: [PATCH] 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. --- tests/test_graph_pipeline.py | 260 +++++++++++++++++++++++++++++++++++ 1 file changed, 260 insertions(+) create mode 100644 tests/test_graph_pipeline.py diff --git a/tests/test_graph_pipeline.py b/tests/test_graph_pipeline.py new file mode 100644 index 0000000..00b78a7 --- /dev/null +++ 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__])