test_task_query.py
6.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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
"""U-025 tests: c_task_list / c_task_get (03-C §3.2), real sqlite (no
mocking needed).
Run: python3 -m unittest src.test_task_query -v
"""
import json
import unittest
from src import schema, task_query
class TaskQueryTestBase(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 _insert_task(self, task_id, status="queued", assigned_org_node="planner", goal_text="goal"):
self.conn.execute(
"INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
"VALUES (?, 'proj-1', ?, 'goal', ?, ?)",
(task_id, goal_text, status, assigned_org_node),
)
self.conn.commit()
class ListTasksTest(TaskQueryTestBase):
def test_lists_tasks_for_project(self):
self._insert_task("task-1")
self._insert_task("task-2")
result = task_query.list_tasks(self.conn, "proj-1")
self.assertEqual({t["task_id"] for t in result}, {"task-1", "task-2"})
def test_other_projects_excluded(self):
self.conn.execute(
"INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
"VALUES ('proj-2', 'P2', '/tmp/proj-2', 1, 'active')"
)
self.conn.commit()
self._insert_task("task-1")
self.conn.execute(
"INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
"VALUES ('task-2', 'proj-2', 'g', 'goal', 'queued', 'planner')"
)
self.conn.commit()
result = task_query.list_tasks(self.conn, "proj-1")
self.assertEqual([t["task_id"] for t in result], ["task-1"])
def test_status_filter(self):
self._insert_task("task-1", status="queued")
self._insert_task("task-2", status="done")
result = task_query.list_tasks(self.conn, "proj-1", status="done")
self.assertEqual([t["task_id"] for t in result], ["task-2"])
def test_org_node_filter(self):
self._insert_task("task-1", assigned_org_node="coder")
self._insert_task("task-2", assigned_org_node="validator")
result = task_query.list_tasks(self.conn, "proj-1", org_node="coder")
self.assertEqual([t["task_id"] for t in result], ["task-1"])
def test_keyword_filter_matches_goal_text_substring(self):
self._insert_task("task-1", goal_text="implement snake game")
self._insert_task("task-2", goal_text="fix login bug")
result = task_query.list_tasks(self.conn, "proj-1", keyword="snake")
self.assertEqual([t["task_id"] for t in result], ["task-1"])
def test_empty_project_returns_empty_list(self):
self.assertEqual(task_query.list_tasks(self.conn, "proj-1"), [])
class GetTaskTest(TaskQueryTestBase):
def test_nonexistent_task_raises(self):
with self.assertRaises(task_query.TaskNotFoundError):
task_query.get_task(self.conn, "no-such-task")
def test_returns_task_fields(self):
self._insert_task("task-1", status="coding", assigned_org_node="coder")
result = task_query.get_task(self.conn, "task-1")
self.assertEqual(result["task_id"], "task-1")
self.assertEqual(result["status"], "coding")
self.assertEqual(result["assigned_org_node"], "coder")
self.assertEqual(result["artifacts"], [])
self.assertEqual(result["bug_records"], [])
self.assertEqual(result["work_graph_runs"], [])
def test_includes_artifacts(self):
self._insert_task("task-1")
self.conn.execute(
"INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
"VALUES ('run-1', 'task-1', 'planner', '{}', 'done')"
)
self.conn.execute(
"INSERT INTO artifact (artifact_id, artifact_type, task_id, produced_by_run_id, file_path, "
"git_commit_sha) VALUES ('art-1', 'brd', 'task-1', 'run-1', 'docs/brd.md', 'sha1')"
)
self.conn.commit()
result = task_query.get_task(self.conn, "task-1")
self.assertEqual(len(result["artifacts"]), 1)
self.assertEqual(result["artifacts"][0]["artifact_id"], "art-1")
def test_includes_bug_records(self):
self._insert_task("task-1")
self.conn.execute(
"INSERT INTO bug_record (bug_id, task_id, source, description, status) "
"VALUES ('bug-1', 'task-1', 'human', 'x', 'open')"
)
self.conn.commit()
result = task_query.get_task(self.conn, "task-1")
self.assertEqual(len(result["bug_records"]), 1)
self.assertEqual(result["bug_records"][0]["bug_id"], "bug-1")
def test_includes_work_graph_runs_with_parsed_graph_manifest_and_nodes(self):
self._insert_task("task-1")
manifest = {"nodes": [{"node_id": "n1", "artifact_type": "brd", "depends_on": []}], "edges": []}
self.conn.execute(
"INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
"VALUES ('run-1', 'task-1', 'planner', ?, 'running')",
(json.dumps(manifest),),
)
self.conn.execute(
"INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
"VALUES ('skill-1', 'planner', 'skills/skill-1/', 'sha', 'new', 'active')"
)
self.conn.execute(
"INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
"input_schema_ref, output_schema_ref, input_payload, output_payload, status, retry_count) "
"VALUES ('node-1', 'run-1', 'n1', 'planner', 'skill-1', 'in.json', 'out.json', '{}', "
"'{\"x\": 1}', 'done', 0)"
)
self.conn.commit()
result = task_query.get_task(self.conn, "task-1")
self.assertEqual(len(result["work_graph_runs"]), 1)
run = result["work_graph_runs"][0]
self.assertEqual(run["graph_manifest"], manifest)
self.assertEqual(len(run["nodes"]), 1)
self.assertEqual(run["nodes"][0]["node_id"], "n1")
self.assertEqual(run["nodes"][0]["status"], "done")
self.assertEqual(run["nodes"][0]["output_payload"], {"x": 1})
if __name__ == "__main__":
unittest.main()