test_skill_query.py
4.46 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
"""U-029 tests: c_skill_diff + list/get read helpers for v_skill_library /
v_skill_audit_log. Real sqlite + real git in a throwaway temp repo (never
graphlemon's own -- same reasoning as test_skill_update.py).
Run: python3 -m unittest src.test_skill_query -v
"""
import subprocess
import tempfile
import unittest
from src import schema, skill_query, skill_update
def _init_git_repo(path):
subprocess.run(["git", "init"], cwd=path, check=True, capture_output=True)
class SkillQueryTestBase(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.commit()
def tearDown(self):
self.conn.close()
def _insert_skill(self, skill_id, owner_org_node, source="new", migrated_from_profile=None):
self.conn.execute(
"INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, "
"migrated_from_profile, status) VALUES (?, ?, ?, 'unknown', ?, ?, 'active')",
(skill_id, owner_org_node, "skills/" + skill_id, source, migrated_from_profile),
)
self.conn.commit()
class ListSkillsTest(SkillQueryTestBase):
def test_lists_all_skills(self):
self._insert_skill("s1", "planner")
self._insert_skill("s2", "coder")
result = skill_query.list_skills(self.conn)
self.assertEqual({s["skill_id"] for s in result}, {"s1", "s2"})
def test_includes_owner_org_node_and_source(self):
self._insert_skill("s1", "validator", source="migrated_from_hermes", migrated_from_profile="qa")
result = skill_query.list_skills(self.conn)
self.assertEqual(result[0]["owner_org_node"], "validator")
self.assertEqual(result[0]["source"], "migrated_from_hermes")
self.assertEqual(result[0]["migrated_from_profile"], "qa")
def test_empty_returns_empty_list(self):
self.assertEqual(skill_query.list_skills(self.conn), [])
class GetSkillAuditLogTest(SkillQueryTestBase):
def test_nonexistent_skill_raises(self):
with self.assertRaises(skill_query.SkillNotFoundError):
skill_query.get_skill_audit_log(self.conn, "no-such-skill")
def test_skill_with_no_iterations_returns_empty_list(self):
self._insert_skill("s1", "planner")
self.assertEqual(skill_query.get_skill_audit_log(self.conn, "s1"), [])
def test_returns_iterations_oldest_first(self):
self._insert_skill("s1", "coder")
self.conn.execute(
"INSERT INTO bug_record (bug_id, source, description, root_cause_category, root_cause_detail, "
"suggested_route, status) VALUES ('bug-1', 'validator', 'x', 'skill_defect', 'd', 'fast_lane', 'routed')"
)
self.conn.execute(
"INSERT INTO skill_iteration (iteration_id, skill_id, triggered_by_bug_id, before_git_commit_sha, "
"after_git_commit_sha, change_summary, applied_at) VALUES "
"('it-1', 's1', 'bug-1', 'sha0', 'sha1', 'fix A', '2026-07-01 00:00:00')"
)
self.conn.execute(
"INSERT INTO skill_iteration (iteration_id, skill_id, triggered_by_bug_id, before_git_commit_sha, "
"after_git_commit_sha, change_summary, applied_at) VALUES "
"('it-2', 's1', 'bug-1', 'sha1', 'sha2', 'fix B', '2026-07-02 00:00:00')"
)
self.conn.commit()
result = skill_query.get_skill_audit_log(self.conn, "s1")
self.assertEqual([r["iteration_id"] for r in result], ["it-1", "it-2"])
self.assertEqual(result[0]["change_summary"], "fix A")
class GetDiffTest(SkillQueryTestBase):
def test_diff_between_two_commits_shows_content_change(self):
before_sha = skill_update._write_and_commit(self.tmp.name, "skills/s1/SKILL.md", "v1 content\n", "seed")
after_sha = skill_update._write_and_commit(self.tmp.name, "skills/s1/SKILL.md", "v2 content\n", "update")
diff_text = skill_query.get_diff(before_sha, after_sha, repo_path=self.tmp.name)
self.assertIn("-v1 content", diff_text)
self.assertIn("+v2 content", diff_text)
def test_identical_commits_produce_empty_diff(self):
sha = skill_update._write_and_commit(self.tmp.name, "skills/s1/SKILL.md", "same\n", "seed")
diff_text = skill_query.get_diff(sha, sha, repo_path=self.tmp.name)
self.assertEqual(diff_text, "")
if __name__ == "__main__":
unittest.main()