"""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()