from __future__ import annotations import unittest from govoplan_dataflow.backend.executor import execute_preview from govoplan_dataflow.backend.expressions import ( ExpressionError, evaluate_expression, infer_expression_type, parse_expression, ) from govoplan_dataflow.backend.graph import validate_graph from govoplan_dataflow.backend.node_library import NODE_TYPES from govoplan_dataflow.backend.operator_registry import OPERATOR_REGISTRY from govoplan_dataflow.backend.schemas import ( GraphEdge, GraphNode, GraphPosition, PipelineGraph, ) from govoplan_dataflow.backend.sql_compiler import render_sql def node( node_id: str, node_type: str, config: dict, *, x: int, ) -> GraphNode: return GraphNode( id=node_id, type=node_type, label=node_id, position=GraphPosition(x=x, y=100), config=config, ) class DataflowOperatorTests(unittest.TestCase): def test_typed_expression_is_safe_and_deterministic(self) -> None: parsed = parse_expression( "case when amount >= 10 then lower(trim(name)) else 'small' end" ) self.assertEqual( "ada", evaluate_expression(parsed, {"amount": 12, "name": " ADA "}), ) self.assertEqual( "string", infer_expression_type( parsed, {"amount": "integer", "name": "string"}, ), ) with self.assertRaises(ExpressionError): parse_expression("(select secret from credentials)") def test_conversion_expression_and_quality_nodes_execute(self) -> None: graph = PipelineGraph( nodes=[ node( "source", "source.inline", { "source_name": "input", "rows": [ {"id": "1", "amount": "12.50", "name": " Ada "}, {"id": "2", "amount": "bad", "name": ""}, ], }, x=0, ), node( "convert", "convert", { "source_column": "amount", "target_column": "amount_number", "target_type": "number", "on_error": "null", }, x=200, ), node( "expression", "expression", { "target_column": "normalized_name", "expression": "lower(trim(name))", "result_type": "string", }, x=400, ), node( "quality", "quality.rules", { "rules": [ { "id": "amount-required", "column": "amount_number", "operator": "not_null", }, { "id": "name-required", "column": "normalized_name", "operator": "not_null", }, ], "action": "annotate", }, x=600, ), node("output", "output", {}, x=800), ], edges=[ GraphEdge(id="e1", source="source", target="convert"), GraphEdge(id="e2", source="convert", target="expression"), GraphEdge(id="e3", source="expression", target="quality"), GraphEdge(id="e4", source="quality", target="output"), ], ) self.assertFalse( [item for item in validate_graph(graph) if item.severity == "error"] ) result = execute_preview(graph, row_limit=100) self.assertEqual("ada", result.rows[0]["normalized_name"]) self.assertTrue(result.rows[0]["_quality_valid"]) self.assertFalse(result.rows[1]["_quality_valid"]) self.assertEqual( ["amount-required", "name-required"], result.rows[1]["_quality_errors"], ) def test_reconciliation_reports_changed_and_missing_rows(self) -> None: graph = PipelineGraph( nodes=[ node( "expected", "source.inline", { "source_name": "expected", "rows": [ {"id": "1", "amount": 10}, {"id": "2", "amount": 20}, ], }, x=0, ), node( "observed", "source.inline", { "source_name": "observed", "rows": [ {"id": "1", "amount": 11}, {"id": "3", "amount": 30}, ], }, x=0, ), node( "reconcile", "reconcile.compare", { "left_keys": ["id"], "right_keys": ["id"], "compare_columns": ["amount"], "right_prefix": "observed_", }, x=300, ), node("output", "output", {}, x=600), ], edges=[ GraphEdge( id="e1", source="expected", target="reconcile", target_port="left", ), GraphEdge( id="e2", source="observed", target="reconcile", target_port="right", ), GraphEdge(id="e3", source="reconcile", target="output"), ], ) result = execute_preview(graph, row_limit=100) self.assertEqual( ["changed", "missing_observed", "missing_expected"], [item["_reconciliation_status"] for item in result.rows], ) def test_parameterized_subflow_runs_a_pinned_graph(self) -> None: nested = { "schema_version": 1, "nodes": [ { "id": "input", "type": "source.inline", "label": "Input", "position": {"x": 0, "y": 100}, "config": { "source_name": "subflow_input", "rows": [], "input_binding": True, }, }, { "id": "filter", "type": "filter.expression", "label": "Minimum", "position": {"x": 200, "y": 100}, "config": {"expression": "amount >= ${minimum}"}, }, { "id": "output", "type": "output", "label": "Output", "position": {"x": 400, "y": 100}, "config": {}, }, ], "edges": [ {"id": "e1", "source": "input", "target": "filter"}, {"id": "e2", "source": "filter", "target": "output"}, ], } graph = PipelineGraph( nodes=[ node( "source", "source.inline", { "source_name": "source", "rows": [{"amount": 5}, {"amount": 15}], }, x=0, ), node( "subflow", "subflow", { "template_ref": "minimum-filter", "template_version": "1", "parameters": {"minimum": 10}, "graph": nested, }, x=300, ), node("output", "output", {}, x=600), ], edges=[ GraphEdge(id="e1", source="source", target="subflow"), GraphEdge(id="e2", source="subflow", target="output"), ], ) result = execute_preview(graph, row_limit=100) self.assertEqual([{"amount": 15}], result.rows) def test_registry_covers_validation_execution_and_sql_rendering(self) -> None: # Importing the runtime modules above registers each independent facet. coverage = OPERATOR_REGISTRY.coverage() self.assertEqual(set(NODE_TYPES), set(coverage)) for node_type, facets in coverage.items(): self.assertIn("config_validator", facets, node_type) self.assertIn("schema_propagator", facets, node_type) self.assertIn("executor", facets, node_type) self.assertIn("sql_renderer", facets, node_type) def test_maximum_size_linear_graph_validates_and_executes(self) -> None: nodes = [ node( "source", "source.inline", { "source_name": "bounded_input", "rows": [{"value": 1}], }, x=0, ) ] edges: list[GraphEdge] = [] previous_id = "source" for index in range(1, 99): node_id = f"limit-{index}" nodes.append( node( node_id, "limit", {"count": 1}, x=index * 10, ) ) edges.append( GraphEdge( id=f"edge-{index}", source=previous_id, target=node_id, ) ) previous_id = node_id nodes.append(node("output", "output", {}, x=990)) edges.append( GraphEdge( id="edge-99", source=previous_id, target="output", ) ) graph = PipelineGraph(nodes=nodes, edges=edges) self.assertEqual([], validate_graph(graph)) result = execute_preview(graph, row_limit=1) self.assertEqual([{"value": 1}], result.rows) self.assertEqual(100, len(result.node_diagnostics)) def test_expression_node_renders_through_registered_sql_compiler(self) -> None: graph = PipelineGraph( nodes=[ node( "source", "source.inline", { "source_name": "records", "rows": [{"name": "Ada"}], }, x=0, ), node( "expression", "expression", { "target_column": "normalized_name", "expression": "lower(name)", "result_type": "string", }, x=200, ), node("output", "output", {}, x=400), ], edges=[ GraphEdge(id="e1", source="source", target="expression"), GraphEdge(id="e2", source="expression", target="output"), ], ) sql, _ = render_sql(graph) self.assertIn("LOWER(name) AS normalized_name", sql) if __name__ == "__main__": unittest.main()