"""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, []) class GetWorkGraphRunTest(WorkGraphRunQueryTestBase): """U-046: c_work_graph_run_get -- v_run_detail's single-run lookup.""" def test_returns_run_with_embedded_nodes(self): self._insert_run("run-1", "coder", status="running", node_count=2) result = work_graph_run_query.get_work_graph_run(self.conn, "run-1") self.assertEqual(result["run_id"], "run-1") self.assertEqual(result["task_id"], "task-1") self.assertEqual(result["org_node"], "coder") self.assertEqual(result["status"], "running") self.assertEqual(result["graph_manifest"], {}) self.assertEqual(len(result["nodes"]), 2) self.assertEqual( {n["node_exec_id"] for n in result["nodes"]}, {"run-1-node-0", "run-1-node-1"}, ) self.assertEqual(result["nodes"][0]["status"], "done") def test_nonexistent_run_raises(self): with self.assertRaises(work_graph_run_query.WorkGraphRunNotFoundError): work_graph_run_query.get_work_graph_run(self.conn, "no-such-run") def test_run_with_no_nodes_returns_empty_list(self): self._insert_run("run-1", "planner", node_count=0) result = work_graph_run_query.get_work_graph_run(self.conn, "run-1") self.assertEqual(result["nodes"], []) if __name__ == "__main__": unittest.main()