sanguo/sanguo_moziplus_v2
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

auto-sync: 2026-05-18 16:12:44

Browse Source
This commit is contained in:
cfdaily
2026-05-18 16:12:44 +08:00
parent c0ec72835e
commit 743b4e3fef
1 changed files with 16 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_e2e_v27.py
+16 -17
View File
@@ -345,23 +345,23 @@ class TestE5ParentAggregation:
def _create_child(self, cid: str, status: str = "pending", stage: str = None):
self.bb.create_task(Task(
id=cid, title=f"子-{cid}", status=status,
id=cid, title=f"子-{cid}", status="pending",
parent_task=self.parent_id, stage=stage,
))
# 如果不是 pending，需要走合法转换
if status != "pending":
for s in ["claimed", "working"]:
if s in VALID_TRANSITIONS.get("pending", set()):
self.bb.update_task_status(cid, s, agent="test")
if status == "review":
self.bb.update_task_status(cid, "review", agent="test")
elif status == "done":
self.bb.update_task_status(cid, "review", agent="test")
self.bb.update_task_status(cid, "done", agent="test")
elif status == "failed":
self.bb.update_task_status(cid, "failed", agent="test")
elif status == "cancelled":
self.bb.update_task_status(cid, "cancelled", agent="test")
if status == "pending":
return
# Walk state machine to target status
path_map = {
"claimed": ["claimed"],
"working": ["claimed", "working"],
"review": ["claimed", "working", "review"],
"done": ["claimed", "working", "review", "done"],
"failed": ["claimed", "working", "failed"],
"blocked": ["claimed", "working", "blocked"],
"cancelled": ["cancelled"],
}
for s in path_map.get(status, []):
self.bb.update_task_status(cid, s, agent="test")
def test_e51_all_done_parent_done(self):
self._create_child("c1", "done")
@@ -377,8 +377,7 @@ class TestE5ParentAggregation:
def test_e53_has_working_parent_working(self):
self._create_child("c5", "done")
self._create_child("c6", "claimed")
self.bb.update_task_status("c6", "working", agent="test")
self._create_child("c6", "working")
result = self.q.compute_parent_status(self.parent_id)
assert result == "working"