feat: Model to track node states and has_ended flag

Author: yuechao-qin

cloud_pipelines_backend/api_server_sql.py

@@ -475,6 +475,22 @@ class ArtifactNodeIdResponse:
     id: bts.IdType
 
 
+@dataclasses.dataclass(kw_only=True)
+class ExecutionSummary:
+    total_nodes: int = 0
+    ended_nodes: int = 0
+    has_ended: bool = False
+
+    def count_node_status(
+        self, *, status: bts.ContainerExecutionStatus, count: int
+    ) -> None:
+        self.total_nodes += count
+        if status in bts.CONTAINER_STATUSES_ENDED:
+            self.ended_nodes += count
+
+        self.has_ended = self.total_nodes > 0 and self.ended_nodes == self.total_nodes
+
+
 @dataclasses.dataclass
 class GetGraphExecutionStateResponse:
     child_execution_status_stats: dict[bts.IdType, dict[str, int]]

tests/test_api_server_sql.py

@@ -0,0 +1,50 @@
+from cloud_pipelines_backend import backend_types_sql as bts
+from cloud_pipelines_backend.api_server_sql import ExecutionSummary
+
+
+class TestExecutionSummary:
+    def test_initial_state(self):
+        summary = ExecutionSummary()
+        assert summary.total_nodes == 0
+        assert summary.ended_nodes == 0
+        assert summary.has_ended is False
+
+    def test_accumulate_all_ended_statuses(self):
+        """Add each ended status one by one, confirming counts after each."""
+        summary = ExecutionSummary()
+        ended_statuses = sorted(bts.CONTAINER_STATUSES_ENDED, key=lambda s: s.value)
+        for i, status in enumerate(ended_statuses, start=1):
+            summary.count_node_status(status=status, count=1)
+            assert summary.total_nodes == i
+            assert summary.ended_nodes == i
+            assert summary.has_ended is True
+
+    def test_accumulate_all_in_progress_statuses(self):
+        """Add each in-progress status one by one, confirming counts after each."""
+        summary = ExecutionSummary()
+        in_progress_statuses = sorted(
+            set(bts.ContainerExecutionStatus) - bts.CONTAINER_STATUSES_ENDED,
+            key=lambda s: s.value,
+        )
+        for i, status in enumerate(in_progress_statuses, start=1):
+            summary.count_node_status(status=status, count=1)
+            assert summary.total_nodes == i
+            assert summary.ended_nodes == 0
+            assert summary.has_ended is False
+
+    def test_accumulate_all_statuses(self):
+        """Add every ContainerExecutionStatus one by one, confirming counts after each."""
+        summary = ExecutionSummary()
+        all_statuses = sorted(bts.ContainerExecutionStatus, key=lambda s: s.value)
+        expected_total = 0
+        expected_ended = 0
+        for status in all_statuses:
+            expected_total += 1
+            if status in bts.CONTAINER_STATUSES_ENDED:
+                expected_ended += 1
+            summary.count_node_status(status=status, count=1)
+            assert summary.total_nodes == expected_total
+            assert summary.ended_nodes == expected_ended
+            assert summary.has_ended == (
+                expected_total > 0 and expected_ended == expected_total
+            )