test_e2e_pipeline_e2e.py
4.74 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
"""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))
if __name__ == "__main__":
unittest.main()