Kevin Veen-BirkenbachandClaude Opus 5.5
919e6a6434
pkgmgr push now signs unpushed commits like git-sign-push and pushes each pushurl with its own --force-with-lease: the fetch URL against the tracking ref, every mirror against its own fetched tip, each only after git cherry shows no commit exists only there. --rewrite allows a deliberate history rewrite. Without a repository selection it pushes the current checkout, including worktrees outside the base dir. pkgmgr release and release --retry sign before tagging and push the branch and tag through the same path. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
207 lines
7.6 KiB
Python
207 lines
7.6 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import redirect_stdout
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from pkgmgr.actions.release.retry import retry_release
|
|
from pkgmgr.actions.repository.push import (
|
|
PushRefusedError,
|
|
push_branch,
|
|
sign_unpushed,
|
|
)
|
|
from pkgmgr.cli.context import CLIContext
|
|
from pkgmgr.cli.parser import create_parser
|
|
from pkgmgr.cli.proxy import maybe_handle_proxy
|
|
|
|
|
|
def git(cwd: str, *args: str) -> str:
|
|
return subprocess.run(
|
|
["git", *args], cwd=cwd, check=True, capture_output=True, text=True
|
|
).stdout.strip()
|
|
|
|
|
|
@unittest.skipUnless(shutil.which("ssh-keygen"), "ssh-keygen signs the commits")
|
|
class TestPushMirrorLeases(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self._tmp.cleanup)
|
|
root = Path(os.path.realpath(self._tmp.name))
|
|
self.root = str(root)
|
|
key = root / "key"
|
|
subprocess.run(
|
|
["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-f", str(key)],
|
|
check=True,
|
|
)
|
|
(root / "allowed").write_text(f"t@example.org {(root / 'key.pub').read_text()}")
|
|
(root / "gitconfig").write_text(
|
|
"[user]\n\tname = t\n\temail = t@example.org\n"
|
|
f"\tsigningkey = {key}\n"
|
|
"[gpg]\n\tformat = ssh\n"
|
|
f'[gpg "ssh"]\n\tallowedSignersFile = {root / "allowed"}\n'
|
|
"[commit]\n\tgpgsign = false\n"
|
|
"[init]\n\tdefaultBranch = main\n"
|
|
)
|
|
env = patch.dict(
|
|
os.environ,
|
|
{"GIT_CONFIG_GLOBAL": str(root / "gitconfig"), "GIT_CONFIG_NOSYSTEM": "1"},
|
|
)
|
|
env.start()
|
|
self.addCleanup(env.stop)
|
|
|
|
self.canonical = str(root / "canonical.git")
|
|
self.mirror = str(root / "mirror.git")
|
|
self.repo = str(root / "repo")
|
|
git(self.root, "init", "-q", "--bare", self.canonical)
|
|
git(self.root, "init", "-q", "--bare", self.mirror)
|
|
git(self.root, "init", "-q", self.repo)
|
|
git(self.repo, "remote", "add", "origin", self.canonical)
|
|
git(self.repo, "remote", "set-url", "--add", "--push", "origin", self.canonical)
|
|
git(self.repo, "remote", "set-url", "--add", "--push", "origin", self.mirror)
|
|
self.init = self._commit("init")
|
|
git(self.repo, "push", "-q", self.canonical, "main")
|
|
git(self.repo, "push", "-q", self.mirror, "main")
|
|
git(self.repo, "fetch", "-q", "origin")
|
|
git(self.repo, "branch", "-q", "--set-upstream-to=origin/main")
|
|
|
|
def _commit(self, name: str, cwd: str | None = None, *flags: str) -> str:
|
|
cwd = cwd or self.repo
|
|
Path(cwd, f"{name}.txt").write_text(name)
|
|
git(cwd, "add", f"{name}.txt")
|
|
git(cwd, "commit", "-q", *flags, "-m", name)
|
|
return git(cwd, "rev-parse", "HEAD")
|
|
|
|
def _push(self, rewrite: bool = False) -> str:
|
|
with redirect_stdout(io.StringIO()) as out:
|
|
sign_unpushed(self.repo)
|
|
push_branch(self.repo, [], rewrite=rewrite)
|
|
return out.getvalue()
|
|
|
|
def _tip(self, bare: str, ref: str = "refs/heads/main") -> str:
|
|
return subprocess.run(
|
|
["git", "rev-parse", "--verify", "--quiet", ref],
|
|
cwd=bare,
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
).stdout.strip()
|
|
|
|
def _assert_shipped_signed(self) -> None:
|
|
head = git(self.repo, "rev-parse", "HEAD")
|
|
self.assertEqual(self._tip(self.canonical), head)
|
|
self.assertEqual(self._tip(self.mirror), head)
|
|
signatures = git(self.repo, "log", "--format=%G?", f"{self.init}..HEAD")
|
|
self.assertEqual(set(signatures.split()), {"G"})
|
|
|
|
def test_mirror_equal_gets_the_signed_commit(self) -> None:
|
|
self._commit("feature")
|
|
|
|
self._push()
|
|
|
|
self._assert_shipped_signed()
|
|
|
|
def test_mirror_behind_is_fast_forwarded(self) -> None:
|
|
self._commit("first", None, "-S")
|
|
git(self.repo, "push", "-q", self.canonical, "main")
|
|
git(self.repo, "fetch", "-q", "origin")
|
|
self._commit("second")
|
|
|
|
self._push()
|
|
|
|
self._assert_shipped_signed()
|
|
|
|
def test_mirror_on_older_rewrite_is_replaced_after_resign(self) -> None:
|
|
unsigned = self._commit("feature")
|
|
git(self.repo, "push", "-q", self.mirror, "main")
|
|
|
|
self._push()
|
|
|
|
self._assert_shipped_signed()
|
|
self.assertNotEqual(self._tip(self.mirror), unsigned)
|
|
|
|
def test_mirror_only_commit_is_refused_and_kept(self) -> None:
|
|
other = os.path.join(self.root, "other")
|
|
git(self.root, "clone", "-q", self.mirror, other)
|
|
foreign = self._commit("foreign", other)
|
|
git(other, "push", "-q", "origin", "main")
|
|
self._commit("feature")
|
|
|
|
with self.assertRaises(PushRefusedError) as cm:
|
|
self._push()
|
|
with self.assertRaises(PushRefusedError):
|
|
self._push(rewrite=True)
|
|
|
|
self.assertIn(self.mirror, str(cm.exception))
|
|
self.assertEqual(self._tip(self.mirror), foreign)
|
|
self.assertEqual(self._tip(self.canonical), git(self.repo, "rev-parse", "HEAD"))
|
|
|
|
def test_content_rewrite_needs_the_rewrite_flag(self) -> None:
|
|
self._commit("feature")
|
|
self._push()
|
|
shipped = git(self.repo, "rev-parse", "HEAD")
|
|
Path(self.repo, "feature.txt").write_text("changed")
|
|
git(self.repo, "commit", "-q", "-a", "--amend", "--no-edit")
|
|
|
|
with self.assertRaises(PushRefusedError) as cm:
|
|
self._push()
|
|
|
|
self.assertIn(self.canonical, str(cm.exception))
|
|
self.assertEqual(self._tip(self.canonical), shipped)
|
|
self.assertEqual(self._tip(self.mirror), shipped)
|
|
|
|
self._push(rewrite=True)
|
|
|
|
self._assert_shipped_signed()
|
|
self.assertNotEqual(self._tip(self.mirror), shipped)
|
|
|
|
def test_worktree_outside_base_dir_is_pushed(self) -> None:
|
|
worktree = os.path.join(self.root, "wt")
|
|
git(self.repo, "worktree", "add", "-q", "-b", "feature", worktree)
|
|
head_before = self._commit("feature", worktree)
|
|
ctx = CLIContext(
|
|
config_merged={},
|
|
repositories_base_dir=os.path.join(self.root, "base"),
|
|
all_repositories=[],
|
|
binaries_dir=os.path.join(self.root, "bin"),
|
|
user_config_path=os.path.join(self.root, "config.yaml"),
|
|
)
|
|
args = create_parser("pkgmgr").parse_args(["push", "-j", "1"])
|
|
cwd = os.getcwd()
|
|
os.chdir(worktree)
|
|
self.addCleanup(os.chdir, cwd)
|
|
|
|
with redirect_stdout(io.StringIO()), self.assertRaises(SystemExit) as cm:
|
|
maybe_handle_proxy(args, ctx)
|
|
|
|
self.assertEqual(cm.exception.code, 0)
|
|
head = git(worktree, "rev-parse", "HEAD")
|
|
self.assertNotEqual(head, head_before)
|
|
self.assertEqual(self._tip(self.canonical, "refs/heads/feature"), head)
|
|
self.assertEqual(self._tip(self.mirror, "refs/heads/feature"), head)
|
|
self.assertEqual(
|
|
git(worktree, "rev-parse", "--abbrev-ref", "@{u}"), "origin/feature"
|
|
)
|
|
|
|
def test_release_retry_refuses_unsigned_range(self) -> None:
|
|
self._commit("release")
|
|
git(self.repo, "tag", "-a", "v1.0.0", "-m", "v1.0.0")
|
|
cwd = os.getcwd()
|
|
os.chdir(self.repo)
|
|
self.addCleanup(os.chdir, cwd)
|
|
|
|
with redirect_stdout(io.StringIO()), self.assertRaises(PushRefusedError):
|
|
retry_release(pyproject_path=os.path.join(self.repo, "pyproject.toml"))
|
|
|
|
self.assertEqual(self._tip(self.canonical), git(self.repo, "rev-parse", "@{u}"))
|
|
self.assertEqual(self._tip(self.canonical, "refs/tags/v1.0.0"), "")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|