feat: add datasource and definition graph contracts

2026-07-28 12:42:49 +02:00
parent d36bb94335
commit 8b1910b5b7
13 changed files with 1297 additions and 4 deletions
--- a/tests/test_definition_graphs.py
+++ b/tests/test_definition_graphs.py
@@ -0,0 +1,145 @@
+from __future__ import annotations
+
+import unittest
+
+from govoplan_core.core.definition_graphs import (
+    DefinitionEdge,
+    DefinitionGraphConstraints,
+    DefinitionGraphLibrary,
+    DefinitionNode,
+    DefinitionNodeCountConstraint,
+    DefinitionNodeType,
+    DefinitionPort,
+    validate_definition_graph,
+)
+
+
+def library(*, allow_cycles: bool) -> DefinitionGraphLibrary:
+    return DefinitionGraphLibrary(
+        id="test",
+        version="0.1.0",
+        category_labels={"control": "Control"},
+        node_types=(
+            DefinitionNodeType(
+                type="start",
+                category="control",
+                label="Start",
+                description="Start",
+                icon="play",
+            ),
+            DefinitionNodeType(
+                type="step",
+                category="control",
+                label="Step",
+                description="Step",
+                icon="square",
+                input_ports=(DefinitionPort(id="input", label="Input"),),
+            ),
+            DefinitionNodeType(
+                type="end",
+                category="control",
+                label="End",
+                description="End",
+                icon="circle-stop",
+                input_ports=(DefinitionPort(id="input", label="Input"),),
+                output_ports=(),
+            ),
+        ),
+        constraints=DefinitionGraphConstraints(
+            allow_cycles=allow_cycles,
+            node_counts=(
+                DefinitionNodeCountConstraint(
+                    code="graph.start_count",
+                    label="start",
+                    minimum=1,
+                    maximum=1,
+                    node_types=("start",),
+                ),
+                DefinitionNodeCountConstraint(
+                    code="graph.end_count",
+                    label="end",
+                    minimum=1,
+                    node_types=("end",),
+                ),
+            ),
+        ),
+    )
+
+
+class DefinitionGraphTests(unittest.TestCase):
+    def test_library_driven_ports_and_counts_validate_a_definition(self) -> None:
+        diagnostics = validate_definition_graph(
+            library(allow_cycles=False),
+            nodes=(
+                DefinitionNode(id="start", type="start"),
+                DefinitionNode(id="step", type="step"),
+                DefinitionNode(id="end", type="end"),
+            ),
+            edges=(
+                DefinitionEdge(id="edge-1", source="start", target="step"),
+                DefinitionEdge(id="edge-2", source="step", target="end"),
+            ),
+        )
+
+        self.assertEqual((), diagnostics)
+
+    def test_constraints_change_cycle_semantics_without_changing_graph_shape(self) -> None:
+        nodes = (
+            DefinitionNode(id="start", type="start"),
+            DefinitionNode(id="step", type="step"),
+            DefinitionNode(id="retry", type="step"),
+            DefinitionNode(id="end", type="end"),
+        )
+        edges = (
+            DefinitionEdge(id="edge-1", source="start", target="step"),
+            DefinitionEdge(id="edge-2", source="step", target="retry"),
+            DefinitionEdge(id="edge-3", source="retry", target="step"),
+            DefinitionEdge(id="edge-4", source="retry", target="end"),
+        )
+
+        dag_codes = {
+            item.code
+            for item in validate_definition_graph(
+                library(allow_cycles=False),
+                nodes=nodes,
+                edges=edges,
+            )
+        }
+        workflow_codes = {
+            item.code
+            for item in validate_definition_graph(
+                library(allow_cycles=True),
+                nodes=nodes,
+                edges=edges,
+            )
+        }
+
+        self.assertIn("graph.cycle", dag_codes)
+        self.assertNotIn("graph.cycle", workflow_codes)
+
+    def test_unknown_ports_and_missing_required_inputs_are_explicit(self) -> None:
+        diagnostics = validate_definition_graph(
+            library(allow_cycles=False),
+            nodes=(
+                DefinitionNode(id="start", type="start"),
+                DefinitionNode(id="end", type="end"),
+            ),
+            edges=(
+                DefinitionEdge(
+                    id="edge",
+                    source="start",
+                    target="end",
+                    target_port="missing",
+                ),
+            ),
+        )
+
+        self.assertTrue(
+            {"edge.unknown_target_port", "node.input_required"}.issubset(
+                {item.code for item in diagnostics}
+            )
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()