test_e2e_pipeline.py
5.46 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
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
"""U-014 tests: end-to-end orchestration wiring, mocked at the subprocess
boundary (CLAUDE.md Testing constraint) -- these test that plan_documents/
implement_code/validate_implementation/run_end_to_end call the right things
in the right order and wire artifact lineage/task-status transitions
correctly, not what a real Planner/Coder/Validator would actually produce.
Real coverage (as much of it as is safe/affordable to run) is in
src/test_e2e_pipeline_e2e.py (RUN_REAL_AGENT_E2E=1).
Run: python3 -m unittest src.test_e2e_pipeline -v
"""
import json
import subprocess
import tempfile
import unittest
from unittest.mock import patch
from src import e2e_pipeline, project_scaffold, schema
_FAKE_DOCS = {"brd": "# BRD\nfake", "prd": "# PRD\nfake", "sdd": "# SDD\nfake", "benchmark": "# Benchmark\nfake"}
def _fake_trigger(org_node, prompt, target_folder_path, timeout=300):
if prompt == e2e_pipeline._READY_PROMPT:
return "Ready."
if "BRD (Business" in prompt:
return _FAKE_DOCS["brd"]
if "PRD (Product" in prompt:
return _FAKE_DOCS["prd"]
if "SDD (System" in prompt:
return _FAKE_DOCS["sdd"]
if "Benchmark/acceptance-test" in prompt:
return _FAKE_DOCS["benchmark"]
if prompt.startswith("Implement this goal"):
return "Implemented. Ran `pytest`: 3 passed."
if prompt.startswith("Independently verify"):
return "VERDICT: PASS\nAll benchmark scenarios reproduced and passing."
raise AssertionError("unexpected prompt: " + prompt[:80])
class PipelineTestBase(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
subprocess.run(["git", "init"], cwd=self.tmp.name, check=True, capture_output=True)
self.conn = schema.connect(":memory:")
schema.create_schema(self.conn)
project_scaffold.create_project(self.conn, self.tmp.name, init_git_if_missing=False)
def tearDown(self):
self.conn.close()
class PlanDocumentsTest(PipelineTestBase):
def test_produces_four_artifacts_with_correct_lineage(self):
self.conn.execute(
"INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
"VALUES ('task-1', (SELECT project_id FROM project), '实现贪吃蛇游戏', 'goal', "
"'planning_brd_prd_sdd', 'planner')"
)
self.conn.execute(
"INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
"VALUES ('run-1', 'task-1', 'planner', '{}', 'running')"
)
self.conn.commit()
with patch("src.org_node_trigger.trigger_org_node", side_effect=_fake_trigger):
artifact_ids = e2e_pipeline.plan_documents(
self.conn, "task-1", "run-1", "实现贪吃蛇游戏", self.tmp.name
)
self.assertEqual(set(artifact_ids), {"brd", "prd", "sdd", "benchmark"})
prd_deps = json.loads(
self.conn.execute(
"SELECT depends_on_artifact_ids FROM artifact WHERE artifact_id = ?", (artifact_ids["prd"],)
).fetchone()[0]
)
self.assertEqual(prd_deps, [artifact_ids["brd"]])
sdd_deps = json.loads(
self.conn.execute(
"SELECT depends_on_artifact_ids FROM artifact WHERE artifact_id = ?", (artifact_ids["sdd"],)
).fetchone()[0]
)
self.assertEqual(sdd_deps, [artifact_ids["prd"]])
class RunEndToEndTest(PipelineTestBase):
def test_full_pipeline_wiring(self):
with patch("src.org_node_lifecycle.trigger_org_node", side_effect=_fake_trigger), \
patch("src.org_node_trigger.trigger_org_node", side_effect=_fake_trigger):
result = e2e_pipeline.run_end_to_end(self.conn, self.tmp.name, "实现贪吃蛇游戏")
self.assertEqual(result["verdict"]["verdict"], "PASS")
self.assertEqual(set(result["doc_artifact_ids"]), {"brd", "prd", "sdd", "benchmark"})
self.assertEqual(set(result["code_artifact_ids"]), {"code_diff", "test_report"})
task_row = self.conn.execute(
"SELECT status FROM task WHERE task_id = ?", (result["task_id"],)
).fetchone()
self.assertEqual(task_row[0], "validating")
all_artifacts = self.conn.execute(
"SELECT COUNT(*) FROM artifact WHERE task_id = ?", (result["task_id"],)
).fetchone()[0]
self.assertEqual(all_artifacts, 6) # brd, prd, sdd, benchmark, code_diff, test_report
class ValidateImplementationTest(unittest.TestCase):
def test_parses_pass_verdict(self):
with patch("src.org_node_trigger.trigger_org_node", return_value="VERDICT: PASS\nlooks good"):
result = e2e_pipeline.validate_implementation("task-1", "benchmark content", "goal", "/tmp/p")
self.assertEqual(result["verdict"], "PASS")
def test_parses_fail_verdict(self):
with patch("src.org_node_trigger.trigger_org_node", return_value="VERDICT: FAIL\nwall collision broken"):
result = e2e_pipeline.validate_implementation("task-1", "benchmark content", "goal", "/tmp/p")
self.assertEqual(result["verdict"], "FAIL")
def test_no_verdict_line_defaults_to_inconclusive(self):
with patch("src.org_node_trigger.trigger_org_node", return_value="I'm not sure what happened."):
result = e2e_pipeline.validate_implementation("task-1", "benchmark content", "goal", "/tmp/p")
self.assertEqual(result["verdict"], "INCONCLUSIVE")
if __name__ == "__main__":
unittest.main()