weichen
ba1d3b2b81
Why: PR #682 merged a stale branch that reset five component pins to pre-batch values, silently dropping the batch-1 composition on main. Evidence: materialized byte-exact from the reviewed transport (gatemat-run.json); regression/forward/same/skip/missing-object cases (verify-pingate.json); make test-governance and test-pins re-run below. Impact: new check script wired into check-pins and task-pr-check; PIN_REGRESSION_SKIP=1 documents deliberate downgrades. Refs: feature/audit-followups-pin-gate-m5
188 lines
7.7 KiB
Python
188 lines
7.7 KiB
Python
"""check_pin_regression.sh：gitlink 相对 origin/main 只能前进（PR #682 事故形态）。
|
||
|
||
全部离线：临时目录内的普通 Git 仓库充当子仓，origin 只指向本地 bare 仓库。
|
||
"""
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
SCRIPT = ROOT / "scripts" / "check_pin_regression.sh"
|
||
PR_CHECK = ROOT / "scripts" / "task_pr_check.py"
|
||
COMP = "deps/comp"
|
||
|
||
|
||
class PinRegressionTests(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = Path(tempfile.mkdtemp())
|
||
self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True)
|
||
cfg = self.tmp / "gitconfig"
|
||
cfg.write_text("")
|
||
self.env = dict(
|
||
os.environ,
|
||
GIT_CONFIG_GLOBAL=str(cfg),
|
||
GIT_CONFIG_NOSYSTEM="1",
|
||
GIT_AUTHOR_NAME="t",
|
||
GIT_AUTHOR_EMAIL="t@example.invalid",
|
||
GIT_COMMITTER_NAME="t",
|
||
GIT_COMMITTER_EMAIL="t@example.invalid",
|
||
)
|
||
for key in ("PIN_REGRESSION_SKIP", "PIN_REGRESSION_NO_FETCH"):
|
||
self.env.pop(key, None)
|
||
self.dev = self.tmp / "dev"
|
||
self.dev.mkdir()
|
||
self.git(self.dev, "init", "-q")
|
||
# 子仓：A ← B，另有从 A 分叉的 C。
|
||
self.comp = self.dev / COMP
|
||
self.comp.mkdir(parents=True)
|
||
self.git(self.comp, "init", "-q")
|
||
self.a = self.commit(self.comp, "a")
|
||
self.b = self.commit(self.comp, "b")
|
||
self.git(self.comp, "checkout", "-q", "-b", "side", self.a)
|
||
self.c = self.commit(self.comp, "c")
|
||
|
||
def git(self, cwd, *args):
|
||
return subprocess.run(
|
||
["git", *args], cwd=cwd, env=self.env, check=True, capture_output=True, text=True
|
||
).stdout.strip()
|
||
|
||
def commit(self, repo, text):
|
||
(repo / "f").write_text(text)
|
||
self.git(repo, "add", "f")
|
||
self.git(repo, "commit", "-qm", text)
|
||
return self.git(repo, "rev-parse", "HEAD")
|
||
|
||
def pin(self, sha, path=COMP):
|
||
self.git(self.dev, "update-index", "--add", "--cacheinfo", f"160000,{sha},{path}")
|
||
self.git(self.dev, "commit", "-qm", f"pin {path} {sha[:7]}")
|
||
return self.git(self.dev, "rev-parse", "HEAD")
|
||
|
||
def compose(self, main_pin, head_pin):
|
||
"""origin/main 钉 main_pin，HEAD（PR 分支）钉 head_pin。"""
|
||
main = self.pin(main_pin)
|
||
self.git(self.dev, "update-ref", "refs/remotes/origin/main", main)
|
||
if head_pin != main_pin:
|
||
self.pin(head_pin)
|
||
|
||
def run_check(self, **extra):
|
||
return subprocess.run(
|
||
["bash", str(SCRIPT)], cwd=self.dev, env=self.env | extra, capture_output=True, text=True
|
||
)
|
||
|
||
def test_same_pin_passes(self):
|
||
self.compose(self.a, self.a)
|
||
res = self.run_check()
|
||
self.assertEqual(res.returncode, 0, res.stderr)
|
||
self.assertIn(f"ok {COMP}\n", res.stdout)
|
||
self.assertIn("pin regression ok", res.stdout)
|
||
|
||
def test_forward_pin_passes(self):
|
||
self.compose(self.a, self.b)
|
||
res = self.run_check()
|
||
self.assertEqual(res.returncode, 0, res.stderr)
|
||
self.assertIn(f"ok {COMP} (forward", res.stdout)
|
||
|
||
def test_pr682_stale_branch_carrying_old_pin_fails(self):
|
||
self.compose(self.b, self.a)
|
||
res = self.run_check()
|
||
self.assertEqual(res.returncode, 1, res.stdout)
|
||
self.assertIn(f"pin: {COMP} moves BACKWARD", res.stderr)
|
||
self.assertIn("PR #682", res.stderr)
|
||
self.assertIn("PIN_REGRESSION_SKIP=1", res.stderr)
|
||
self.assertNotIn("pin regression ok", res.stdout)
|
||
|
||
def test_diverged_pin_fails(self):
|
||
self.compose(self.b, self.c)
|
||
res = self.run_check()
|
||
self.assertEqual(res.returncode, 1, res.stdout)
|
||
self.assertIn(f"pin: {COMP} moves BACKWARD", res.stderr)
|
||
|
||
def test_skip_env_downgrades_to_warning(self):
|
||
self.compose(self.b, self.a)
|
||
res = self.run_check(PIN_REGRESSION_SKIP="1")
|
||
self.assertEqual(res.returncode, 0, res.stderr)
|
||
self.assertIn("warn: PIN_REGRESSION_SKIP=1", res.stderr)
|
||
self.assertIn("requires owner review", res.stderr)
|
||
|
||
def test_missing_origin_main_warns_and_passes(self):
|
||
self.pin(self.a)
|
||
res = self.run_check()
|
||
self.assertEqual(res.returncode, 0, res.stderr)
|
||
self.assertIn("warn: origin/main is not resolvable locally", res.stderr)
|
||
|
||
def test_new_component_not_on_main_passes(self):
|
||
self.compose(self.a, self.a)
|
||
other = self.dev / "deps" / "other"
|
||
other.mkdir()
|
||
self.git(other, "init", "-q")
|
||
self.pin(self.commit(other, "o"), "deps/other")
|
||
res = self.run_check()
|
||
self.assertEqual(res.returncode, 0, res.stderr)
|
||
self.assertIn("ok deps/other (new, not on origin/main)", res.stdout)
|
||
|
||
def test_missing_old_object_with_fetch_disabled_fails(self):
|
||
self.compose("1" * 40, self.a)
|
||
res = self.run_check(PIN_REGRESSION_NO_FETCH="1")
|
||
self.assertEqual(res.returncode, 1, res.stdout)
|
||
self.assertIn(f"pin: {COMP} origin/main pin {'1' * 40} is missing locally", res.stderr)
|
||
|
||
def test_missing_old_object_unfetchable_from_local_origin_fails(self):
|
||
bare = self.tmp / "comp-origin.git"
|
||
self.git(self.tmp, "init", "-q", "--bare", str(bare))
|
||
self.git(self.comp, "remote", "add", "origin", str(bare))
|
||
self.compose("2" * 40, self.a)
|
||
res = self.run_check()
|
||
self.assertEqual(res.returncode, 1, res.stdout)
|
||
self.assertIn("could not be fetched", res.stderr)
|
||
|
||
def test_missing_old_object_is_fetched_from_local_origin(self):
|
||
upstream = self.tmp / "upstream"
|
||
self.git(self.tmp, "clone", "-q", str(self.comp), str(upstream))
|
||
newer = self.commit(upstream, "z")
|
||
bare = self.tmp / "comp-origin.git"
|
||
self.git(self.tmp, "clone", "-q", "--bare", str(upstream), str(bare))
|
||
self.git(self.comp, "remote", "add", "origin", str(bare))
|
||
self.compose(newer, self.b)
|
||
res = self.run_check()
|
||
self.assertEqual(res.returncode, 1, res.stdout)
|
||
self.assertNotIn("could not be fetched", res.stderr)
|
||
self.assertIn(f"pin: {COMP} moves BACKWARD", res.stderr)
|
||
self.git(self.comp, "cat-file", "-e", f"{newer}^{{commit}}")
|
||
|
||
def test_uninitialized_component_cannot_be_verified(self):
|
||
self.compose(self.b, self.a)
|
||
shutil.rmtree(self.comp)
|
||
self.comp.mkdir()
|
||
res = self.run_check()
|
||
self.assertEqual(res.returncode, 1, res.stdout)
|
||
self.assertIn(f"pin: {COMP} is not checked out", res.stderr)
|
||
|
||
def run_pr_check_ci(self, **extra):
|
||
env = self.env | {"GITHUB_EVENT_NAME": "pull_request", "GITHUB_BASE_REF": "main",
|
||
"GITHUB_HEAD_REF": "feature/stale", "TASK_TARGET_BRANCH": "main"} | extra
|
||
return subprocess.run([sys.executable, str(PR_CHECK), "ci"], cwd=self.dev, env=env,
|
||
capture_output=True, text=True, timeout=60)
|
||
|
||
def test_task_pr_check_ci_hard_fails_on_regression(self):
|
||
self.compose(self.b, self.a)
|
||
res = self.run_pr_check_ci()
|
||
self.assertEqual(res.returncode, 1, res.stdout + res.stderr)
|
||
self.assertIn(f"[pin-regression] pin: {COMP} moves BACKWARD", res.stdout)
|
||
self.assertIn("[pr-check] FAIL pin regression gate failed", res.stdout)
|
||
self.assertIn("[verdict] PR_CHECK_FAILED", res.stdout)
|
||
|
||
def test_task_pr_check_ci_skip_env_passes(self):
|
||
self.compose(self.b, self.a)
|
||
res = self.run_pr_check_ci(PIN_REGRESSION_SKIP="1")
|
||
self.assertEqual(res.returncode, 0, res.stdout + res.stderr)
|
||
self.assertIn("[pin-regression] warn: PIN_REGRESSION_SKIP=1", res.stdout)
|
||
self.assertIn("[verdict] PR_CHECK_OK", res.stdout)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|