"""One end-to-end check that the spawn payload reaches a real cell as an import."""
from __future__ import annotations
import json
import tempfile
import unittest
from pathlib import Path
from peer import Peer
ROOT = Path(__file__).resolve().parents[3]
SKILLS = ROOT / "skills"
def skills_payload(directory: Path) -> Path:
"""Reproduce the host's payload shape from the real builtin bodies."""
skills = []
for skill in sorted(SKILLS.iterdir()):
body = skill / "SKILL.md"
if body.is_file():
skills.append({
"id": skill.name,
"origin": "builtin",
"path": str(body),
"digest": f"{skill.name:0<64}"[:64],
"text": body.read_text(encoding="utf-8"),
})
document = {"revision": "test-revision", "skills": skills, "refused": []}
path = directory / "skills.json"
path.write_text(json.dumps(document), encoding="utf-8")
return path
class WorkbenchSkillImportTests(unittest.IsolatedAsyncioTestCase):
"""The worker reads what the host wrote, and a cell can import it."""
async def test_a_real_worker_imports_a_skill_body_by_id(self):
with tempfile.TemporaryDirectory(prefix="klbr-skills-") as name:
payload = skills_payload(Path(name))
async with Peer("workbench", skills=payload) as peer:
operation = await peer.begin(
"import klbr.skills as skills\n"
"print(sorted(skills.index) == ['behavior', 'diagnostics', 'maintenance',"
" 'orientation', 'peer', 'remote', 'skills', 'workspace'])\n"
"print(skills.workspace.__doc__.splitlines()[0])\n"
"help(skills.workspace)\n"
)
outcome = await peer.finish(operation)
self.assertEqual(outcome["status"], "ok", outcome)
printed = "".join(record["text"] for record in outcome["output"])
self.assertIn("True", printed)
self.assertIn("# workspace", printed)
async def test_a_worker_without_a_payload_says_so_instead_of_inventing_skills(self):
async with Peer("workbench") as peer:
operation = await peer.begin(
"import klbr.skills as skills\n"
"print(skills.index, skills.revision)\n"
"try:\n"
" skills.workspace\n"
"except AttributeError as error:\n"
" print('refused:', error)\n"
)
outcome = await peer.finish(operation)
self.assertEqual(outcome["status"], "ok", outcome)
printed = "".join(record["text"] for record in outcome["output"])
self.assertIn("spawned with no skill set", printed)
self.assertIn("no-skills", printed)
if __name__ == "__main__":
unittest.main()