test_org_level_loop.py
4.15 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-017 tests: the full Org-level Loop wiring (judge -> record -> route ->
fix -> re-verify -> done), mocked at the subprocess boundary (CLAUDE.md
Testing constraint). Real coverage is in src/test_org_level_loop_e2e.py
(RUN_REAL_AGENT_E2E=1).
Run: python3 -m unittest src.test_org_level_loop -v
"""
import unittest
from unittest.mock import patch
from src import org_level_loop, schema
def _make_trigger(perspective_category, fix_ok=True):
"""Fakes org_node_trigger.trigger_org_node for the whole loop: 3
perspective calls (all agreeing on perspective_category), one fixer
call, one verify call."""
def fake(org_node, prompt, target_folder_path, timeout=300):
if "CATEGORY:" in prompt:
return "CATEGORY: {}\nDETAIL: because reasons".format(perspective_category)
if org_node in ("coder", "planner"):
return "Fixed it."
if org_node == "validator":
return "VERDICT: PASS\nlooks fixed" if fix_ok else "VERDICT: FAIL\nstill broken"
raise AssertionError("unexpected call: " + org_node)
return fake
class RunFixLoopTestBase(unittest.TestCase):
def setUp(self):
self.conn = schema.connect(":memory:")
schema.create_schema(self.conn)
self.conn.execute(
"INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
"VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')"
)
self.conn.execute(
"INSERT INTO task (task_id, project_id, goal_text, task_type, status, loop_round, assigned_org_node) "
"VALUES ('task-1', 'proj-1', 'goal', 'goal', 'validating', 0, 'validator')"
)
self.conn.commit()
def tearDown(self):
self.conn.close()
class CoderCategoryFastLaneTest(RunFixLoopTestBase):
def test_coder_category_routes_fast_lane_and_reaches_done(self):
with patch("src.org_node_trigger.trigger_org_node", side_effect=_make_trigger("coder", fix_ok=True)):
result = org_level_loop.run_fix_loop(
self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", "fix it", "verify it"
)
self.assertEqual(result["root_cause_category"], "coder")
self.assertEqual(result["route"], "fast_lane")
self.assertTrue(result["verified"])
self.assertEqual(result["final_status"], "done")
row = self.conn.execute("SELECT status, loop_round FROM task WHERE task_id = 'task-1'").fetchone()
self.assertEqual(row, ("done", 1))
bug_row = self.conn.execute(
"SELECT root_cause_category, suggested_route FROM bug_record WHERE task_id = 'task-1'"
).fetchone()
self.assertEqual(bug_row, ("coder", "fast_lane"))
class PlanningCategoryFullReplanTest(RunFixLoopTestBase):
def test_planning_category_routes_full_replan(self):
with patch("src.org_node_trigger.trigger_org_node", side_effect=_make_trigger("planning", fix_ok=True)):
result = org_level_loop.run_fix_loop(
self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", "fix it", "verify it"
)
self.assertEqual(result["route"], "full_replan")
self.assertEqual(result["final_status"], "done")
assigned = self.conn.execute("SELECT assigned_org_node FROM task WHERE task_id = 'task-1'").fetchone()[0]
# after close_task_done the task is 'done'; assigned_org_node was set
# to 'planner' by route_org_loop and close_task_done doesn't touch it
self.assertEqual(assigned, "planner")
class VerificationFailsTest(RunFixLoopTestBase):
def test_fix_not_verified_leaves_task_out_of_done(self):
with patch("src.org_node_trigger.trigger_org_node", side_effect=_make_trigger("coder", fix_ok=False)):
result = org_level_loop.run_fix_loop(
self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", "fix it", "verify it"
)
self.assertFalse(result["verified"])
self.assertNotEqual(result["final_status"], "done")
status = self.conn.execute("SELECT status FROM task WHERE task_id = 'task-1'").fetchone()[0]
self.assertNotEqual(status, "done")
if __name__ == "__main__":
unittest.main()