"""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.assertIsNone(run["parent_run_id"]) self.assertIsNone(run["fork_of_node_exec_id"]) def test_forked_run_surfaces_parent_and_source_node(self): # D (2026-08-06): v_task_detail must be able to tell a fork branch # apart from the original run it descends from. self._insert_task("task-1") self.conn.execute( "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) " "VALUES ('run-orig', 'task-1', 'planner', '{}', 'done')" ) 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) " "VALUES ('node-1', 'run-orig', 'n1', 'planner', 'skill-1', 'in.json', 'out.json', '{}', '{}', 'done')" ) self.conn.execute( "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status, " "parent_run_id, fork_of_node_exec_id) " "VALUES ('run-fork', 'task-1', 'planner', '{}', 'done', 'run-orig', 'node-1')" ) self.conn.commit() result = task_query.get_task(self.conn, "task-1") by_run_id = {r["run_id"]: r for r in result["work_graph_runs"]} self.assertIsNone(by_run_id["run-orig"]["parent_run_id"]) self.assertEqual(by_run_id["run-fork"]["parent_run_id"], "run-orig") self.assertEqual(by_run_id["run-fork"]["fork_of_node_exec_id"], "node-1") def test_work_node_includes_quota_wait_fields_for_v_graph_visualizer(self): # U-026: v_graph_visualizer's waiting_for_quota state needs # quota_limit_type + how long it's been waiting (02-V §2.7) 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', 'coder', '{}', 'running')" ) self.conn.execute( "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) " "VALUES ('skill-1', 'coder', '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, status, quota_limit_type, " "quota_wait_started_at) " "VALUES ('node-1', 'run-1', 'n1', 'coder', 'skill-1', 'in.json', 'out.json', '{}', " "'waiting_for_quota', 'session', '2026-07-31 09:00:00')" ) self.conn.commit() result = task_query.get_task(self.conn, "task-1") node = result["work_graph_runs"][0]["nodes"][0] self.assertEqual(node["quota_limit_type"], "session") self.assertEqual(node["quota_wait_started_at"], "2026-07-31 09:00:00") class ListTasksSortTest(TaskQueryTestBase): def setUp(self): super().setUp() self._insert_task("task-a", goal_text="banana") self.conn.execute( "UPDATE task SET created_at = '2026-01-01 00:00:00', updated_at = '2026-01-01 00:00:00' " "WHERE task_id = 'task-a'" ) self._insert_task("task-b", goal_text="apple") self.conn.execute( "UPDATE task SET created_at = '2026-01-02 00:00:00', updated_at = '2026-01-02 00:00:00' " "WHERE task_id = 'task-b'" ) self._insert_task("task-c", goal_text="cherry") self.conn.execute( "UPDATE task SET created_at = '2026-01-03 00:00:00', updated_at = '2026-01-03 00:00:00' " "WHERE task_id = 'task-c'" ) self.conn.commit() def test_default_sort_is_updated_at_desc_unchanged(self): # existing v_task_board behavior must not shift under it result = task_query.list_tasks(self.conn, "proj-1") self.assertEqual([t["task_id"] for t in result], ["task-c", "task-b", "task-a"]) def test_sort_by_created_at_asc(self): result = task_query.list_tasks(self.conn, "proj-1", sort="created_at", order="asc") self.assertEqual([t["task_id"] for t in result], ["task-a", "task-b", "task-c"]) def test_sort_by_goal_text_asc(self): result = task_query.list_tasks(self.conn, "proj-1", sort="goal_text", order="asc") self.assertEqual([t["task_id"] for t in result], ["task-b", "task-a", "task-c"]) # apple, banana, cherry def test_invalid_sort_column_rejected(self): with self.assertRaises(task_query.InvalidParamError): task_query.list_tasks(self.conn, "proj-1", sort="task_id; DROP TABLE task") class UpdateTaskGoalTextTest(TaskQueryTestBase): def test_queued_task_can_be_edited(self): self._insert_task("task-1", status="queued", goal_text="old") result = task_query.update_task_goal_text(self.conn, "task-1", "new goal text") self.assertEqual(result["goal_text"], "new goal text") def test_nonexistent_task_raises_not_found(self): with self.assertRaises(task_query.TaskNotFoundError): task_query.update_task_goal_text(self.conn, "no-such-task", "x") def test_in_progress_task_cannot_be_edited(self): self._insert_task("task-1", status="coding") with self.assertRaises(task_query.TaskNotEditableError): task_query.update_task_goal_text(self.conn, "task-1", "new text") class DeleteTaskTest(TaskQueryTestBase): def test_queued_task_can_be_deleted(self): self._insert_task("task-1", status="queued") task_query.delete_task(self.conn, "task-1") self.assertEqual(self.conn.execute("SELECT COUNT(*) FROM task WHERE task_id = 'task-1'").fetchone()[0], 0) def test_nonexistent_task_raises_not_found(self): with self.assertRaises(task_query.TaskNotFoundError): task_query.delete_task(self.conn, "no-such-task") def test_done_task_cannot_be_deleted(self): self._insert_task("task-1", status="done") with self.assertRaises(task_query.TaskNotEditableError): task_query.delete_task(self.conn, "task-1") self.assertEqual(self.conn.execute("SELECT COUNT(*) FROM task WHERE task_id = 'task-1'").fetchone()[0], 1) def test_real_task_create_task_can_still_be_deleted(self): # 2026-08-03 regression: _insert_task (this file's own raw-INSERT # helper) never writes an execution_event row, so it never caught # the real bug -- task_create.create_task() always writes one # execution_event in the same transaction as the task itself (01-M's # own FK, no ON DELETE CASCADE), so even a never-touched queued task # already has one referencing it, and a real DELETE FROM task raised # sqlite3.IntegrityError until delete_task() also cleared it. from src import task_create result = task_create.create_task(self.conn, "proj-1", "a real goal") self.assertEqual( self.conn.execute( "SELECT COUNT(*) FROM execution_event WHERE task_id = ?", (result["task_id"],) ).fetchone()[0], 1, ) task_query.delete_task(self.conn, result["task_id"]) self.assertEqual( self.conn.execute("SELECT COUNT(*) FROM task WHERE task_id = ?", (result["task_id"],)).fetchone()[0], 0 ) self.assertEqual( self.conn.execute( "SELECT COUNT(*) FROM execution_event WHERE task_id = ?", (result["task_id"],) ).fetchone()[0], 0, ) if __name__ == "__main__": unittest.main()