test_work_graph_run_query.py
5.85 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
"""U-GRAPH-01 tests: c_work_graph_run_list (list_work_graph_runs). Real
sqlite fixtures, no mocking, same style as test_artifact_query.py.
Run: python3 -m unittest src.test_work_graph_run_query -v
"""
import unittest
from src import schema, work_graph_run_query
class WorkGraphRunQueryTestBase(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, assigned_org_node) "
"VALUES ('task-1', 'proj-1', 'g', 'goal', 'coding', 'coder')"
)
self.conn.commit()
def tearDown(self):
self.conn.close()
def _insert_run(self, run_id, org_node, status="done", started_at="2026-08-01T00:00:00",
completed_at=None, task_id="task-1", node_count=0):
self.conn.execute(
"INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status, "
"started_at, completed_at) VALUES (?, ?, ?, '{}', ?, ?, ?)",
(run_id, task_id, org_node, status, started_at, completed_at),
)
for i in range(node_count):
node_exec_id = "{}-node-{}".format(run_id, i)
self.conn.execute(
"INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
"VALUES (?, 'coder', 'skills/x.md', 'sha1', 'new', 'active')",
("skill-{}".format(node_exec_id),),
)
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, status) "
"VALUES (?, ?, ?, 'coder', ?, 'schema', 'schema', '{}', 'done')",
(node_exec_id, run_id, "node-{}".format(i), "skill-{}".format(node_exec_id)),
)
self.conn.commit()
class ListWorkGraphRunsTest(WorkGraphRunQueryTestBase):
def test_lists_runs_across_org_nodes_for_project(self):
self._insert_run("run-1", "planner", started_at="2026-08-01T00:00:00")
self._insert_run("run-2", "coder", started_at="2026-08-02T00:00:00")
self._insert_run("run-3", "validator", started_at="2026-08-03T00:00:00")
result = work_graph_run_query.list_work_graph_runs(self.conn, "proj-1")
self.assertEqual({r["run_id"] for r in result}, {"run-1", "run-2", "run-3"})
def test_sorted_started_at_desc(self):
self._insert_run("run-1", "planner", started_at="2026-08-01T00:00:00")
self._insert_run("run-2", "coder", started_at="2026-08-03T00:00:00")
self._insert_run("run-3", "validator", started_at="2026-08-02T00:00:00")
result = work_graph_run_query.list_work_graph_runs(self.conn, "proj-1")
self.assertEqual([r["run_id"] for r in result], ["run-2", "run-3", "run-1"])
def test_node_count_matches_actual_row_count(self):
self._insert_run("run-1", "coder", node_count=5)
result = work_graph_run_query.list_work_graph_runs(self.conn, "proj-1")
self.assertEqual(result[0]["node_count"], 5)
def test_other_project_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.execute(
"INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
"VALUES ('task-2', 'proj-2', 'g', 'goal', 'coding', 'coder')"
)
self.conn.commit()
self._insert_run("run-1", "planner", task_id="task-1")
self._insert_run("run-2", "coder", task_id="task-2")
result = work_graph_run_query.list_work_graph_runs(self.conn, "proj-1")
self.assertEqual([r["run_id"] for r in result], ["run-1"])
def test_org_node_filter(self):
self._insert_run("run-1", "planner", started_at="2026-08-01T00:00:00")
self._insert_run("run-2", "coder", started_at="2026-08-02T00:00:00")
self._insert_run("run-3", "coder", started_at="2026-08-03T00:00:00")
result = work_graph_run_query.list_work_graph_runs(self.conn, "proj-1", org_node="coder")
self.assertEqual({r["run_id"] for r in result}, {"run-2", "run-3"})
def test_limit_one_returns_latest(self):
self._insert_run("run-1", "planner", started_at="2026-08-01T00:00:00")
self._insert_run("run-2", "coder", started_at="2026-08-02T00:00:00")
result = work_graph_run_query.list_work_graph_runs(self.conn, "proj-1", limit=1)
self.assertEqual([r["run_id"] for r in result], ["run-2"])
def test_limit_over_100_raises(self):
with self.assertRaises(work_graph_run_query.InvalidParamError):
work_graph_run_query.list_work_graph_runs(self.conn, "proj-1", limit=101)
def test_limit_below_1_raises(self):
with self.assertRaises(work_graph_run_query.InvalidParamError):
work_graph_run_query.list_work_graph_runs(self.conn, "proj-1", limit=0)
def test_missing_project_id_raises(self):
with self.assertRaises(work_graph_run_query.InvalidParamError):
work_graph_run_query.list_work_graph_runs(self.conn, None)
def test_invalid_org_node_raises(self):
with self.assertRaises(work_graph_run_query.InvalidParamError):
work_graph_run_query.list_work_graph_runs(self.conn, "proj-1", org_node="not_a_node")
def test_nonexistent_project_returns_empty_list(self):
result = work_graph_run_query.list_work_graph_runs(self.conn, "no-such-project")
self.assertEqual(result, [])
if __name__ == "__main__":
unittest.main()