220 lines
7.7 KiB
Python
220 lines
7.7 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from skiff import catalog
|
|
from skiff.catalog import catalog_checkout_path, catalog_repo
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def write_skill(root: Path, name: str) -> Path:
|
|
skill = root / name
|
|
skill.mkdir(parents=True)
|
|
skill.joinpath("SKILL.md").write_text(
|
|
f"---\nname: {name}\ndescription: >-\n"
|
|
f" Test {name} skill.\n---\n\n# {name}\n",
|
|
encoding="utf-8",
|
|
)
|
|
return skill
|
|
|
|
|
|
class CatalogCollectionTests(unittest.TestCase):
|
|
def test_load_catalog_falls_back_to_legacy_registry_file(self) -> None:
|
|
legacy = self.home / "legacy-registry.yaml"
|
|
legacy.write_text("legacy:\n repo: https://example.test/legacy.git\n")
|
|
with (
|
|
patch.object(catalog, "CATALOG_FILE", self.home / "missing-catalog.yaml"),
|
|
patch.object(catalog, "LEGACY_REGISTRY_FILE", legacy),
|
|
):
|
|
self.assertIn("legacy", catalog.load_catalog())
|
|
|
|
def setUp(self) -> None:
|
|
self.temp_dir = tempfile.TemporaryDirectory()
|
|
self.home = Path(self.temp_dir.name)
|
|
self.skills_home = self.home / ".skills"
|
|
(self.skills_home / "skills").mkdir(parents=True)
|
|
|
|
self.upstream = self.home / "upstream"
|
|
self.first = write_skill(self.upstream / "skills", "first-skill")
|
|
self.second = write_skill(self.upstream / "skills", "second-skill")
|
|
subprocess.run(
|
|
["git", "init", "-b", "main", str(self.upstream)],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
subprocess.run(["git", "-C", str(self.upstream), "add", "."], check=True)
|
|
subprocess.run(
|
|
[
|
|
"git",
|
|
"-C",
|
|
str(self.upstream),
|
|
"-c",
|
|
"user.name=Test",
|
|
"-c",
|
|
"user.email=test@example.com",
|
|
"commit",
|
|
"-m",
|
|
"initial",
|
|
],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
self.skills_home.joinpath("catalog.yaml").write_text(
|
|
"test-pack:\n"
|
|
f" repo: {self.upstream}\n"
|
|
" ref: main\n"
|
|
" path: skills\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def tearDown(self) -> None:
|
|
self.temp_dir.cleanup()
|
|
|
|
def run_skiff(self, *args: str) -> subprocess.CompletedProcess[str]:
|
|
env = os.environ.copy()
|
|
env["HOME"] = str(self.home)
|
|
env["PYTHONPATH"] = str(REPO_ROOT)
|
|
return subprocess.run(
|
|
[sys.executable, "-m", "skiff", *args],
|
|
cwd=REPO_ROOT,
|
|
env=env,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
|
|
def test_catalog_repo_expands_home_relative_local_path(self) -> None:
|
|
with patch.dict(os.environ, {"HOME": str(self.home)}):
|
|
self.assertEqual(
|
|
catalog_repo({"repo": "~/.skills"}),
|
|
str(self.skills_home),
|
|
)
|
|
self.assertEqual(
|
|
catalog_checkout_path("skills", {"repo": "~/.skills"}),
|
|
self.skills_home.resolve(),
|
|
)
|
|
|
|
def test_add_collection_installs_every_discovered_skill(self) -> None:
|
|
result = self.run_skiff("add", "test-pack", "-g", "-a", "codex")
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
skill_dir = self.home / ".codex" / "skills"
|
|
self.assertTrue((skill_dir / "first-skill").is_symlink())
|
|
self.assertTrue((skill_dir / "first-skill" / "SKILL.md").is_file())
|
|
self.assertTrue((skill_dir / "second-skill").is_symlink())
|
|
self.assertTrue((skill_dir / "second-skill" / "SKILL.md").is_file())
|
|
self.assertFalse((skill_dir / "test-pack").exists())
|
|
|
|
def test_add_qualified_collection_skill_installs_only_that_skill(self) -> None:
|
|
result = self.run_skiff("add", "test-pack/second-skill", "-g", "-a", "codex")
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
skill_dir = self.home / ".codex" / "skills"
|
|
self.assertFalse((skill_dir / "first-skill").exists())
|
|
self.assertTrue((skill_dir / "second-skill").is_symlink())
|
|
self.assertTrue((skill_dir / "second-skill" / "SKILL.md").is_file())
|
|
|
|
def test_add_collection_with_catalog_source_installs_all(self) -> None:
|
|
result = self.run_skiff(
|
|
"add",
|
|
"test-pack",
|
|
"--source",
|
|
"catalog",
|
|
"-g",
|
|
"-a",
|
|
"codex",
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
skill_dir = self.home / ".codex" / "skills"
|
|
self.assertTrue((skill_dir / "first-skill").is_symlink())
|
|
self.assertTrue((skill_dir / "second-skill").is_symlink())
|
|
|
|
def test_legacy_registry_source_alias_still_installs(self) -> None:
|
|
result = self.run_skiff(
|
|
"add",
|
|
"test-pack",
|
|
"--source",
|
|
"registry",
|
|
"-g",
|
|
"-a",
|
|
"codex",
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertTrue(
|
|
(self.home / ".codex" / "skills" / "first-skill").is_symlink()
|
|
)
|
|
|
|
def test_project_add_records_collection_for_sync(self) -> None:
|
|
project = self.home / "project"
|
|
project.mkdir()
|
|
|
|
added = self.run_skiff(
|
|
"add",
|
|
"test-pack/first-skill",
|
|
"--project",
|
|
str(project),
|
|
"-a",
|
|
"codex",
|
|
)
|
|
link = project / ".agents" / "skills" / "first-skill"
|
|
link.unlink()
|
|
shutil.rmtree(self.home / ".local" / "share" / "skills" / "externals")
|
|
synced = self.run_skiff(
|
|
"sync",
|
|
"--project",
|
|
str(project),
|
|
"-a",
|
|
"codex",
|
|
)
|
|
|
|
self.assertEqual(added.returncode, 0, added.stderr)
|
|
self.assertIn('source: "catalog:test-pack"', project.joinpath(".skills.yaml").read_text())
|
|
self.assertEqual(synced.returncode, 0, synced.stderr)
|
|
self.assertTrue(link.is_symlink())
|
|
self.assertTrue((link / "SKILL.md").is_file())
|
|
|
|
def test_collection_preflight_prevents_partial_install(self) -> None:
|
|
blocked = self.home / ".codex" / "skills" / "second-skill"
|
|
blocked.mkdir(parents=True)
|
|
|
|
result = self.run_skiff("add", "test-pack", "-g", "-a", "codex")
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertFalse((self.home / ".codex" / "skills" / "first-skill").exists())
|
|
self.assertTrue(blocked.is_dir())
|
|
|
|
def test_remove_collection_removes_all_child_links(self) -> None:
|
|
installed = self.run_skiff("add", "test-pack", "-g", "-a", "codex")
|
|
removed = self.run_skiff("remove", "test-pack", "-g", "-a", "codex")
|
|
|
|
self.assertEqual(installed.returncode, 0, installed.stderr)
|
|
self.assertEqual(removed.returncode, 0, removed.stderr)
|
|
skill_dir = self.home / ".codex" / "skills"
|
|
self.assertFalse((skill_dir / "first-skill").exists())
|
|
self.assertFalse((skill_dir / "second-skill").exists())
|
|
|
|
def test_status_expands_collection_children(self) -> None:
|
|
installed = self.run_skiff("add", "test-pack", "-g", "-a", "codex")
|
|
status = self.run_skiff("status", "-a", "codex")
|
|
|
|
self.assertEqual(installed.returncode, 0, installed.stderr)
|
|
self.assertEqual(status.returncode, 0, status.stderr)
|
|
self.assertIn("[catalog:test-pack] first-skill", status.stdout)
|
|
self.assertIn("[catalog:test-pack] second-skill", status.stdout)
|
|
self.assertNotIn("[catalog] test-pack\n (未 fetch)", status.stdout)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|