"""Standalone tests: imports only the registry, never production bot modules.""" import dataclasses import importlib.util import os from pathlib import Path import sys import tempfile import unittest from concurrent.futures import ThreadPoolExecutor import yaml _MODULE_PATH = Path(__file__).resolve().parents[1] / "bot-utility" / "skill_registry.py" _SPEC = importlib.util.spec_from_file_location("_isolated_skill_registry", _MODULE_PATH) skills = importlib.util.module_from_spec(_SPEC) sys.modules[_SPEC.name] = skills _SPEC.loader.exec_module(skills) class SkillRegistryTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.base = Path(self.tmp.name) self.root = self.base / "skills" self.root.mkdir() self.config = self.base / "skills_config.yml" self.registry = skills.SkillRegistry(self.root, self.config) def configure(self, **kwargs): config = {"enabled": True, "allowlist": [], "always": [], "scenes": {}} config.update(kwargs) self.config.write_text(yaml.safe_dump(config), encoding="utf-8") def skill(self, name="alpha", body="Pinned instructions.\n", references=None, extra=""): directory = self.root / name directory.mkdir(exist_ok=True) (directory / "SKILL.md").write_text( f"---\nname: {name}\ndescription: Description for {name}\n{extra}---\n{body}", encoding="utf-8") for path, content in (references or {}).items(): target = directory / "references" / path target.parent.mkdir(parents=True, exist_ok=True) target.write_text(content, encoding="utf-8") return directory def error(self, code, fn, *args): with self.assertRaises(skills.SkillError) as caught: fn(*args) self.assertEqual(caught.exception.code, code) self.assertNotIn(str(self.base), str(caught.exception)) def test_missing_empty_and_default_config(self): for text in (None, "", "# empty\n", "{}\n"): if text is not None: self.config.write_text(text, encoding="utf-8") snapshot = self.registry.snapshot() self.assertTrue(snapshot.enabled) self.assertEqual(dict(snapshot.records), {}) self.assertEqual(snapshot.catalog_text(), "") self.assertEqual(snapshot.bound_names("private_chat"), ()) self.assertEqual(snapshot.limits["max_tokens"], 12000) self.assertEqual(skills.registry.root, _MODULE_PATH.parent / "skills") self.assertEqual(skills.registry.config_path, _MODULE_PATH.parent / "skills_config.yml") def test_catalog_allowlist_bindings_and_metadata(self): self.skill("alpha", references={"nested/info.md": "Secret ref"}, extra="metadata:\n author: Tester\nallowed-tools: Bash Read\n") self.skill("beta") self.skill("excluded") self.configure(allowlist=["beta", "alpha"], always=["alpha"], scenes={"notification": ["beta", "alpha"], "reminder": ["alpha"], "private_chat": ["beta"]}) snapshot = self.registry.snapshot() self.assertEqual(tuple(snapshot.records), ("alpha", "beta")) self.assertEqual(snapshot.catalog_text(), "alpha: Description for alpha\nbeta: Description for beta") self.assertNotIn("Secret ref", snapshot.catalog_text()) self.assertEqual(snapshot.bound_names("private_chat"), ("alpha", "beta")) self.assertEqual(snapshot.bound_names("reminder"), ("alpha", "beta")) self.assertEqual(snapshot.scenes["reminder"], ("beta", "alpha")) self.assertEqual(snapshot.records["alpha"].allowed_tools, ("Bash", "Read")) self.assertEqual(snapshot.records["alpha"].metadata["author"], "Tester") self.assertEqual(snapshot.read("alpha", "references/nested/info.md"), "Secret ref") self.error("unknown_skill", snapshot.load, "excluded") def test_snapshot_pinning_immutability_and_hash(self): directory = self.skill(references={"info.txt": "old"}) self.configure(allowlist=["alpha"]) first = self.registry.snapshot() with self.assertRaises(TypeError): first.records["new"] = first.records["alpha"] with self.assertRaises(TypeError): first.records["alpha"].references["references/info.txt"] = "new" with self.assertRaises(TypeError): first.limits["max_tokens"] = 3 with self.assertRaises(dataclasses.FrozenInstanceError): first.records["alpha"].body = "new" os.utime(directory / "SKILL.md", (1, 1)) self.assertIs(first, self.registry.snapshot()) other = skills.SkillRegistry(self.root, self.config).snapshot() self.assertEqual(first.fingerprint, other.fingerprint) self.skill(body="New body", references={"info.txt": "new"}) second = self.registry.snapshot() self.assertNotEqual(first.fingerprint, second.fingerprint) self.assertEqual(first.load("alpha"), "Pinned instructions.\n") self.assertEqual(first.read("alpha", "references/info.txt"), "old") self.assertEqual(second.load("alpha"), "New body") self.assertEqual(second.read("alpha", "references/info.txt"), "new") self.configure(allowlist=["alpha"], limits={"max_tokens": 100}) self.assertNotEqual(second.fingerprint, self.registry.snapshot().fingerprint) def test_invalid_config_keeps_last_good_and_disable_wins(self): self.skill() self.configure(allowlist=["alpha"], always=["alpha"]) good = self.registry.snapshot() for text in ("allowlist: [", "enabled: true\nenabled: false\n", "[]", "enabled: yesplease", "allowlist: [alpha, alpha]", "allowlist: [../alpha]", "always: [alpha]", "scenes: {unknown: []}", "limits: {max_rounds: true}", "allowlist: &a [alpha]\nalways: *a", "!!python/object:os.system {}", "unexpected: value"): with self.subTest(text=text): self.config.write_text(text, encoding="utf-8") self.assertIs(self.registry.snapshot(), good) self.config.write_text("enabled: false\nalways: [missing]\nlimits: broken\n", encoding="utf-8") disabled = self.registry.snapshot() self.assertFalse(disabled.enabled) self.assertEqual(disabled.catalog_text(), "") self.assertEqual(disabled.bound_names("help"), ()) self.assertFalse(disabled.records) self.error("disabled", disabled.load, "alpha") self.error("disabled", disabled.read, "alpha", "references/info.md") self.configure(allowlist=["alpha"]) self.assertTrue(self.registry.snapshot().enabled) def test_invalid_first_config_is_fail_closed(self): self.config.write_text("allowlist: [", encoding="utf-8") self.assertFalse(self.registry.snapshot().enabled) def test_limit_clamping_and_types(self): self.configure(limits={"body_bytes": -4, "reference_bytes": 10**9, "max_skills": 0, "max_rounds": 10**9, "max_tokens": -10}) result = self.registry.snapshot() self.assertEqual(dict(result.limits), {"body_bytes": 1, "reference_bytes": 65536, "max_skills": 1, "max_rounds": 8, "max_tokens": 1}) for limits in ({"max_tokens": "100"}, {"unknown": 1}, [], None): self.configure(limits=limits) self.assertIs(self.registry.snapshot(), result) def test_invalid_skills_quarantined_independently(self): self.skill("valid") bad = self.skill("invalid") self.configure(allowlist=["valid", "invalid"]) for document in ("no frontmatter", "---\nname: invalid\ndescription: [a]\n---\nbody", "---\nname: valid\ndescription: duplicate name\n---\nbody", "---\nname: invalid\nname: invalid\ndescription: desc\n---\nbody", "---\nname: invalid\ndescription: ''\n---\nbody", "---\nname: invalid\ndescription: " + "x" * 1025 + "\n---\nbody", "---\nname: invalid\ndescription: hi\nmetadata: {x: []}\n---\nbody"): with self.subTest(document=document): (bad / "SKILL.md").write_text(document, encoding="utf-8") with self.assertLogs(skills.__name__, level="WARNING") as logs: snapshot = self.registry.snapshot() self.assertEqual(tuple(snapshot.records), ("valid",)) self.assertTrue(any("quarantined" in line for line in logs.output)) self.assertTrue(all(str(self.base) not in line for line in logs.output)) def test_bound_missing_or_quarantined_record_retains_last_good(self): directory = self.skill() self.configure(allowlist=["alpha"], always=["alpha"]) good = self.registry.snapshot() (directory / "SKILL.md").write_text("broken", encoding="utf-8") self.assertIs(self.registry.snapshot(), good) self.configure(allowlist=["missing"], scenes={"help": ["missing"]}) self.assertIs(self.registry.snapshot(), good) def test_name_validation(self): for name in ("Upper", "two--parts", "-first", "last-", "with_space", "x" * 65, "../escape"): self.configure(allowlist=[name]) self.assertFalse(self.registry.snapshot().enabled) self.skill("abc-123") self.configure(allowlist=["abc-123"]) self.assertIn("abc-123", self.registry.snapshot().records) def test_safe_reference_errors_and_no_script_exposure(self): self.skill(references={"note.md": "note", "script.py": "raise RuntimeError()"}) self.configure(allowlist=["alpha"]) snapshot = self.registry.snapshot() self.assertEqual(tuple(snapshot.records["alpha"].references), ("references/note.md",)) for path in ("../SKILL.md", "/etc/passwd", "references/../SKILL.md", "references//note.md", "references/./note.md", "references\\note.md", "references/script.py", "note.md", "references/\x00.md", "references/dir/", 3): self.error("invalid_path", snapshot.read, "alpha", path) self.error("unknown_reference", snapshot.read, "alpha", "references/missing.md") self.error("unknown_skill", snapshot.load, "../../secret") def test_binary_oversized_body_and_references(self): directory = self.skill() self.configure(allowlist=["alpha"], limits={"body_bytes": 5, "reference_bytes": 5}) self.assertNotIn("alpha", self.registry.snapshot().records) self.skill(body="短文") # six UTF-8 bytes, not two self.assertNotIn("alpha", self.registry.snapshot().records) self.skill(body="ok", references={"big.txt": "123456"}) self.assertNotIn("alpha", self.registry.snapshot().records) ref = directory / "references" / "big.txt" for data in (b"\xff", b"\x00", b"hi\x01", b"\x7f"): ref.write_bytes(data) self.assertNotIn("alpha", self.registry.snapshot().records) ref.write_text("ok", encoding="utf-8") self.assertIn("alpha", self.registry.snapshot().records) (directory / "SKILL.md").write_bytes(b"\x00") self.assertNotIn("alpha", self.registry.snapshot().records) def test_symlinks_and_outside_config_are_rejected(self): outside = self.base / "outside" outside.mkdir() (outside / "SKILL.md").write_text("---\nname: escape\ndescription: desc\n---\nbody") (self.root / "escape").symlink_to(outside, target_is_directory=True) self.configure(allowlist=["escape"]) self.assertFalse(self.registry.snapshot().records) directory = self.skill() refs = directory / "references" refs.mkdir() (refs / "cycle").symlink_to(refs, target_is_directory=True) self.configure(allowlist=["alpha"]) self.assertFalse(self.registry.snapshot().records) (refs / "cycle").unlink() (refs / "outside.md").symlink_to(outside / "SKILL.md") self.assertFalse(self.registry.snapshot().records) (refs / "outside.md").unlink() good = self.registry.snapshot() original = self.base / "original.yml" original.write_text("enabled: false") self.config.unlink() self.config.symlink_to(original) self.assertIs(self.registry.snapshot(), good) # A config outside root.parent is rejected even if it is a regular file. nested = outside / "nested" / "skills" nested.mkdir(parents=True) other = skills.SkillRegistry(nested, original) self.assertFalse(other.snapshot().enabled) def test_symlinked_skill_document_and_config_parent(self): directory = self.skill() doc = directory / "SKILL.md" target = self.base / "document.md" target.write_text(doc.read_text(), encoding="utf-8") doc.unlink() doc.symlink_to(target) self.configure(allowlist=["alpha"]) self.assertFalse(self.registry.snapshot().records) link = self.base / "link" link.symlink_to(self.base, target_is_directory=True) registry = skills.SkillRegistry(link / "skills", link / "skills_config.yml") self.assertFalse(registry.snapshot().enabled) def test_reference_only_change_changes_hash(self): directory = self.skill(references={"info.md": "before"}) self.configure(allowlist=["alpha"]) before = self.registry.snapshot() (directory / "references" / "info.md").write_text("after") after = self.registry.snapshot() self.assertNotEqual(before.fingerprint, after.fingerprint) self.assertEqual(before.read("alpha", "references/info.md"), "before") def test_config_order_and_file_timestamps_do_not_affect_hash(self): self.skill("alpha") self.skill("beta") self.configure(allowlist=["alpha", "beta"]) before = self.registry.snapshot() self.configure(allowlist=["beta", "alpha"]) self.assertIs(before, self.registry.snapshot()) def test_concurrent_snapshot_reads(self): self.skill() self.configure(allowlist=["alpha"]) with ThreadPoolExecutor(max_workers=6) as pool: snapshots = list(pool.map(lambda _: self.registry.snapshot(), range(30))) self.assertTrue(all(s is snapshots[0] for s in snapshots)) self.assertEqual(snapshots[0].load("alpha"), "Pinned instructions.\n") if __name__ == "__main__": unittest.main()