test_new_goal_dispatch.py
5.43 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
"""new_goal_dispatch.py tests (2026-08-13, FR-1 --
docs/done/scheduler-new-goal-dispatch-requirements.md): same mocked-at-the-
subprocess-boundary / real-sqlite split as test_task_reconciliation.py.
Run: python3 -m unittest src.test_new_goal_dispatch -v
"""
import subprocess
import tempfile
import unittest
from unittest.mock import patch
from src import new_goal_dispatch, project_scaffold, schema, task_execution
class NewGoalDispatchTestBase(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 = project_scaffold.create_project(self.conn, self.tmp.name, init_git_if_missing=False)
self.project_id = project["project_id"]
def tearDown(self):
self.conn.close()
def _insert_task(self, task_id, status, goal_text="a goal"):
self.conn.execute(
"INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node, created_at) "
"VALUES (?, ?, ?, 'goal', ?, 'planner', CURRENT_TIMESTAMP)",
(task_id, self.project_id, goal_text, status),
)
self.conn.commit()
class FindQueuedTasksTest(NewGoalDispatchTestBase):
def test_finds_queued_task(self):
self._insert_task("task-1", "queued")
result = new_goal_dispatch.find_queued_tasks(self.conn)
self.assertEqual(result, [{"task_id": "task-1", "target_folder_path": self.tmp.name, "goal_text": "a goal"}])
def test_skips_non_queued_statuses(self):
for status in ("coding", "validating", "done", "blocked_pending_human", "planning_brd_prd_sdd"):
self._insert_task("task-" + status, status)
result = new_goal_dispatch.find_queued_tasks(self.conn)
self.assertEqual(result, [])
def test_excludes_in_flight_task(self):
self._insert_task("task-1", "queued")
with patch.object(task_execution, "is_in_flight", return_value=True):
result = new_goal_dispatch.find_queued_tasks(self.conn)
self.assertEqual(result, [])
def test_no_candidates_returns_empty_list(self):
result = new_goal_dispatch.find_queued_tasks(self.conn)
self.assertEqual(result, [])
class DispatchOnceTest(unittest.TestCase):
"""Real file db (own connection, mirrors ReconcileOnceTest) -- dispatch
itself is mocked so this stays at the "did we correctly decide to call
it" boundary."""
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.tmp_db = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
self.tmp_db.close()
self.db_path = self.tmp_db.name
self.addCleanup(lambda: __import__("os").path.exists(self.db_path) and __import__("os").remove(self.db_path))
conn = schema.connect(self.db_path)
schema.create_schema(conn)
project = project_scaffold.create_project(conn, self.tmp.name, init_git_if_missing=False)
self.project_id = project["project_id"]
conn.execute(
"INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node, created_at) "
"VALUES ('queued-1', ?, 'goal 1', 'goal', 'queued', 'planner', CURRENT_TIMESTAMP)",
(self.project_id,),
)
conn.execute(
"INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node, created_at) "
"VALUES ('coding-1', ?, 'goal 2', 'goal', 'coding', 'coder', CURRENT_TIMESTAMP)",
(self.project_id,),
)
conn.commit()
conn.close()
def test_dispatches_only_queued_candidates(self):
with patch.object(task_execution, "start_task_execution_in_background") as mock_start:
result = new_goal_dispatch.dispatch_once(self.db_path)
self.assertEqual([c["task_id"] for c in result], ["queued-1"])
mock_start.assert_called_once_with(self.db_path, "queued-1", self.tmp.name, "goal 1")
def test_one_dispatch_failure_does_not_block_others(self):
conn = schema.connect(self.db_path)
conn.execute(
"INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node, created_at) "
"VALUES ('queued-2', ?, 'goal 3', 'goal', 'queued', 'planner', CURRENT_TIMESTAMP)",
(self.project_id,),
)
conn.commit()
conn.close()
with patch.object(
task_execution, "start_task_execution_in_background",
side_effect=[task_execution.TaskExecutionError("already in flight"), None],
) as mock_start:
result = new_goal_dispatch.dispatch_once(self.db_path)
self.assertEqual(len(result), 2)
self.assertEqual(mock_start.call_count, 2)
def test_no_candidates_returns_gracefully(self):
conn = schema.connect(self.db_path)
conn.execute("UPDATE task SET status = 'done'")
conn.commit()
conn.close()
with patch.object(task_execution, "start_task_execution_in_background") as mock_start:
result = new_goal_dispatch.dispatch_once(self.db_path)
self.assertEqual(result, [])
mock_start.assert_not_called()
if __name__ == "__main__":
unittest.main()