test_skill_update.py
5.1 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
"""U-018 tests: c_skill_update / c_skill_rollback, real git in a throwaway
temp repo (never graphlemon's own -- see skill_update.py's module docstring
on why repo_path has no default). Sqlite + local git are fast/free, no
mocking needed.
Run: python3 -m unittest src.test_skill_update -v
"""
import subprocess
import tempfile
import unittest
from src import bug_record, schema, skill_update
def _init_git_repo(path):
subprocess.run(["git", "init"], cwd=path, check=True, capture_output=True)
class SkillUpdateTestBase(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
_init_git_repo(self.tmp.name)
self.conn = schema.connect(":memory:")
schema.create_schema(self.conn)
self.conn.execute(
"INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
"VALUES ('write-brd', 'planner', 'skills/write-brd', 'unknown', 'migrated_from_hermes', 'active')"
)
self.conn.execute(
"INSERT INTO bug_record (bug_id, source, description, root_cause_category, root_cause_detail, "
"suggested_route, status) VALUES ('bug-1', 'validator', 'skill produces bad BRDs', 'skill_defect', "
"'systematic gap', 'fast_lane', 'routed')"
)
self.conn.commit()
skill_update._write_and_commit(self.tmp.name, "skills/write-brd/SKILL.md", "v1 content", "seed skill")
def tearDown(self):
self.conn.close()
class UpdateSkillTest(SkillUpdateTestBase):
def test_happy_path_creates_iteration_and_commits(self):
result = skill_update.update_skill(
self.conn, "write-brd", "bug-1", "v2 content", "fix the systematic gap", self.tmp.name
)
self.assertTrue(result["iteration_id"])
self.assertTrue(result["after_git_commit_sha"])
with open(self.tmp.name + "/skills/write-brd/SKILL.md") as f:
self.assertEqual(f.read(), "v2 content")
row = self.conn.execute(
"SELECT skill_id, triggered_by_bug_id, change_summary FROM skill_iteration WHERE iteration_id = ?",
(result["iteration_id"],),
).fetchone()
self.assertEqual(row, ("write-brd", "bug-1", "fix the systematic gap"))
current_sha = self.conn.execute("SELECT current_git_commit_sha FROM skill WHERE skill_id = 'write-brd'").fetchone()[0]
self.assertEqual(current_sha, result["after_git_commit_sha"])
def test_fills_in_bug_records_skill_iteration_suggestion_id(self):
# this is the loop U-015 deferred: skill_defect bug_records start
# with skill_iteration_suggestion_id=null, filled in once the
# iteration this bug triggered actually exists
result = skill_update.update_skill(
self.conn, "write-brd", "bug-1", "v2 content", "fix it", self.tmp.name
)
suggestion_id = self.conn.execute(
"SELECT skill_iteration_suggestion_id FROM bug_record WHERE bug_id = 'bug-1'"
).fetchone()[0]
self.assertEqual(suggestion_id, result["iteration_id"])
def test_missing_trigger_ref_rejected(self):
with self.assertRaises(skill_update.MissingTriggerRefError):
skill_update.update_skill(self.conn, "write-brd", None, "v2 content", "x", self.tmp.name)
count = self.conn.execute("SELECT COUNT(*) FROM skill_iteration").fetchone()[0]
self.assertEqual(count, 0)
def test_each_iteration_produces_its_own_commit(self):
skill_update.update_skill(self.conn, "write-brd", "bug-1", "v2", "first fix", self.tmp.name)
skill_update.update_skill(self.conn, "write-brd", "bug-1", "v3", "second fix", self.tmp.name)
log = subprocess.run(["git", "log", "--oneline"], cwd=self.tmp.name, capture_output=True, text=True)
self.assertEqual(len(log.stdout.strip().splitlines()), 3) # seed + 2 iterations
class RollbackSkillTest(SkillUpdateTestBase):
def test_rollback_restores_prior_content_as_new_iteration(self):
first = skill_update.update_skill(self.conn, "write-brd", "bug-1", "v2 broken", "bad fix", self.tmp.name)
result = skill_update.rollback_skill(self.conn, first["iteration_id"], self.tmp.name)
self.assertTrue(result["new_iteration_id"])
self.assertNotEqual(result["new_iteration_id"], first["iteration_id"])
with open(self.tmp.name + "/skills/write-brd/SKILL.md") as f:
self.assertEqual(f.read(), "v1 content")
row = self.conn.execute(
"SELECT rollback_of_iteration_id FROM skill_iteration WHERE iteration_id = ?",
(result["new_iteration_id"],),
).fetchone()
self.assertEqual(row[0], first["iteration_id"])
count = self.conn.execute("SELECT COUNT(*) FROM skill_iteration").fetchone()[0]
self.assertEqual(count, 2) # rollback is a new record, not a history rewrite
def test_nonexistent_iteration_raises_not_found(self):
with self.assertRaises(skill_update.IterationNotFoundError):
skill_update.rollback_skill(self.conn, "no-such-iteration", self.tmp.name)
if __name__ == "__main__":
unittest.main()