test_e2e_pipeline_e2e.py
8.06 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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
"""U-014's real-call coverage, scoped deliberately: plan_documents (4 real
Planner calls producing BRD/PRD/SDD/Benchmark for the snake-game scenario
REQ-01~14 uses throughout the SDD) is cheap and fast, like every other real
test this session (~$0.05-0.15/call). The Coder/Validator half
(run_end_to_end in full) is a different order of magnitude -- real code
generation + real test execution, plausibly minutes and real $ -- and isn't
run automatically here; see the session notes for why.
Run: RUN_REAL_AGENT_E2E=1 python3 -m unittest src.test_e2e_pipeline_e2e -v
"""
import json
import subprocess
import tempfile
import unittest
from src import agent_profile, e2e_pipeline, project_scaffold, schema, task_create
from src.test_support import skip_unless_env
@skip_unless_env("RUN_REAL_AGENT_E2E")
class RealPlanDocumentsTest(unittest.TestCase):
def test_planner_produces_four_real_lineaged_artifacts(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)
project_id = conn.execute("SELECT project_id FROM project").fetchone()[0]
task_id = task_create.create_task(conn, project_id, "实现贪吃蛇游戏")["task_id"]
# plan_documents (2026-08-01 rewiring) creates its own real
# work_graph_run via plan_work_graph() -- no manual seed row
# needed anymore.
plan_result = e2e_pipeline.plan_documents(conn, task_id, "实现贪吃蛇游戏", tmp)
artifact_ids = plan_result["artifact_ids"]
for artifact_type, artifact_id in artifact_ids.items():
row = conn.execute(
"SELECT validation_status, file_path FROM artifact WHERE artifact_id = ?", (artifact_id,)
).fetchone()
self.assertEqual(row[0], "passed")
with open(tmp + "/" + row[1], encoding="utf-8") as f:
content = f.read()
self.assertTrue(len(content) > 100, "{} content suspiciously short: {!r}".format(artifact_type, content))
prd_deps = json.loads(
conn.execute(
"SELECT depends_on_artifact_ids FROM artifact WHERE artifact_id = ?", (artifact_ids["prd"],)
).fetchone()[0]
)
self.assertEqual(prd_deps, [artifact_ids["brd"]])
@skip_unless_env("RUN_REAL_AGENT_E2E")
class RealPrdConsistencyCheckTest(unittest.TestCase):
"""REQ-02 boundary, real coverage (2026-08-02): the new _check_prd_consistency
call is a genuinely new LLM-judgment capability, not a wiring change --
validated for real once each way (contradiction / no contradiction),
same discipline as every other new agent() call this project has added."""
def test_detects_a_real_deliberate_contradiction(self):
with tempfile.TemporaryDirectory() as tmp:
subprocess.run(["git", "init"], cwd=tmp, check=True, capture_output=True)
agent_profile.deploy_agent_definitions(tmp)
contradictory_prd = (
"# PRD: Snake Game\n\n"
"## Wall Collision\n"
"Requirement 3.1: When the snake's head touches the board boundary, the game MUST "
"immediately end with a 'Game Over' screen.\n\n"
"Requirement 3.2: The snake must be able to pass through the board boundary and "
"wrap around to the opposite side, continuing play without ending the game.\n"
)
ok, issues = e2e_pipeline._check_prd_consistency(tmp, contradictory_prd)
self.assertFalse(ok)
self.assertTrue(len(issues) > 0)
def test_passes_a_real_consistent_prd(self):
with tempfile.TemporaryDirectory() as tmp:
subprocess.run(["git", "init"], cwd=tmp, check=True, capture_output=True)
agent_profile.deploy_agent_definitions(tmp)
consistent_prd = (
"# PRD: Snake Game\n\n"
"## Wall Collision\n"
"Requirement 3.1: When the snake's head touches the board boundary, the game "
"immediately ends with a 'Game Over' screen showing the final score.\n\n"
"## Food\n"
"Requirement 4.1: Eating food grows the snake by one segment and spawns a new "
"food item at a random empty cell.\n"
)
ok, issues = e2e_pipeline._check_prd_consistency(tmp, consistent_prd)
self.assertTrue(ok, "false positive on a consistent PRD: {}".format(issues))
@skip_unless_env("RUN_REAL_AGENT_E2E")
class RealPlannerReviewTest(unittest.TestCase):
"""Org Graph unification Phase 1, real coverage (2026-08-08): the new
planner_reviewer role is a genuinely new LLM-judgment capability (not a
wiring change) -- validated for real once each way (real narrative-text
repro / real structured content), same discipline as
RealPrdConsistencyCheckTest above. Also the cheapest real check that the
new role actually deploys correctly (.claude/agents/planner_reviewer.md
is discoverable and loadable by a real `claude --agent planner_reviewer`
call) -- if role.json/SOUL.md/AGENTS.md have a real wiring mistake, this
is where it surfaces, not deep inside a full run_end_to_end() call."""
def test_detects_real_narrative_text_instead_of_structured_content(self):
with tempfile.TemporaryDirectory() as tmp:
subprocess.run(["git", "init"], cwd=tmp, check=True, capture_output=True)
agent_profile.deploy_agent_definitions(tmp)
conn = schema.connect(":memory:")
schema.create_schema(conn)
project_scaffold.create_project(conn, tmp, init_git_if_missing=False)
project_id = conn.execute("SELECT project_id FROM project").fetchone()[0]
task_id = task_create.create_task(conn, project_id, "实现贪吃蛇游戏")["task_id"]
# real repro text from docs/todo/validator-test-script-narrative-
# output-issue.md, fed in as the SDD content.
narrative_sdd = (
"Launched the 'graphlemon-node-f9cefb87af4c492aac738b9736d41e4e' workflow "
"(Work Node validator) in the background, run ID wf_93345ca8-8c6. It's running "
"asynchronously — I'll report results once it completes; no further action is "
"needed right now."
)
result = e2e_pipeline.review_planner_output(
conn, task_id, "实现贪吃蛇游戏", tmp, sdd_content=narrative_sdd,
)
self.assertIn(result["verdict"], ("FAIL", "INCONCLUSIVE"))
def test_passes_real_structured_content(self):
with tempfile.TemporaryDirectory() as tmp:
subprocess.run(["git", "init"], cwd=tmp, check=True, capture_output=True)
agent_profile.deploy_agent_definitions(tmp)
conn = schema.connect(":memory:")
schema.create_schema(conn)
project_scaffold.create_project(conn, tmp, init_git_if_missing=False)
project_id = conn.execute("SELECT project_id FROM project").fetchone()[0]
task_id = task_create.create_task(conn, project_id, "实现一个返回两数之和的函数")["task_id"]
real_sdd = (
"# SDD: add(a, b)\n\n"
"## Module\n\n`add(a: int, b: int) -> int` in `math_utils.py`.\n\n"
"## Behavior\n\nReturns the sum of the two arguments. No validation needed -- "
"both arguments are guaranteed to be integers by the caller.\n\n"
"## Tests\n\n`test_add_positive`, `test_add_negative`, `test_add_zero`.\n"
)
result = e2e_pipeline.review_planner_output(
conn, task_id, "实现一个返回两数之和的函数", tmp, sdd_content=real_sdd,
)
self.assertEqual(result["verdict"], "PASS", result["raw_response"])
if __name__ == "__main__":
unittest.main()