.pouch/tests/test_catalog_collections.py

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 pouch import catalog
from pouch.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 / ".pouch"
        (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_pouch(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", "pouch", *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": "~/.pouch"}),
                str(self.skills_home),
            )
            self.assertEqual(
                catalog_checkout_path("skills", {"repo": "~/.pouch"}),
                self.skills_home.resolve(),
            )

    def test_add_collection_installs_every_discovered_skill(self) -> None:
        result = self.run_pouch("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_pouch("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_pouch(
            "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_pouch(
            "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_pouch(
            "add",
            "test-pack/first-skill",
            "--project",
            str(project),
            "-a",
            "codex",
        )
        link = project / ".agents" / "skills" / "first-skill"
        link.unlink()
        shutil.rmtree(self.home / ".local" / "share" / "pouch" / "externals")
        synced = self.run_pouch(
            "sync",
            "--project",
            str(project),
            "-a",
            "codex",
        )

        self.assertEqual(added.returncode, 0, added.stderr)
        self.assertIn('source: "catalog:test-pack"', project.joinpath(".pouch.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_pouch("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_pouch("add", "test-pack", "-g", "-a", "codex")
        removed = self.run_pouch("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_add_refreshes_collection_before_installing_new_children(self) -> None:
        first = self.run_pouch("add", "test-pack/first-skill", "-g", "-a", "codex")
        self.assertEqual(first.returncode, 0, first.stderr)

        write_skill(self.upstream / "skills", "third-skill")
        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",
                "add third",
            ],
            check=True,
            capture_output=True,
        )

        second = self.run_pouch("add", "test-pack", "-g", "-a", "codex")
        skill_dir = self.home / ".codex" / "skills"

        self.assertEqual(second.returncode, 0, second.stderr)
        self.assertTrue((skill_dir / "third-skill").is_symlink())
        self.assertTrue((skill_dir / "third-skill" / "SKILL.md").is_file())

    def test_status_expands_collection_children(self) -> None:
        installed = self.run_pouch("add", "test-pack", "-g", "-a", "codex")
        status = self.run_pouch("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()