test_org_loop_route.py
7.1 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
"""U-016 tests: c_org_loop_route + close_task_done. All real (no mocking
needed -- sqlite is fast/free/local).
Run: python3 -m unittest src.test_org_loop_route -v
"""
import json
import unittest
from src import org_loop_route, schema
class OrgLoopRouteTestBase(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.commit()
def tearDown(self):
self.conn.close()
def _make_task(self, task_id="task-1", status="validating", loop_round=0, assigned_org_node="validator"):
self.conn.execute(
"INSERT INTO task (task_id, project_id, goal_text, task_type, status, loop_round, assigned_org_node) "
"VALUES (?, 'proj-1', 'goal', 'goal', ?, ?, ?)",
(task_id, status, loop_round, assigned_org_node),
)
self.conn.commit()
return task_id
class CloseTaskDoneTest(OrgLoopRouteTestBase):
def test_validating_to_done(self):
self._make_task()
result = org_loop_route.close_task_done(self.conn, "task-1")
self.assertEqual(result["new_status"], "done")
status = self.conn.execute("SELECT status FROM task WHERE task_id = 'task-1'").fetchone()[0]
self.assertEqual(status, "done")
def test_rejects_non_validating_task(self):
self._make_task(status="coding")
with self.assertRaises(org_loop_route.TaskNotValidatingError):
org_loop_route.close_task_done(self.conn, "task-1")
class FastLaneRoutingTest(OrgLoopRouteTestBase):
"""REQ-05 happy: fast_lane routes straight to Coder, loop_round increments."""
def test_routes_to_coding_and_increments_loop_round(self):
self._make_task(loop_round=0)
result = org_loop_route.route_org_loop(self.conn, "task-1", "fast_lane")
self.assertEqual(result["new_status"], "coding")
self.assertEqual(result["new_assigned_org_node"], "coder")
row = self.conn.execute("SELECT status, assigned_org_node, loop_round FROM task WHERE task_id = 'task-1'").fetchone()
self.assertEqual(row, ("coding", "coder", 1))
def test_writes_loop_routed_event(self):
self._make_task()
org_loop_route.route_org_loop(self.conn, "task-1", "fast_lane")
row = self.conn.execute(
"SELECT payload FROM execution_event WHERE task_id = 'task-1' AND event_type = 'loop_routed'"
).fetchone()
self.assertEqual(json.loads(row[0]), {"decided_route": "fast_lane"})
def test_full_chain_reaches_done_after_fix(self):
# REQ-05 happy, full chain: fast_lane -> coding -> (fix happens
# elsewhere, U-013) -> validating -> done, loop_round ends at 1
self._make_task(loop_round=0)
org_loop_route.route_org_loop(self.conn, "task-1", "fast_lane")
self.conn.execute("UPDATE task SET status = 'validating' WHERE task_id = 'task-1'")
self.conn.commit()
result = org_loop_route.close_task_done(self.conn, "task-1")
self.assertEqual(result["new_status"], "done")
loop_round = self.conn.execute("SELECT loop_round FROM task WHERE task_id = 'task-1'").fetchone()[0]
self.assertEqual(loop_round, 1)
class FullReplanRoutingTest(OrgLoopRouteTestBase):
"""REQ-05 boundary: full_replan goes back to Planner, not Coder."""
def test_routes_to_planning_not_coding(self):
self._make_task()
result = org_loop_route.route_org_loop(self.conn, "task-1", "full_replan")
self.assertEqual(result["new_status"], "planning_brd_prd_sdd")
self.assertEqual(result["new_assigned_org_node"], "planner")
class LoopLimitTest(OrgLoopRouteTestBase):
"""REQ-05 failure: 3 unconverged rounds -> blocked_pending_human."""
def test_third_route_attempt_at_limit_blocks_and_raises(self):
self._make_task(loop_round=3)
with self.assertRaises(org_loop_route.LoopLimitReachedError):
org_loop_route.route_org_loop(self.conn, "task-1", "fast_lane")
row = self.conn.execute("SELECT status, blocked_reason FROM task WHERE task_id = 'task-1'").fetchone()
self.assertEqual(row, ("blocked_pending_human", "loop_limit_reached"))
def test_scheduler_can_still_progress_other_tasks(self):
# REQ-09 boundary: one blocked task doesn't lock out others -- no
# scheduler exists yet to test against directly, but this proves
# route_org_loop/close_task_done have no cross-task side effects
self._make_task("task-1", loop_round=3)
self._make_task("task-2", loop_round=0)
with self.assertRaises(org_loop_route.LoopLimitReachedError):
org_loop_route.route_org_loop(self.conn, "task-1", "fast_lane")
result = org_loop_route.route_org_loop(self.conn, "task-2", "fast_lane")
self.assertEqual(result["new_status"], "coding")
class ManualResumeTest(OrgLoopRouteTestBase):
"""REQ-09 happy/failure: manual_resume restores pre-blocked state, or
errors clearly if the task isn't actually blocked."""
def test_restores_status_before_it_was_blocked(self):
self._make_task(loop_round=3)
with self.assertRaises(org_loop_route.LoopLimitReachedError):
org_loop_route.route_org_loop(self.conn, "task-1", "fast_lane")
# task was 'validating' immediately before the loop-limit block
result = org_loop_route.route_org_loop(self.conn, "task-1", "manual_resume")
self.assertEqual(result["new_status"], "validating")
status = self.conn.execute("SELECT status, blocked_reason FROM task WHERE task_id = 'task-1'").fetchone()
self.assertEqual(status, ("validating", None))
def test_rejects_resume_on_non_blocked_task(self):
self._make_task(status="done")
with self.assertRaises(org_loop_route.TaskNotBlockedError):
org_loop_route.route_org_loop(self.conn, "task-1", "manual_resume")
def test_resume_after_multiple_block_cycles_uses_most_recent(self):
self._make_task(loop_round=3)
with self.assertRaises(org_loop_route.LoopLimitReachedError):
org_loop_route.route_org_loop(self.conn, "task-1", "fast_lane")
org_loop_route.route_org_loop(self.conn, "task-1", "manual_resume")
# force loop_round back up and block again from a *different* status
self.conn.execute("UPDATE task SET status = 'coding', loop_round = 3 WHERE task_id = 'task-1'")
self.conn.commit()
with self.assertRaises(org_loop_route.LoopLimitReachedError):
# fast_lane routing requires status=validating, so force that
# first to hit the loop-limit branch again from a known state
self.conn.execute("UPDATE task SET status = 'validating' WHERE task_id = 'task-1'")
self.conn.commit()
org_loop_route.route_org_loop(self.conn, "task-1", "fast_lane")
result = org_loop_route.route_org_loop(self.conn, "task-1", "manual_resume")
self.assertEqual(result["new_status"], "validating")
if __name__ == "__main__":
unittest.main()