test_artifact_query.py
7.33 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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
"""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()