"""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()