test_skill_query.py
8.36 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
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
"""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 os
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 GetSkillContentTest(SkillQueryTestBase):
"""Bug fix: the skill detail page showed only skill_iteration history
and nothing for the real migrated SKILL.md content sitting on disk.
Uses the same tmp git repo as the rest of this file's tests -- these
file reads don't need git at all, but keeping repo_path consistent with
the other tests here avoids inventing a second fixture shape."""
def _write_skill_file(self, skill_id, content, extra_files=None):
skill_dir = os.path.join(self.tmp.name, "skills", skill_id)
os.makedirs(skill_dir, exist_ok=True)
with open(os.path.join(skill_dir, "SKILL.md"), "w", encoding="utf-8") as f:
f.write(content)
for rel in extra_files or []:
full = os.path.join(skill_dir, rel)
os.makedirs(os.path.dirname(full), exist_ok=True)
with open(full, "w", encoding="utf-8") as f:
f.write("support file content")
def test_nonexistent_skill_raises_not_found(self):
with self.assertRaises(skill_query.SkillNotFoundError):
skill_query.get_skill_content(self.conn, "no-such-skill", repo_path=self.tmp.name)
def test_returns_real_file_content(self):
self._insert_skill("s1", "coder")
self._write_skill_file("s1", "real skill content\n")
result = skill_query.get_skill_content(self.conn, "s1", repo_path=self.tmp.name)
self.assertEqual(result["skill_id"], "s1")
self.assertEqual(result["content"], "real skill content\n")
self.assertEqual(result["support_files"], [])
def test_lists_support_files_alongside_content(self):
self._insert_skill("s1", "coder")
self._write_skill_file(
"s1", "content\n",
extra_files=["references/notes.md", "templates/t.md", "scripts/run.sh"],
)
result = skill_query.get_skill_content(self.conn, "s1", repo_path=self.tmp.name)
self.assertEqual(
set(result["support_files"]),
{"references/notes.md", "templates/t.md", "scripts/run.sh"},
)
def test_row_exists_but_file_missing_raises_content_missing(self):
self._insert_skill("s1", "coder")
with self.assertRaises(skill_query.SkillContentMissingError):
skill_query.get_skill_content(self.conn, "s1", repo_path=self.tmp.name)
def test_file_path_already_ending_in_md_is_read_directly(self):
self._insert_skill("s1", "coder")
self.conn.execute("UPDATE skill SET file_path = ? WHERE skill_id = ?", ("skills/s1/SKILL.md", "s1"))
self.conn.commit()
self._write_skill_file("s1", "direct path content\n")
result = skill_query.get_skill_content(self.conn, "s1", repo_path=self.tmp.name)
self.assertEqual(result["content"], "direct path content\n")
def test_path_traversal_in_stored_file_path_is_rejected(self):
"""Defense-in-depth: even if skill.file_path were somehow poisoned
with a traversal sequence (not reachable via the HTTP skill_id
param today, since that only drives a parameterized WHERE clause),
the resolved path must stay confined to skills/."""
self._insert_skill("s1", "coder")
self.conn.execute(
"UPDATE skill SET file_path = ? WHERE skill_id = ?", ("../../../../etc/passwd", "s1"),
)
self.conn.commit()
with self.assertRaises(skill_query.SkillContentMissingError):
skill_query.get_skill_content(self.conn, "s1", repo_path=self.tmp.name)
def test_path_traversal_via_dotdot_segments_stays_confined(self):
self._insert_skill("s1", "coder")
self.conn.execute(
"UPDATE skill SET file_path = ? WHERE skill_id = ?",
("skills/../../outside/SKILL.md", "s1"),
)
self.conn.commit()
with self.assertRaises(skill_query.SkillContentMissingError):
skill_query.get_skill_content(self.conn, "s1", repo_path=self.tmp.name)
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()