auto-sync: 2026-05-18 12:08:50

tests/test_v27_subtasks.py

@@ -280,23 +280,44 @@ class TestStagesJson:
 
 class TestTickerParentRefresh:
     def test_refresh_parent_status(self, bb, queries):
-        """聚合刷新写入父 Task"""
+        """Ticker._refresh_parent_statuses 聚合写入父 Task"""
         from src.daemon.ticker import Ticker
-        from src.blackboard.registry import ProjectRegistry
 
         bb.create_task(Task(id="p1", title="Parent"))
         bb.create_task(Task(id="c1", title="Child", parent_task="p1", status="done"))
         bb.create_task(Task(id="c2", title="Child", parent_task="p1", status="done"))
 
-        # 手动刷新
-        conn = get_connection(bb.db_path)
-        try:
-            # 模拟 Ticker 的 _refresh_parent_statuses
-            computed = queries.compute_parent_status("p1")
-            conn.execute("UPDATE tasks SET status=? WHERE id=?", (computed, "p1"))
-            conn.commit()
-        finally:
-            conn.close()
+        ticker = Ticker(registry=None, dispatcher=None, spawner=None)
+        refreshed = ticker._refresh_parent_statuses(bb.db_path)
 
+        assert "p1" in refreshed
         parent = bb.get_task("p1")
         assert parent.status == "done"
+
+    def test_refresh_skips_manual_status(self, bb, queries):
+        """cancelled 的父 Task 不参与聚合"""
+        from src.daemon.ticker import Ticker
+
+        bb.create_task(Task(id="p1", title="Parent", status="cancelled"))
+        bb.create_task(Task(id="c1", title="Child", parent_task="p1", status="done"))
+
+        ticker = Ticker(registry=None, dispatcher=None, spawner=None)
+        refreshed = ticker._refresh_parent_statuses(bb.db_path)
+
+        assert "p1" not in refreshed
+        parent = bb.get_task("p1")
+        assert parent.status == "cancelled"
+
+    def test_refresh_failed_over_blocked(self, bb, queries):
+        """failed 优先于 blocked"""
+        from src.daemon.ticker import Ticker
+
+        bb.create_task(Task(id="p1", title="Parent"))
+        bb.create_task(Task(id="c1", title="Child", parent_task="p1", status="blocked"))
+        bb.create_task(Task(id="c2", title="Child", parent_task="p1", status="failed"))
+
+        ticker = Ticker(registry=None, dispatcher=None, spawner=None)
+        ticker._refresh_parent_statuses(bb.db_path)
+
+        parent = bb.get_task("p1")
+        assert parent.status == "failed"