from __future__ import annotations import os import subprocess import sys import tempfile import unittest from pathlib import Path from pouch.config import load_auto_update_settings, parse_interval from pouch.sources import save_sources 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 def git_init(path: Path) -> None: subprocess.run( ["git", "init", "-b", "main", str(path)], check=True, capture_output=True, ) subprocess.run( ["git", "-C", str(path), "config", "user.name", "Test"], check=True, capture_output=True, ) subprocess.run( ["git", "-C", str(path), "config", "user.email", "test@example.com"], check=True, capture_output=True, ) subprocess.run( ["git", "-C", str(path), "config", "commit.gpgsign", "false"], check=True, capture_output=True, ) def git_commit(path: Path, message: str) -> None: subprocess.run(["git", "-C", str(path), "add", "."], check=True, capture_output=True) subprocess.run( ["git", "-C", str(path), "commit", "-m", message], check=True, capture_output=True, ) class IntervalTests(unittest.TestCase): def test_parse_interval_units(self) -> None: self.assertEqual(parse_interval("24h"), 24 * 3600) self.assertEqual(parse_interval("1d"), 86400) self.assertEqual(parse_interval("30m"), 1800) self.assertEqual(parse_interval(0), 0) self.assertEqual(parse_interval("90"), 90) def test_auto_update_false_disables(self) -> None: self.assertFalse(load_auto_update_settings({"auto_update": False}).enabled) self.assertFalse( load_auto_update_settings({"auto_update": {"enabled": False}}).enabled ) def test_auto_update_interval_from_mapping(self) -> None: settings = load_auto_update_settings({"auto_update": {"interval": "1h"}}) self.assertTrue(settings.enabled) self.assertEqual(settings.interval_seconds, 3600) class SaveSourcesPreservesConfigTests(unittest.TestCase): def test_save_sources_keeps_auto_update_settings(self) -> None: with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "config.yaml" path.write_text( "auto_update:\n enabled: false\n interval: 1h\nsources: {}\n", encoding="utf-8", ) save_sources( { "company": { "local_path": str(Path(tmp) / "skills"), "skills_path": ".", } }, path, ) text = path.read_text(encoding="utf-8") self.assertIn("enabled: false", text) self.assertIn("interval: 1h", text) self.assertIn("company:", text) class AutoUpdateCliTests(unittest.TestCase): def setUp(self) -> None: self.temp_dir = tempfile.TemporaryDirectory() self.home = Path(self.temp_dir.name) self.catalog_upstream = self.home / "catalog-upstream" write_skill(self.catalog_upstream / "skills", "alpha") git_init(self.catalog_upstream) git_commit(self.catalog_upstream, "initial catalog") self.upstream = self.home / "pouch-upstream" self.upstream.mkdir() (self.upstream / "skills").mkdir() (self.upstream / "catalog.yaml").write_text( "demo:\n" f" repo: {self.catalog_upstream}\n" " ref: main\n" " path: skills\n", encoding="utf-8", ) git_init(self.upstream) git_commit(self.upstream, "initial pouch") subprocess.run( ["git", "clone", str(self.upstream), str(self.home / ".pouch")], check=True, capture_output=True, ) subprocess.run( ["git", "-C", str(self.home / ".pouch"), "config", "pull.ff", "only"], check=True, capture_output=True, ) 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) env.pop("POUCH_NO_AUTO_UPDATE", None) return subprocess.run( [sys.executable, "-m", "pouch", *args], cwd=REPO_ROOT, env=env, text=True, capture_output=True, check=False, ) def write_config(self, text: str) -> None: path = self.home / ".config" / "pouch" / "config.yaml" path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text, encoding="utf-8") def test_auto_update_pulls_pouch_on_first_command(self) -> None: marker = self.upstream / "AUTO_UPDATE_MARKER" marker.write_text("yes\n", encoding="utf-8") git_commit(self.upstream, "marker") result = self.run_pouch("list") self.assertEqual(result.returncode, 0, result.stderr) self.assertTrue((self.home / ".pouch" / "AUTO_UPDATE_MARKER").is_file()) self.assertIn("已自动更新: pouch", result.stderr) def test_no_auto_update_flag_skips_pull(self) -> None: marker = self.upstream / "AUTO_UPDATE_MARKER" marker.write_text("yes\n", encoding="utf-8") git_commit(self.upstream, "marker") result = self.run_pouch("--no-auto-update", "list") self.assertEqual(result.returncode, 0, result.stderr) self.assertFalse((self.home / ".pouch" / "AUTO_UPDATE_MARKER").is_file()) self.assertNotIn("已自动更新", result.stderr) def test_config_can_disable_auto_update(self) -> None: self.write_config("auto_update:\n enabled: false\n") marker = self.upstream / "AUTO_UPDATE_MARKER" marker.write_text("yes\n", encoding="utf-8") git_commit(self.upstream, "marker") result = self.run_pouch("list") self.assertEqual(result.returncode, 0, result.stderr) self.assertFalse((self.home / ".pouch" / "AUTO_UPDATE_MARKER").is_file()) def test_dirty_pouch_is_skipped(self) -> None: marker = self.upstream / "AUTO_UPDATE_MARKER" marker.write_text("yes\n", encoding="utf-8") git_commit(self.upstream, "marker") (self.home / ".pouch" / "dirty.txt").write_text("local\n", encoding="utf-8") result = self.run_pouch("list") self.assertEqual(result.returncode, 0, result.stderr) self.assertFalse((self.home / ".pouch" / "AUTO_UPDATE_MARKER").is_file()) def test_add_refreshes_cloned_catalog_source(self) -> None: first = self.run_pouch("add", "demo/alpha", "-g", "-a", "codex") self.assertEqual(first.returncode, 0, first.stderr) write_skill(self.catalog_upstream / "skills", "beta") git_commit(self.catalog_upstream, "add beta") second = self.run_pouch("add", "demo", "-g", "-a", "codex") self.assertEqual(second.returncode, 0, second.stderr) self.assertTrue((self.home / ".codex" / "skills" / "beta").is_symlink()) def test_update_refreshes_cloned_catalog(self) -> None: added = self.run_pouch("--no-auto-update", "add", "demo/alpha", "-g", "-a", "codex") self.assertEqual(added.returncode, 0, added.stderr) write_skill(self.catalog_upstream / "skills", "beta") git_commit(self.catalog_upstream, "add beta") updated = self.run_pouch("--no-auto-update", "update") self.assertEqual(updated.returncode, 0, updated.stderr) checkout = None externals = self.home / ".local" / "share" / "pouch" / "externals" for candidate in externals.rglob("SKILL.md"): if candidate.parent.name == "alpha": checkout = candidate.parents[1] break self.assertIsNotNone(checkout) self.assertTrue((checkout / "beta" / "SKILL.md").is_file()) self.assertIn("catalog:demo", updated.stdout) class ShippedCatalogTests(unittest.TestCase): def test_catalog_includes_taste_skill_like_waza(self) -> None: from pouch import yaml_io data = yaml_io.safe_load( (REPO_ROOT / "catalog.yaml").read_text(encoding="utf-8") ) self.assertIn("waza", data) self.assertIn("taste-skill", data) entry = data["taste-skill"] self.assertIn("Leonxlnx/taste-skill", entry["repo"]) self.assertEqual(entry["path"], "skills") self.assertEqual(entry["ref"], "main") if __name__ == "__main__": unittest.main()