test_org_node_lifecycle_e2e.py
2.96 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
"""U-010's real-call coverage: "手工依次触发 planner/coder/validator 三个 Org
Agent" (05 §5.5) with actual `claude --agent <node> -p` calls for all three,
not just planner (which U-004 already validated). Artifact-validation
preconditions are seeded directly -- no real Work Graph exists yet to
produce them (U-011+), same as the mocked tests. Gated behind
RUN_REAL_AGENT_E2E=1 (CLAUDE.md Testing constraint).
Run: RUN_REAL_AGENT_E2E=1 python3 -m unittest src.test_org_node_lifecycle_e2e -v
"""
import subprocess
import tempfile
import unittest
import uuid
from src import agent_profile, org_node_lifecycle, project_scaffold, schema, task_create
from src.test_support import skip_unless_env
_PROMPT = "In one short sentence, confirm you're ready and state your role. Take no other action."
@skip_unless_env("RUN_REAL_AGENT_E2E")
class RealChainedTriggerTest(unittest.TestCase):
def test_all_three_org_agents_trigger_for_real_and_status_advances_each_time(self):
with tempfile.TemporaryDirectory() as tmp:
subprocess.run(["git", "init"], cwd=tmp, check=True, capture_output=True)
conn = schema.connect(":memory:")
schema.create_schema(conn)
project_scaffold.create_project(conn, tmp, init_git_if_missing=False)
agent_profile.deploy_agent_definitions(tmp)
task_id = task_create.create_task(conn, self._project_id(conn), "实现贪吃蛇游戏")["task_id"]
planner_result = org_node_lifecycle.trigger_org_node_full(conn, "planner", task_id, tmp, _PROMPT)
self.assertEqual(planner_result["status"], "planning_brd_prd_sdd")
self._seed_passing_artifacts(conn, task_id, ("brd", "prd", "sdd", "benchmark"))
coder_result = org_node_lifecycle.trigger_org_node_full(conn, "coder", task_id, tmp, _PROMPT)
self.assertEqual(coder_result["status"], "coding")
self._seed_passing_artifacts(conn, task_id, ("code_diff", "test_report"))
validator_result = org_node_lifecycle.trigger_org_node_full(conn, "validator", task_id, tmp, _PROMPT)
self.assertEqual(validator_result["status"], "validating")
@staticmethod
def _project_id(conn):
return conn.execute("SELECT project_id FROM project").fetchone()[0]
@staticmethod
def _seed_passing_artifacts(conn, task_id, artifact_types):
run_id = str(uuid.uuid4())
conn.execute(
"INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) VALUES (?, ?, '{}', 'done')",
(run_id, task_id),
)
for artifact_type in artifact_types:
conn.execute(
"INSERT INTO artifact (artifact_id, artifact_type, task_id, produced_by_run_id, "
"file_path, git_commit_sha, validation_status) VALUES (?, ?, ?, ?, 'f.md', ?, 'passed')",
(str(uuid.uuid4()), artifact_type, task_id, run_id, artifact_type),
)
conn.commit()
if __name__ == "__main__":
unittest.main()