Harden Dataflow preview behavior

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()

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()