diff --git a/src/pkgmgr/actions/repository/push.py b/src/pkgmgr/actions/repository/push.py index f62a272..97e56f1 100644 --- a/src/pkgmgr/actions/repository/push.py +++ b/src/pkgmgr/actions/repository/push.py @@ -1,6 +1,5 @@ from __future__ import annotations -import os import subprocess import threading from typing import Any @@ -42,51 +41,65 @@ def _push_target(repo_dir: str) -> tuple[str, str]: return branch, remote -def _unpushed_signatures(repo_dir: str, remote: str) -> tuple[str, list[str]]: - for ref in ("@{u}", f"{remote}/main", f"{remote}/master"): +def _unpushed_signatures( + repo_dir: str, remote: str, branch: str +) -> tuple[str, dict[str, str]]: + for ref in ( + f"refs/remotes/{remote}/{branch}", + "@{u}", + f"refs/remotes/{remote}/main", + f"refs/remotes/{remote}/master", + ): base = _try_git(repo_dir, "merge-base", "HEAD", ref) if base: - return base, run( - ["log", "--format=%G?", f"{base}..HEAD"], cwd=repo_dir - ).split() + lines = run( + ["log", "--format=%H %G?", f"{base}..HEAD"], cwd=repo_dir + ).splitlines() + return base, dict(line.split() for line in lines) raise PushRefusedError( - f"{repo_dir}: no upstream and no {remote}/main or {remote}/master to sign against" + f"{repo_dir}: no {remote}/{branch}, no upstream and no {remote}/main or " + f"{remote}/master to sign against" ) def sign_unpushed(repo_dir: str, *, preview: bool = False) -> None: """ - GPG-sign every unpushed commit that is not signed yet, like git-sign-push. - Args: repo_dir: checkout whose current branch is signed. - preview: print the fetch and rebase instead of running them. + preview: print the fetch and the amend instead of running them. """ - _branch, remote = _push_target(repo_dir) + branch, remote = _push_target(repo_dir) run(["fetch", "--quiet", remote], cwd=repo_dir, preview=preview) - base, signatures = _unpushed_signatures(repo_dir, remote) - unsigned = sum(1 for s in signatures if s != "G") - if not unsigned: + base, signatures = _unpushed_signatures(repo_dir, remote, branch) + head = run(["rev-parse", "HEAD"], cwd=repo_dir) + older = sum(1 for sha, s in signatures.items() if sha != head and s != "G") + if older: + print( + f"[push] WARN {older} older unsigned commit(s) in {base[:10]}..HEAD~ of " + f"{repo_dir} stay unsigned: signing them would rewrite history" + ) + if signatures.get(head, "G") == "G": return if run(["status", "--porcelain", "--untracked-files=no"], cwd=repo_dir): - raise PushRefusedError(f"{repo_dir}: uncommitted changes, cannot re-sign") + raise PushRefusedError(f"{repo_dir}: uncommitted changes, cannot sign HEAD") - cmd = ["git", "rebase", "--rebase-merges", "--force-rebase", "-S", base] - print(f"[push] Signing {unsigned} of {len(signatures)} commit(s) in {repo_dir}") + cmd = [ + "git", + "commit", + "--amend", + "--no-edit", + "--no-verify", + "--allow-empty", + "-S", + ] + print(f"[push] Signing HEAD {head[:10]} in {repo_dir}") if preview: print(f"[PREVIEW] Would run in {repo_dir!r}: {' '.join(cmd)}") return with _SIGN_LOCK: - result = subprocess.run( - cmd, - cwd=repo_dir, - env={**os.environ, "GIT_SEQUENCE_EDITOR": ":"}, - check=False, - ) + result = subprocess.run(cmd, cwd=repo_dir, check=False) if result.returncode != 0: - raise PushRefusedError( - f"{repo_dir}: re-sign rebase failed; resolve or `git rebase --abort`" - ) + raise PushRefusedError(f"{repo_dir}: signing HEAD {head[:10]} failed") def _lease( @@ -145,22 +158,21 @@ def push_branch( Args: repo_dir: checkout whose current branch is pushed. extra_args: flags appended to every `git push`. - sign: refuse unsigned commits in the push range; False pushes them as they are. + sign: refuse an unpushed, unsigned HEAD; False pushes it as it is. refs: extra refs (e.g. a release tag) pushed to every URL. rewrite: replace commits of the old canonical tip; the fetch URL skips the cherry check, mirrors only refuse commits the old tip never had. preview: print the pushes instead of running them. """ branch, remote = _push_target(repo_dir) + head = run(["rev-parse", "HEAD"], cwd=repo_dir) if sign: - _base, signatures = _unpushed_signatures(repo_dir, remote) - if not preview and any(s != "G" for s in signatures): + _base, signatures = _unpushed_signatures(repo_dir, remote, branch) + if not preview and signatures.get(head, "G") != "G": raise PushRefusedError( - f"{repo_dir}: {sum(s != 'G' for s in signatures)} unsigned commit(s) " - "in the push range; run `pkgmgr push` to sign them" + f"{repo_dir}: HEAD {head[:10]} is unsigned; run `pkgmgr push` to sign it" ) - head = run(["rev-parse", "HEAD"], cwd=repo_dir) fetch_url = run(["remote", "get-url", remote], cwd=repo_dir) push_urls = run( ["remote", "get-url", "--push", "--all", remote], cwd=repo_dir diff --git a/tests/integration/test_push_mirror_leases.py b/tests/integration/test_push_mirror_leases.py index 7032fa2..206079e 100644 --- a/tests/integration/test_push_mirror_leases.py +++ b/tests/integration/test_push_mirror_leases.py @@ -203,6 +203,43 @@ class TestPushMirrorLeases(unittest.TestCase): self.assertEqual(self._tip(self.mirror), unsigned) self.assertEqual(git(self.repo, "log", "-1", "--format=%G?"), "N") + def test_only_head_is_signed_and_older_unsigned_commits_stay(self) -> None: + older = self._commit("older") + self._commit("head") + + out = self._push() + + head = git(self.repo, "rev-parse", "HEAD") + self.assertEqual(git(self.repo, "rev-parse", "HEAD^"), older) + self.assertEqual(git(self.repo, "log", "-1", "--format=%G?", head), "G") + self.assertEqual(git(self.repo, "log", "-1", "--format=%G?", older), "N") + self.assertIn("WARN 1 older unsigned commit(s)", out) + self.assertEqual(self._tip(self.canonical), head) + self.assertEqual(self._tip(self.mirror), head) + + def test_pushed_remote_branch_is_the_base_without_upstream(self) -> None: + git(self.repo, "checkout", "-q", "-b", "feature") + with patch.dict(os.environ, {"GIT_COMMITTER_DATE": "2001-01-01T00:00:00+0000"}): + pushed = self._commit("pushed", None, "-S") + git(self.repo, "push", "-q", "origin", "feature") + git(self.repo, "fetch", "-q", "origin") + self._commit("new") + upstream = subprocess.run( + ["git", "rev-parse", "--abbrev-ref", "@{u}"], + cwd=self.repo, + check=False, + capture_output=True, + ) + self.assertNotEqual(upstream.returncode, 0) + + out = self._push() + + head = git(self.repo, "rev-parse", "HEAD") + self.assertEqual(git(self.repo, "rev-parse", "HEAD^"), pushed) + self.assertEqual(git(self.repo, "log", "-1", "--format=%G?", head), "G") + self.assertNotIn("WARN", out) + self.assertEqual(self._tip(self.canonical, "refs/heads/feature"), head) + def test_release_retry_refuses_unsigned_range(self) -> None: self._commit("release") git(self.repo, "tag", "-a", "v1.0.0", "-m", "v1.0.0")