"""Tests for the skill payload: what a generation can import, and why it cannot import more."""
from __future__ import annotations
import json
import tempfile
import unittest
from pathlib import Path
from klbr_runtime import skill_index
def payload(bodies, refused=(), revision="rev-1"):
return {
"revision": revision,
"skills": [
{
"id": skill_id,
"origin": origin,
"path": path,
"digest": digest,
"text": text,
}
for skill_id, origin, path, digest, text in bodies
],
"refused": [{"id": entry_id, "reason": reason} for entry_id, reason in refused],
}
class SkillPayloadTests(unittest.TestCase):
"""The payload is the whole contract between host and worker."""
def setUp(self):
self.temporary = tempfile.TemporaryDirectory(prefix="klbr-skill-test-")
self.path = Path(self.temporary.name) / "skills.json"
def tearDown(self):
self.temporary.cleanup()
# Module installation is global by nature; put the index back to empty so one
# test's ids cannot satisfy another test's import.
skill_index.install(Path(self.temporary.name) / "absent.json")
def write(self, document):
self.path.write_text(json.dumps(document), encoding="utf-8")
return self.path
def test_a_missing_payload_is_no_skills_rather_than_a_failure(self):
skill_index.install(self.path)
self.assertFalse(skill_index.loaded())
self.assertEqual(skill_index.bodies(), ())
self.assertEqual(skill_index.revision(), "no-skills")
def test_bodies_carry_their_digest_and_refusals_their_reason(self):
body = "# workspace\n\nOnce you have a workspace you can find code.\n"
document = payload(
[("workspace", "builtin", "/roots/skills/workspace/SKILL.md", "a" * 64, body)],
refused=[("Broken-Name", "skill id must be lower-case")],
revision="rev-9",
)
document["roots"] = {"builtin": "/roots/skills", "managed": "/data/skills"}
self.write(document)
skill_index.install(self.path)
self.assertEqual(skill_index.revision(), "rev-9")
self.assertEqual(
skill_index.roots(), {"builtin": "/roots/skills", "managed": "/data/skills"}
)
loaded = skill_index.bodies()
self.assertEqual(len(loaded), 1)
self.assertEqual(loaded[0].id, "workspace")
self.assertEqual(loaded[0].digest, "a" * 64)
self.assertEqual(loaded[0].text, body)
self.assertEqual(skill_index.refused()[0].id, "Broken-Name")
def test_a_malformed_entry_is_skipped_rather_than_half_read(self):
document = payload([("peer", "managed", "/roots/peer/SKILL.md", "b" * 64, "# peer\n")])
document["skills"].append({"id": "truncated"})
self.write(document)
skill_index.install(self.path)
self.assertEqual([body.id for body in skill_index.bodies()], ["peer"])
def test_a_malformed_payload_replaces_the_set_rather_than_keeping_stale_bodies(self):
self.write(payload([("peer", "managed", "/roots/peer/SKILL.md", "b" * 64, "# peer\n")]))
skill_index.install(self.path)
self.assertEqual([body.id for body in skill_index.bodies()], ["peer"])
self.path.write_text("[1, 2, 3]", encoding="utf-8")
skill_index.install(self.path)
self.assertFalse(skill_index.loaded())
self.assertEqual(skill_index.bodies(), ())
self.assertEqual(skill_index.revision(), "no-skills")
if __name__ == "__main__":
unittest.main()