541fb48c1c
Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>
125 lines
4.5 KiB
Python
125 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import stat
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
CHECK = ROOT / "tools" / "check_deploy_baseline.sh"
|
|
BUILD = ROOT / "tools" / "build_image.sh"
|
|
|
|
|
|
def run_check(repo: Path, candidate: str, live: str) -> subprocess.CompletedProcess[str]:
|
|
env = os.environ.copy()
|
|
env["GIT_DIR"] = str(repo / ".git")
|
|
env["GIT_WORK_TREE"] = str(repo)
|
|
return subprocess.run(
|
|
["bash", str(CHECK), candidate, "--live-revision", live],
|
|
cwd=repo,
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
check=False,
|
|
)
|
|
|
|
|
|
def git(repo: Path, *args: str) -> str:
|
|
result = subprocess.run(
|
|
["git", *args],
|
|
cwd=repo,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
return result.stdout.strip()
|
|
|
|
|
|
class DeployBaselineGateTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.tmpdir = tempfile.TemporaryDirectory()
|
|
cls.repo = Path(cls.tmpdir.name) / "repo"
|
|
cls.repo.mkdir()
|
|
git(cls.repo, "init")
|
|
git(cls.repo, "config", "user.email", "gate@example.com")
|
|
git(cls.repo, "config", "user.name", "Gate")
|
|
(cls.repo / "README").write_text("base\n", encoding="utf-8")
|
|
git(cls.repo, "add", "README")
|
|
git(cls.repo, "commit", "-m", "base")
|
|
cls.base = git(cls.repo, "rev-parse", "HEAD")
|
|
|
|
(cls.repo / "online.txt").write_text("live\n", encoding="utf-8")
|
|
git(cls.repo, "add", "online.txt")
|
|
git(cls.repo, "commit", "-m", "online")
|
|
cls.live = git(cls.repo, "rev-parse", "HEAD")
|
|
|
|
git(cls.repo, "checkout", "-b", "successor")
|
|
(cls.repo / "next.txt").write_text("next\n", encoding="utf-8")
|
|
git(cls.repo, "add", "next.txt")
|
|
git(cls.repo, "commit", "-m", "successor of live")
|
|
cls.successor = git(cls.repo, "rev-parse", "HEAD")
|
|
|
|
git(cls.repo, "checkout", "-B", "lagging-main", cls.base)
|
|
(cls.repo / "stale.txt").write_text("stale main\n", encoding="utf-8")
|
|
git(cls.repo, "add", "stale.txt")
|
|
git(cls.repo, "commit", "-m", "lagging main")
|
|
cls.lagging = git(cls.repo, "rev-parse", "HEAD")
|
|
|
|
git(cls.repo, "checkout", "-B", "side", cls.base)
|
|
(cls.repo / "side.txt").write_text("side branch\n", encoding="utf-8")
|
|
git(cls.repo, "add", "side.txt")
|
|
git(cls.repo, "commit", "-m", "unrelated side branch")
|
|
cls.side = git(cls.repo, "rev-parse", "HEAD")
|
|
|
|
git(cls.repo, "checkout", "-B", "successor", cls.successor)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls) -> None:
|
|
cls.tmpdir.cleanup()
|
|
|
|
def test_check_script_is_executable(self) -> None:
|
|
self.assertTrue(CHECK.exists())
|
|
self.assertTrue(stat.S_IXUSR & CHECK.stat().st_mode)
|
|
|
|
def test_successor_of_live_passes(self) -> None:
|
|
result = run_check(self.repo, self.successor, self.live)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn(self.live, result.stdout)
|
|
self.assertIn(self.successor, result.stdout)
|
|
self.assertIn("next.txt", result.stdout)
|
|
self.assertIn("祖先关系通过", result.stdout)
|
|
|
|
def test_lagging_main_is_blocked(self) -> None:
|
|
result = run_check(self.repo, self.lagging, self.live)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("拒绝", result.stderr)
|
|
|
|
def test_side_branch_is_blocked(self) -> None:
|
|
result = run_check(self.repo, self.side, self.live)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("拒绝", result.stderr)
|
|
|
|
def test_unknown_commit_is_blocked(self) -> None:
|
|
result = run_check(self.repo, "deadbeefdeadbeefdeadbeefdeadbeefdeadbeef", self.live)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("无法解析", result.stderr)
|
|
|
|
def test_build_image_calls_the_gate_and_rejects_latest(self) -> None:
|
|
source = BUILD.read_text(encoding="utf-8")
|
|
self.assertIn("check_deploy_baseline.sh", source)
|
|
self.assertIn("禁止构建 latest", source)
|
|
self.assertIn("org.opencontainers.image.revision", source)
|
|
gate = CHECK.read_text(encoding="utf-8")
|
|
self.assertIn("org.opencontainers.image.revision", gate)
|
|
self.assertIn("merge-base --is-ancestor", gate)
|
|
self.assertIn("候选将丢失的提交", gate)
|
|
self.assertIn("禁止人工填写", gate)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|