195 lines
7.6 KiB
Python
195 lines
7.6 KiB
Python
"""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()
|