"""U-028 tests: c_artifact_list / c_artifact_get / lineage_chain. Real sqlite + real tempfile-backed artifact content, no mocking needed. Run: python3 -m unittest src.test_artifact_query -v """ import json import os import tempfile import unittest from src import artifact_query, schema class ArtifactQueryTestBase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) 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', ?, 1, 'active')", (self.tmp.name,), ) 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.execute( "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) " "VALUES ('run-1', 'task-1', '{}', 'done')" ) self.conn.commit() def tearDown(self): self.conn.close() def _insert_artifact(self, artifact_id, artifact_type, file_path, depends_on=None, content=None, task_id="task-1"): if content is not None: abs_path = os.path.join(self.tmp.name, file_path) os.makedirs(os.path.dirname(abs_path), exist_ok=True) with open(abs_path, "w") as f: f.write(content) self.conn.execute( "INSERT INTO artifact (artifact_id, artifact_type, task_id, produced_by_run_id, file_path, " "git_commit_sha, depends_on_artifact_ids) VALUES (?, ?, ?, 'run-1', ?, 'sha1', ?)", (artifact_id, artifact_type, task_id, file_path, json.dumps(depends_on) if depends_on else None), ) self.conn.commit() class ListArtifactsTest(ArtifactQueryTestBase): def test_lists_artifacts_for_project(self): self._insert_artifact("art-1", "brd", "docs/brd.md") self._insert_artifact("art-2", "sdd", "docs/sdd.md") result = artifact_query.list_artifacts(self.conn, "proj-1") self.assertEqual({a["artifact_id"] for a in result}, {"art-1", "art-2"}) 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_artifact("art-1", "brd", "docs/brd.md") self._insert_artifact("art-2", "brd", "docs/brd2.md", task_id="task-2") result = artifact_query.list_artifacts(self.conn, "proj-1") self.assertEqual([a["artifact_id"] for a in result], ["art-1"]) def test_task_id_filter(self): self.conn.execute( "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) " "VALUES ('task-2', 'proj-1', 'g', 'goal', 'coding', 'coder')" ) self.conn.commit() self._insert_artifact("art-1", "brd", "docs/brd.md", task_id="task-1") self._insert_artifact("art-2", "brd", "docs/brd2.md", task_id="task-2") result = artifact_query.list_artifacts(self.conn, "proj-1", task_id="task-2") self.assertEqual([a["artifact_id"] for a in result], ["art-2"]) def test_artifact_type_filter(self): self._insert_artifact("art-1", "brd", "docs/brd.md") self._insert_artifact("art-2", "sdd", "docs/sdd.md") result = artifact_query.list_artifacts(self.conn, "proj-1", artifact_type="sdd") self.assertEqual([a["artifact_id"] for a in result], ["art-2"]) def test_depends_on_artifact_ids_parsed_as_json(self): self._insert_artifact("art-1", "brd", "docs/brd.md") self._insert_artifact("art-2", "prd", "docs/prd.md", depends_on=["art-1"]) result = artifact_query.list_artifacts(self.conn, "proj-1", artifact_type="prd") self.assertEqual(result[0]["depends_on_artifact_ids"], ["art-1"]) def test_no_deps_returns_empty_list_not_null(self): self._insert_artifact("art-1", "brd", "docs/brd.md") result = artifact_query.list_artifacts(self.conn, "proj-1") self.assertEqual(result[0]["depends_on_artifact_ids"], []) class GetArtifactTest(ArtifactQueryTestBase): def test_nonexistent_raises(self): with self.assertRaises(artifact_query.ArtifactNotFoundError): artifact_query.get_artifact(self.conn, "no-such-artifact") def test_returns_content_read_from_disk(self): self._insert_artifact("art-1", "sdd", "docs/sdd.md", content="# hello\nworld") result = artifact_query.get_artifact(self.conn, "art-1") self.assertEqual(result["content"], "# hello\nworld") self.assertEqual(result["artifact_type"], "sdd") def test_missing_physical_file_returns_none_content_not_error(self): self._insert_artifact("art-1", "sdd", "docs/does-not-exist.md") result = artifact_query.get_artifact(self.conn, "art-1") self.assertIsNone(result["content"]) def test_includes_lineage(self): self._insert_artifact("art-1", "brd", "docs/brd.md") self._insert_artifact("art-2", "prd", "docs/prd.md", depends_on=["art-1"]) result = artifact_query.get_artifact(self.conn, "art-2") self.assertEqual( result["lineage"], [{"artifact_id": "art-1", "artifact_type": "brd"}, {"artifact_id": "art-2", "artifact_type": "prd"}], ) class LineageChainTest(ArtifactQueryTestBase): def test_single_artifact_no_deps(self): self._insert_artifact("art-1", "brd", "docs/brd.md") self.assertEqual(artifact_query.lineage_chain(self.conn, "art-1"), [{"artifact_id": "art-1", "artifact_type": "brd"}]) def test_three_level_chain_oldest_first(self): self._insert_artifact("art-1", "brd", "docs/brd.md") self._insert_artifact("art-2", "prd", "docs/prd.md", depends_on=["art-1"]) self._insert_artifact("art-3", "sdd", "docs/sdd.md", depends_on=["art-2"]) chain = artifact_query.lineage_chain(self.conn, "art-3") self.assertEqual([c["artifact_id"] for c in chain], ["art-1", "art-2", "art-3"]) def test_multiple_deps_follows_first_only(self): self._insert_artifact("art-1", "brd", "docs/brd.md") self._insert_artifact("art-1b", "benchmark", "docs/bench.md") self._insert_artifact("art-2", "prd", "docs/prd.md", depends_on=["art-1", "art-1b"]) chain = artifact_query.lineage_chain(self.conn, "art-2") self.assertEqual([c["artifact_id"] for c in chain], ["art-1", "art-2"]) def test_cycle_does_not_infinite_loop(self): # simulates externally-corrupted data (c_artifact_write itself # never allows this) -- a read path shouldn't hang on it self._insert_artifact("art-1", "brd", "docs/brd.md", depends_on=["art-2"]) self._insert_artifact("art-2", "prd", "docs/prd.md", depends_on=["art-1"]) chain = artifact_query.lineage_chain(self.conn, "art-1") self.assertLessEqual(len(chain), 2) if __name__ == "__main__": unittest.main()