"""scripts/agent-exec 薄入口与 pre-commit 委派测试：临时真实 Git 仓库 + 假 dm-server，不访问网络。""" from __future__ import annotations import os import shutil import subprocess import tempfile import unittest from pathlib import Path SOURCE = Path(__file__).resolve().parents[1] HAS_GO = shutil.which("go") is not None GO_MAIN = r'''package main import ( "fmt" "os" "strings" ) func main() { fmt.Printf("fake-dm-server %s\n", strings.Join(os.Args[1:], " ")) } ''' # 预置在缓存中的桩二进制：用于不依赖 go 的 pre-commit 委派测试（入口命中缓存即直接 exec）。 STUB = "#!/bin/sh\nprintf 'fake-dm-server %s\\n' \"$*\"\n" def _clean_env(**extra: str) -> dict[str, str]: env = {k: v for k, v in os.environ.items() if not k.startswith("GIT_")} env.update( GIT_TERMINAL_PROMPT="0", GIT_AUTHOR_NAME="Agent Exec", GIT_AUTHOR_EMAIL="agent-exec@example.com", GIT_COMMITTER_NAME="Agent Exec", GIT_COMMITTER_EMAIL="agent-exec@example.com", ) env.update(extra) return env def _git(cwd: Path, *args: str) -> str: proc = subprocess.run( ["git", "-C", str(cwd), *args], capture_output=True, text=True, env=_clean_env(), check=True, ) return proc.stdout.strip() class AgentExecEntryTests(unittest.TestCase): def setUp(self) -> None: self.tmp = tempfile.TemporaryDirectory(prefix="agent-exec-entry-") self.addCleanup(self.tmp.cleanup) self.base = Path(os.path.realpath(self.tmp.name)) self.cache = self.base / "cache" self.refused_cache = self.base / "refused-cache" self.dm_src = self._make_fake_dm() self.pin = _git(self.dm_src, "rev-parse", "HEAD") self.repo = self._make_super_repo() self.dm = self.repo / "deps" / "development-manager" def _make_fake_dm(self) -> Path: src = self.base / "fake-dm" (src / "cmd" / "dm-server").mkdir(parents=True) (src / "go.mod").write_text("module example.com/fakedm\n\ngo 1.22\n", encoding="utf-8") (src / "cmd" / "dm-server" / "main.go").write_text(GO_MAIN, encoding="utf-8") (src / "README.md").write_text("fake dm\n", encoding="utf-8") _git(src, "init", "-q", "-b", "main") _git(src, "add", "-A") _git(src, "commit", "-q", "-m", "fake dm") return src def _make_super_repo(self) -> Path: repo = self.base / "super" (repo / "scripts" / "hooks").mkdir(parents=True) for src, rel in ( (SOURCE / "scripts" / "agent-exec", "scripts/agent-exec"), (SOURCE / "scripts" / "hooks" / "pre-commit", "scripts/hooks/pre-commit"), ): dest = repo / rel shutil.copy(src, dest) dest.chmod(0o755) (repo / "README.md").write_text("super\n", encoding="utf-8") _git(repo, "init", "-q", "-b", "main") _git(repo, "update-index", "--add", "--cacheinfo", f"160000,{self.pin},deps/development-manager") dm = repo / "deps" / "development-manager" dm.parent.mkdir() _git(repo, "clone", "-q", "--no-checkout", str(self.dm_src), str(dm)) _git(dm, "checkout", "-q", "--detach", self.pin) _git(repo, "add", "-A") _git(repo, "commit", "-q", "-m", "init") return repo def _entry(self, *args: str, cache: Path | None = None) -> subprocess.CompletedProcess[str]: return subprocess.run( [str(self.repo / "scripts" / "agent-exec"), *args], cwd=str(self.repo), capture_output=True, text=True, env=_clean_env(AGENT_EXEC_CACHE=str(cache or self.cache)), ) def _binary(self, cache: Path | None = None) -> Path: return (cache or self.cache) / f"dm-server-{self.pin}" def _assert_no_binary(self, cache: Path) -> None: leftovers = list(cache.glob("dm-server-*")) if cache.exists() else [] self.assertEqual(leftovers, []) def _assert_refused(self, proc: subprocess.CompletedProcess[str], needle: str) -> None: combined = proc.stdout + proc.stderr self.assertEqual(proc.returncode, 2, combined) self.assertIn(needle, proc.stderr) self.assertNotIn("fake-dm-server", combined) self._assert_no_binary(self.refused_cache) @unittest.skipUnless(HAS_GO, "go is not on PATH") def test_positive_run_builds_and_execs_pinned_binary(self) -> None: proc = self._entry("validate", "--repo", ".") self.assertEqual(proc.returncode, 0, proc.stderr) self.assertIn("fake-dm-server agent-exec validate --repo .", proc.stdout) self.assertTrue(self._binary().is_file()) @unittest.skipUnless(HAS_GO, "go is not on PATH") def test_second_run_reuses_cached_binary(self) -> None: first = self._entry("version") self.assertEqual(first.returncode, 0, first.stderr) before = self._binary().stat().st_mtime_ns second = self._entry("version") self.assertEqual(second.returncode, 0, second.stderr) self.assertIn("fake-dm-server agent-exec version", second.stdout) self.assertEqual(self._binary().stat().st_mtime_ns, before) self.assertEqual(list(self.cache.glob(".dm-server-*")), []) def test_refuses_when_dm_head_differs_from_pin(self) -> None: _git(self.dm, "commit", "-q", "--allow-empty", "-m", "move head") proc = self._entry("validate", "--repo", ".", cache=self.refused_cache) self._assert_refused(proc, "与 index 钉住的") self.assertIn("git submodule update --init deps/development-manager", proc.stderr) def test_refuses_tracked_modification_in_dm_checkout(self) -> None: (self.dm / "README.md").write_text("changed\n", encoding="utf-8") proc = self._entry("validate", "--repo", ".", cache=self.refused_cache) self._assert_refused(proc, "存在未提交的修改") def test_refuses_when_gitlink_absent(self) -> None: _git(self.repo, "rm", "--cached", "-q", "deps/development-manager") proc = self._entry("validate", "--repo", ".", cache=self.refused_cache) self._assert_refused(proc, "gitlink") def test_precommit_delegates_admit_to_pinned_binary(self) -> None: _git(self.repo, "config", "core.hooksPath", "scripts/hooks") (self.repo / "src").mkdir() (self.repo / "src" / "b.txt").write_text("staged\n", encoding="utf-8") _git(self.repo, "add", "src/b.txt") main_cache = self.base / "main-cache" rejected = self._commit(main_cache, "on main") combined = rejected.stdout + rejected.stderr self.assertNotEqual(rejected.returncode, 0, combined) self.assertIn("集成分支", combined) self.assertNotIn("fake-dm-server", combined) self._assert_no_binary(main_cache) cache = self.base / "hook-cache" cache.mkdir() stub = self._binary(cache) stub.write_text(STUB, encoding="utf-8") stub.chmod(0o755) _git(self.repo, "checkout", "-q", "-b", "feature/x") accepted = self._commit(cache, "feat: x") combined = accepted.stdout + accepted.stderr self.assertEqual(accepted.returncode, 0, combined) self.assertIn(f"fake-dm-server agent-exec admit --repo {self.repo}", combined) def _commit(self, cache: Path, message: str) -> subprocess.CompletedProcess[str]: return subprocess.run( ["git", "-C", str(self.repo), "commit", "-q", "-m", message], capture_output=True, text=True, env=_clean_env(AGENT_EXEC_CACHE=str(cache)), ) if __name__ == "__main__": unittest.main()