diff --git a/README.md b/README.md index 33f85bb..7c09143 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ GovOPlaN Dataflow defines and runs governed tabular transformation pipelines. Power users can work with the same pipeline as a graphical node graph or as a -constrained SQL query. Every saved change produces an immutable revision, and +constrained SQL query. Every saved definition change produces an immutable revision, and every preview records diagnostics and reproducibility metadata without storing the previewed row contents. diff --git a/src/govoplan_dataflow/backend/executor.py b/src/govoplan_dataflow/backend/executor.py index 1e5f253..05f6583 100644 --- a/src/govoplan_dataflow/backend/executor.py +++ b/src/govoplan_dataflow/backend/executor.py @@ -267,10 +267,10 @@ def _sort_rows(rows: list[dict[str, Any]], config: dict[str, Any]) -> list[dict[ for field in reversed(config["fields"]): column = str(field["column"]) reverse = field.get("direction", "asc") == "desc" - result.sort( - key=lambda row: (row.get(column) is None, _sortable_value(row.get(column))), - reverse=reverse, - ) + concrete = [row for row in result if row.get(column) is not None] + nulls = [row for row in result if row.get(column) is None] + concrete.sort(key=lambda row: _sortable_value(row[column]), reverse=reverse) + result = [*concrete, *nulls] return result diff --git a/tests/test_graph_and_sql.py b/tests/test_graph_and_sql.py index 84bb0e1..4dd0f9b 100644 --- a/tests/test_graph_and_sql.py +++ b/tests/test_graph_and_sql.py @@ -2,7 +2,7 @@ from __future__ import annotations import unittest -from govoplan_dataflow.backend.executor import execute_preview +from govoplan_dataflow.backend.executor import PipelineExecutionError, execute_preview from govoplan_dataflow.backend.schemas import ( GraphEdge, GraphNode, @@ -120,6 +120,89 @@ class DataflowGraphAndSqlTests(unittest.TestCase): self.assertEqual(3, result.total_rows) self.assertTrue(result.truncated) + def test_empty_aggregate_and_null_sort_are_deterministic(self) -> None: + graph, _, _ = compile_sql( + """ + SELECT status, COUNT(*) AS records + FROM monthly_files + GROUP BY status + ORDER BY status DESC + """, + source_nodes=[ + inline_source().model_copy( + update={ + "config": { + "source_name": "monthly_files", + "rows": [ + {"status": None}, + {"status": "open"}, + {"status": "closed"}, + ], + } + } + ) + ], + ) + + first = execute_preview(graph, row_limit=100) + second = execute_preview(graph, row_limit=100) + + self.assertEqual(first.rows, second.rows) + self.assertEqual( + [ + {"status": "open", "records": 1}, + {"status": "closed", "records": 1}, + {"status": None, "records": 1}, + ], + first.rows, + ) + self.assertEqual( + [ + diagnostic.model_dump(exclude={"duration_ms"}) + for diagnostic in first.node_diagnostics + ], + [ + diagnostic.model_dump(exclude={"duration_ms"}) + for diagnostic in second.node_diagnostics + ], + ) + + def test_empty_global_aggregate_returns_a_count_row(self) -> None: + graph, _, _ = compile_sql( + "SELECT COUNT(*) AS records FROM monthly_files", + source_nodes=[ + inline_source().model_copy( + update={"config": {"source_name": "monthly_files", "rows": []}} + ) + ], + ) + + result = execute_preview(graph, row_limit=100) + + self.assertEqual([{"records": 0}], result.rows) + self.assertEqual(0, result.input_row_count) + + def test_invalid_runtime_comparison_is_scoped_to_the_node(self) -> None: + graph, _, _ = compile_sql( + "SELECT * FROM monthly_files WHERE amount > 10", + source_nodes=[ + inline_source().model_copy( + update={ + "config": { + "source_name": "monthly_files", + "rows": [{"amount": "not-a-number"}], + } + } + ) + ], + ) + + with self.assertRaises(PipelineExecutionError) as raised: + execute_preview(graph, row_limit=100) + + self.assertEqual("filter-1", raised.exception.node_id) + self.assertIn("Cannot apply", str(raised.exception)) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_service.py b/tests/test_service.py index 43e74a3..09371b4 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -24,7 +24,9 @@ from govoplan_dataflow.backend.service import ( DataflowConflictError, DataflowNotFoundError, create_pipeline, + delete_pipeline, get_pipeline, + list_pipelines, preview_pipeline, update_pipeline, ) @@ -168,6 +170,37 @@ class DataflowServiceTests(unittest.TestCase): pipeline_id=pipeline.id, ) + def test_list_and_soft_delete_are_tenant_isolated(self) -> None: + first = self._create() + second = self._create(tenant_id="tenant-2") + + self.assertEqual([first.id], [item.id for item in list_pipelines(self.session, tenant_id="tenant-1")]) + self.assertEqual([second.id], [item.id for item in list_pipelines(self.session, tenant_id="tenant-2")]) + + delete_pipeline( + self.session, + tenant_id="tenant-1", + pipeline_id=first.id, + actor_id="user-2", + ) + self.session.commit() + + self.assertEqual([], list_pipelines(self.session, tenant_id="tenant-1")) + with self.assertRaises(DataflowNotFoundError): + get_pipeline( + self.session, + tenant_id="tenant-1", + pipeline_id=first.id, + ) + self.assertEqual( + 1, + self.session.scalar( + select(func.count()) + .select_from(DataflowPipelineRevision) + .where(DataflowPipelineRevision.pipeline_id == first.id) + ), + ) + def test_saved_preview_records_lineage_but_not_result_rows(self) -> None: pipeline = self._create()