Files
.pouch/tests/test_custom_sources.py
T

234 lines
8.4 KiB
Python

from __future__ import annotations
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
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" 测试 {name} skill。\n---\n\n# {name}\n",
encoding="utf-8",
)
return skill
class CustomSourceTests(unittest.TestCase):
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.skills_home / "registry.yaml").write_text("", 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_local_source_lists_and_installs_qualified_skill(self) -> None:
company = self.home / "company"
expected = write_skill(company / "internal" / "skills", "code-review")
added = self.run_skiff(
"source",
"add",
"company",
"--local",
str(company),
"--skills-path",
"internal/skills",
)
listed = self.run_skiff("list", "--source", "company")
installed = self.run_skiff("add", "company/code-review", "-g", "-a", "codex")
self.assertEqual(added.returncode, 0, added.stderr)
self.assertEqual(listed.returncode, 0, listed.stderr)
self.assertIn("code-review", listed.stdout)
self.assertEqual(installed.returncode, 0, installed.stderr)
link = self.home / ".codex" / "skills" / "code-review"
self.assertTrue(link.is_symlink())
self.assertEqual(link.resolve(), expected.resolve())
config = self.home / ".config" / "skiff" / "config.yaml"
self.assertIn("company:", config.read_text(encoding="utf-8"))
def test_project_add_persists_resolved_source_in_manifest(self) -> None:
company = self.home / "company"
expected = write_skill(company / "skills", "code-review")
project = self.home / "project"
project.mkdir()
added = self.run_skiff("source", "add", "company", "--local", str(company))
self.assertEqual(added.returncode, 0, added.stderr)
result = self.run_skiff(
"add",
"company/code-review",
"--project",
str(project),
"-a",
"codex",
)
self.assertEqual(result.returncode, 0, result.stderr)
manifest = project.joinpath(".skills.yaml").read_text(encoding="utf-8")
self.assertIn('name: "code-review"', manifest)
self.assertIn("source: company", manifest)
self.assertEqual(
(project / ".agents" / "skills" / "code-review").resolve(),
expected.resolve(),
)
def test_project_add_updates_manifest_when_source_changes(self) -> None:
write_skill(self.skills_home / "skills", "code-review")
company = self.home / "company"
write_skill(company / "skills", "code-review")
project = self.home / "project"
project.mkdir()
project.joinpath(".skills.yaml").write_text(
"skills:\n - code-review\n",
encoding="utf-8",
)
added = self.run_skiff("source", "add", "company", "--local", str(company))
self.assertEqual(added.returncode, 0, added.stderr)
result = self.run_skiff(
"add",
"company/code-review",
"--project",
str(project),
"-a",
"codex",
)
self.assertEqual(result.returncode, 0, result.stderr)
manifest = project.joinpath(".skills.yaml").read_text(encoding="utf-8")
self.assertIn("source: company", manifest)
def test_unqualified_duplicate_requires_explicit_source(self) -> None:
write_skill(self.skills_home / "skills", "code-review")
company = self.home / "company"
write_skill(company / "skills", "code-review")
added = self.run_skiff("source", "add", "company", "--local", str(company))
self.assertEqual(added.returncode, 0, added.stderr)
result = self.run_skiff("add", "code-review", "-g", "-a", "codex")
self.assertNotEqual(result.returncode, 0)
self.assertIn("owned/code-review", result.stderr)
self.assertIn("company/code-review", result.stderr)
def test_custom_source_namespace_is_not_shadowed_by_registry_collection(self) -> None:
company = self.home / "company"
expected = write_skill(company / "skills", "code-review")
registry_repo = self.home / "registry-repo"
write_skill(registry_repo / "skills", "other-skill")
self.skills_home.joinpath("registry.yaml").write_text(
"company:\n"
f" repo: {registry_repo}\n"
" ref: main\n"
" path: skills\n",
encoding="utf-8",
)
added = self.run_skiff("source", "add", "company", "--local", str(company))
result = self.run_skiff(
"add",
"company/code-review",
"-g",
"-a",
"codex",
)
self.assertEqual(added.returncode, 0, added.stderr)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(
(self.home / ".codex" / "skills" / "code-review").resolve(),
expected.resolve(),
)
def test_sync_uses_manifest_source(self) -> None:
company = self.home / "company"
expected = write_skill(company / "skills", "code-review")
project = self.home / "project"
project.mkdir()
project.joinpath(".skills.yaml").write_text(
"skills:\n - name: code-review\n source: company\n"
"targets:\n - codex\n",
encoding="utf-8",
)
added = self.run_skiff("source", "add", "company", "--local", str(company))
self.assertEqual(added.returncode, 0, added.stderr)
result = self.run_skiff("sync", "--project", str(project), "-a", "codex")
self.assertEqual(result.returncode, 0, result.stderr)
link = project / ".agents" / "skills" / "code-review"
self.assertEqual(link.resolve(), expected.resolve())
def test_sync_reports_missing_machine_source(self) -> None:
project = self.home / "project"
project.mkdir()
project.joinpath(".skills.yaml").write_text(
"skills:\n - name: code-review\n source: company\n",
encoding="utf-8",
)
result = self.run_skiff("sync", "--project", str(project), "-a", "codex")
self.assertNotEqual(result.returncode, 0)
self.assertIn("本机尚未配置", result.stderr)
self.assertIn("skiff source add company", result.stderr)
def test_git_source_is_cloned_to_default_checkout(self) -> None:
upstream = self.home / "upstream"
write_skill(upstream / "skills", "release-check")
subprocess.run(["git", "init", "-b", "main", str(upstream)], check=True, capture_output=True)
subprocess.run(["git", "-C", str(upstream), "add", "."], check=True)
subprocess.run(
[
"git",
"-C",
str(upstream),
"-c",
"user.name=Test",
"-c",
"user.email=test@example.com",
"commit",
"-m",
"initial",
],
check=True,
capture_output=True,
)
result = self.run_skiff("source", "add", "company", str(upstream))
self.assertEqual(result.returncode, 0, result.stderr)
checkout = self.home / ".local" / "share" / "skiff" / "sources" / "company"
self.assertTrue((checkout / ".git").is_dir())
self.assertTrue((checkout / "skills" / "release-check" / "SKILL.md").is_file())
if __name__ == "__main__":
unittest.main()