test_artifact_write.py
4.75 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
"""U-014 (c_artifact_write) tests: real git, no mocking needed -- writing a
file and committing it is fast/free/local, same reasoning as U-013's
worktree tests.
Run: python3 -m unittest src.test_artifact_write -v
"""
import json
import os
import subprocess
import tempfile
import unittest
from src import artifact_write, schema
def _init_git_repo(path):
subprocess.run(["git", "init"], cwd=path, check=True, capture_output=True)
class ArtifactWriteTestBase(unittest.TestCase):
def setUp(self):
self.conn = schema.connect(":memory:")
schema.create_schema(self.conn)
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
_init_git_repo(self.tmp.name)
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', 'goal', 'goal', 'planning_brd_prd_sdd', 'planner')"
)
self.conn.execute(
"INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
"VALUES ('run-1', 'task-1', '{}', 'running')"
)
self.conn.commit()
def tearDown(self):
self.conn.close()
class HappyPathTest(ArtifactWriteTestBase):
def test_writes_file_commits_and_creates_row(self):
result = artifact_write.write_artifact(
self.conn, self.tmp.name, "brd", "task-1", "run-1",
".graphlemon/runs/task-1/brd.md", "# BRD\n\ncontent",
)
self.assertTrue(result["artifact_id"])
self.assertTrue(result["git_commit_sha"])
path = os.path.join(self.tmp.name, ".graphlemon/runs/task-1/brd.md")
with open(path) as f:
self.assertEqual(f.read(), "# BRD\n\ncontent")
log = subprocess.run(
["git", "log", "--oneline"], cwd=self.tmp.name, check=True, capture_output=True, text=True
)
self.assertIn("brd artifact for task task-1", log.stdout)
row = self.conn.execute(
"SELECT artifact_type, task_id, validation_status, depends_on_artifact_ids "
"FROM artifact WHERE artifact_id = ?",
(result["artifact_id"],),
).fetchone()
self.assertEqual(row[0], "brd")
self.assertEqual(row[1], "task-1")
self.assertEqual(row[2], "passed")
self.assertEqual(json.loads(row[3]), [])
def test_lineage_chain_brd_to_prd_to_sdd(self):
brd = artifact_write.write_artifact(
self.conn, self.tmp.name, "brd", "task-1", "run-1", ".graphlemon/runs/task-1/brd.md", "brd content"
)
prd = artifact_write.write_artifact(
self.conn, self.tmp.name, "prd", "task-1", "run-1", ".graphlemon/runs/task-1/prd.md", "prd content",
depends_on_artifact_ids=[brd["artifact_id"]],
)
sdd = artifact_write.write_artifact(
self.conn, self.tmp.name, "sdd", "task-1", "run-1", ".graphlemon/runs/task-1/sdd.md", "sdd content",
depends_on_artifact_ids=[prd["artifact_id"]],
)
row = self.conn.execute(
"SELECT depends_on_artifact_ids FROM artifact WHERE artifact_id = ?", (sdd["artifact_id"],)
).fetchone()
self.assertEqual(json.loads(row[0]), [prd["artifact_id"]])
class InvalidDependencyTest(ArtifactWriteTestBase):
def test_nonexistent_dependency_rejected_before_any_write(self):
with self.assertRaises(artifact_write.InvalidDependencyError):
artifact_write.write_artifact(
self.conn, self.tmp.name, "sdd", "task-1", "run-1",
".graphlemon/runs/task-1/sdd.md", "content",
depends_on_artifact_ids=["no-such-artifact"],
)
self.assertFalse(os.path.exists(os.path.join(self.tmp.name, ".graphlemon/runs/task-1/sdd.md")))
self.assertEqual(self.conn.execute("SELECT COUNT(*) FROM artifact").fetchone()[0], 0)
class GitCommitFailedTest(ArtifactWriteTestBase):
def test_committing_identical_content_twice_raises_git_commit_failed(self):
artifact_write.write_artifact(
self.conn, self.tmp.name, "brd", "task-1", "run-1", ".graphlemon/runs/task-1/brd.md", "same content"
)
# second write with byte-identical content -> `git commit` has nothing
# to commit -> exits nonzero -> GitCommitFailedError
with self.assertRaises(artifact_write.GitCommitFailedError):
artifact_write.write_artifact(
self.conn, self.tmp.name, "brd", "task-1", "run-1", ".graphlemon/runs/task-1/brd.md", "same content"
)
if __name__ == "__main__":
unittest.main()