diff --git a/scripts/bump_package.py b/scripts/bump_package.py index ed1627a..a521d20 100644 --- a/scripts/bump_package.py +++ b/scripts/bump_package.py @@ -1,40 +1,51 @@ #!/usr/bin/env python3 -"""Bump package version via an auto-merging PR. +"""Bump package version via a PR that the release app fast-forwards itself. Compatible with branch-protection rulesets that require all changes to land through PRs and require commits to be signed: 0. If the configured version already has a merged bump commit on main but - no release tag (a prior run's merge-wait timed out before tagging), the - missing tag is pushed and the run stops. Re-running cz in that state - would double-bump. + no release tag (a prior run failed between merge and tag), the missing + tag is pushed and the run stops. Re-running cz in that state would + double-bump. 1. ``cz bump --files-only`` updates ``pyproject.toml`` (cz version field) and ``CHANGELOG.md`` in the package directory; no local commit or tag. 2. A release-line-specific bump branch is created on the remote at the - current target-branch tip via the REST refs API. + current target-branch tip via the REST refs API. A stale branch left by a + failed prior run is force-moved to that tip instead of failing the run. 3. The bumped files are committed onto that branch via the GraphQL ``createCommitOnBranch`` mutation, which signs the commit as the authenticated bot identity. -4. A PR is opened with auto-merge armed. Once its checks are green the - script also asks for a squash merge, and auto-merge stays armed as the - fallback: it only fires when every branch requirement is actually - satisfied. -5. The script polls until the PR merges, captures the squash-merge SHA on - the target branch, then creates and pushes the ``-`` tag - at that SHA. Tags trigger the existing ``release.yml`` publish workflow. +4. A PR is opened so the bump has CI, a changelog entry and an audit trail. + If an open PR for the branch already exists, it is reused. +5. The script waits for every check on the PR to conclude successfully. A + PR whose checks have not registered yet reads as pending, never passing, + so a fresh commit cannot slip through before Actions reports on it. +6. The script fast-forwards the target branch to the PR head with a strict + (non-force) refs API update, performed by the release app. The + require-review ruleset lists the app as a bypass actor, and GitHub only + honours app bypass on direct ref updates, never on the merge API or + auto-merge, so this is the one merge path that does not strand the PR + waiting for a human review. Before updating, the PR's base SHA is + compared against the live target branch: if the branch moved (seen by + that guard, or by the strict update being refused), the bump branch is + rebuilt on the new tip and its checks are awaited again. +7. GitHub marks the PR merged once its head is reachable from the base. The + script verifies that, then creates and pushes the ``-`` + tag at the merged SHA. Tags trigger the existing ``release.yml`` publish + workflow. The bump branch is deleted once the tag is pushed. The tag is the only thing that publishes a release, so it is created from a -merge and nothing else. If the branch policy refuses the merge, the job fails -with the PR link instead of forcing the ref or tagging an unmerged commit: a -released version that never landed on the target branch leaves the version -file behind PyPI. - -The runner needs: - -- ``GH_TOKEN`` (or ``GITHUB_TOKEN``) in env, scoped to allow ``gh api`` - calls and PR creation. Provided by the workflow via - ``secrets.RELEASE_GITHUB_PAT``. -- A repo configured with auto-merge enabled and a squash-merge option. +verified merge and nothing else. If the checks fail, the fast-forward is +refused or the PR does not report merged, the job fails with the PR link +instead of forcing the ref or tagging an unmerged commit: a released version +that never landed on the target branch leaves the version file behind PyPI. +A PR that someone merges by hand while the script waits is adopted: the run +tags that merge commit and exits successfully instead of failing. + +The runner needs ``GH_TOKEN`` in env: a GitHub App installation token with +contents write (branch and tag refs) and pull-requests write, generated by +the workflow for the release app. """ import argparse @@ -130,10 +141,12 @@ def tag_exists_on_remote(tag: str) -> bool: def recover_untagged_bump(repo: str, package_name: str, package_dir: str) -> bool | None: """Push the missing release tag for a bump that merged without one. - A bump PR can merge after this job's merge-wait times out, leaving the - version files ahead of the last release tag with nothing to trigger the - publish. When the configured version has a merged bump commit on the - target branch but no tag, push the tag at that commit and stop. + A bump can land on the target branch without its tag when a run fails + between the fast-forward and the tag step, or when a bump PR is merged by + hand. That leaves the version files ahead of the last release tag with + nothing to trigger the publish. When the configured version has a merged + bump commit on the target branch but no tag, push the tag at that commit + and stop. Returns ``True`` when the missing tag was pushed (the bump is complete), ``False`` when the tag push failed, and ``None`` when there is nothing @@ -219,6 +232,14 @@ def bump_branch_name(target_branch: str, package_name: str, version: str) -> str def create_remote_branch(repo: str, branch: str, sha: str) -> bool: + """Create the bump branch at ``sha``, reusing a stale one if it exists. + + A failed prior run leaves its bump branch behind, and re-running the job + computes the same version and therefore the same branch name. An existing + ref is force-moved to the intended SHA instead of failing the run. Only + bump branches are ever force-moved; the target branch is only ever + fast-forwarded. + """ print(f"Creating remote branch {branch} at {sha[:8]}...") exit_code, _, stderr = run_command( [ @@ -233,12 +254,45 @@ def create_remote_branch(repo: str, branch: str, sha: str) -> bool: f"sha={sha}", ] ) - if exit_code != 0: + if exit_code == 0: + return True + if "already exists" not in stderr.lower(): print(f"Failed to create remote branch: {stderr}") return False + + print( + f"Branch {branch} already exists (stale from a prior run); " + f"force-moving it to {sha[:8]}..." + ) + exit_code, _, stderr = run_command( + [ + "gh", + "api", + "-X", + "PATCH", + f"repos/{repo}/git/refs/heads/{branch}", + "-f", + f"sha={sha}", + "-F", + "force=true", + ] + ) + if exit_code != 0: + print(f"Failed to move existing branch {branch}: {stderr}") + return False return True +def delete_remote_branch(repo: str, branch: str) -> None: + """Delete the bump branch. Best-effort: the release is already complete.""" + print(f"Deleting branch {branch}...") + exit_code, _, stderr = run_command( + ["gh", "api", "-X", "DELETE", f"repos/{repo}/git/refs/heads/{branch}"] + ) + if exit_code != 0: + print(f"Could not delete branch {branch} (continuing): {stderr}") + + def create_signed_commit_on_branch( repo: str, branch: str, @@ -320,16 +374,23 @@ def create_signed_commit_on_branch( def wait_for_pr_stable(pr_number: int, timeout_seconds: int = 120) -> bool: """Poll mergeStateStatus until GitHub has a definite state for the PR. - A freshly opened PR starts as UNKNOWN or UNSTABLE while required checks - register. Auto-merge can only be enabled once the PR leaves that limbo. + A freshly opened PR, and a bump branch that was just rebuilt, start as + UNKNOWN or UNSTABLE while GitHub computes mergeability. This is a + settling delay, not the checks gate: :func:`checks_verdict` treats an + unregistered rollup as pending, so a commit can never merge before its + checks appear. A PR that is not OPEN returns immediately; the caller's + next read handles MERGED or CLOSED. """ deadline = time.time() + timeout_seconds while time.time() < deadline: exit_code, stdout, _ = run_command( - ["gh", "pr", "view", str(pr_number), "--json", "mergeStateStatus"] + ["gh", "pr", "view", str(pr_number), "--json", "state,mergeStateStatus"] ) if exit_code == 0: - state = json.loads(stdout).get("mergeStateStatus", "") + data = json.loads(stdout) + if data.get("state") != "OPEN": + return True + state = data.get("mergeStateStatus", "") print(f"PR #{pr_number} merge state: {state}") if state not in ("UNKNOWN", "UNSTABLE"): return True @@ -338,19 +399,22 @@ def wait_for_pr_stable(pr_number: int, timeout_seconds: int = 120) -> bool: return False -def create_pr_with_automerge( +def create_bump_pr( branch: str, target_branch: str, package_name: str, new_version: str ) -> int | None: - """Open a PR for the bump branch with auto-merge (squash) enabled. + """Open a PR for the bump branch, reusing an open one if it exists. + A failed prior run can leave its PR open; gh refuses to create a second + PR for the same head and prints the existing PR's URL, which is adopted. Returns the PR number on success, ``None`` on failure. """ title = f"bump: {package_name} → {new_version}" pr_body = ( f"Auto-bump for `{package_name}` to `{new_version}`.\n\n" - "Generated by `scripts/bump_package.py`. The branch tag will be " - "created at the squash-merge SHA after this PR merges, which " - "triggers the publish workflow." + "Generated by `scripts/bump_package.py`. Once the checks pass, the " + "release app fast-forwards the target branch to this PR's head " + "under its ruleset bypass, then creates the release tag at that " + "SHA, which triggers the publish workflow." ) print(f"Opening PR for {branch}...") @@ -370,142 +434,385 @@ def create_pr_with_automerge( ] ) if exit_code != 0: - print(f"Failed to create PR: {stderr}") - return None - - pr_url = stdout.strip().splitlines()[-1] - pr_number_match = re.search(r"/pull/(\d+)", pr_url) - if not pr_number_match: - print(f"Could not parse PR number from gh output: {pr_url}") - return None - pr_number = int(pr_number_match.group(1)) - print(f"Opened PR #{pr_number}: {pr_url}") + existing = re.search(r"/pull/(\d+)", stderr) + if not existing: + print(f"Failed to create PR: {stderr}") + return None + pr_number = int(existing.group(1)) + print(f"Reusing existing open PR #{pr_number} for {branch}.") + else: + pr_url = stdout.strip().splitlines()[-1] + pr_number_match = re.search(r"/pull/(\d+)", pr_url) + if not pr_number_match: + print(f"Could not parse PR number from gh output: {pr_url}") + return None + pr_number = int(pr_number_match.group(1)) + print(f"Opened PR #{pr_number}: {pr_url}") if not wait_for_pr_stable(pr_number): return None - - print("Enabling auto-merge (squash)...") - exit_code, _, stderr = run_command( - ["gh", "pr", "merge", str(pr_number), "--auto", "--squash"] - ) - if exit_code != 0: - print(f"Failed to enable auto-merge: {stderr}") - return None return pr_number -def checks_green(pr_data: dict) -> bool: - """True when every check in the PR's status rollup has finished cleanly. +OK_CONCLUSIONS = {"SUCCESS", "NEUTRAL", "SKIPPED"} + + +def check_name(check: dict) -> str: + """Name a rollup entry: check runs carry ``name``, statuses ``context``.""" + return check.get("name") or check.get("context") or "" + + +def checks_verdict(pr_data: dict) -> tuple[str, str | None]: + """Summarise the PR's status rollup as ``SUCCESS``, ``PENDING`` or + ``FAILURE``, plus the name of the check that decided a non-success verdict. - An empty rollup counts as green (no required checks registered). + Every check on the PR must succeed, required or not. That is deliberately + conservative for a gate that merges under ruleset bypass: a failure in an + optional check aborts the bump, and a re-run retries it. + + ``FAILURE`` wins over ``PENDING``: one failed check sinks the bump even + while others are still running. An empty rollup is ``PENDING``, never + success: right after a commit is created, Actions has not registered its + check runs yet (and a token without checks read sees nothing at all); + reading that emptiness as success would merge an untested commit. The + pending state is bounded by the caller's timeout. """ - ok_conclusions = {"SUCCESS", "NEUTRAL", "SKIPPED"} - for check in pr_data.get("statusCheckRollup") or []: + rollup = pr_data.get("statusCheckRollup") or [] + if not rollup: + return "PENDING", None + verdict = "SUCCESS" + blocker: str | None = None + for check in rollup: status = (check.get("status") or "").upper() conclusion = (check.get("conclusion") or "").upper() - if status and status != "COMPLETED": - return False - if conclusion and conclusion not in ok_conclusions: - return False - if not status and not conclusion: - return False - return True + if conclusion and conclusion not in OK_CONCLUSIONS: + return "FAILURE", check_name(check) + if (status and status != "COMPLETED") or (not status and not conclusion): + verdict = "PENDING" + blocker = check_name(check) + return verdict, blocker def pr_url(repo: str, pr_number: int) -> str: return f"https://github.com/{repo}/pull/{pr_number}" -def wait_for_pr_merge( +def read_pr(pr_number: int, fields: str) -> dict | None: + exit_code, stdout, stderr = run_command( + ["gh", "pr", "view", str(pr_number), "--json", fields] + ) + if exit_code != 0: + print(f"Failed to read PR #{pr_number}: {stderr}") + return None + try: + return json.loads(stdout) + except json.JSONDecodeError: + print(f"Could not parse PR JSON: {stdout}") + return None + + +def wait_for_checks( repo: str, pr_number: int, - target_branch: str, - timeout_seconds: int = 1800, -) -> str | None: - """Poll the PR until it merges. Returns the merge commit SHA on the target branch. - - Returns ``None`` if the PR is closed without merging, if the merge is - refused, or if the timeout elapses; the caller must not tag in that case. - Polls every 30s; logs each status change so the run is debuggable. + timeout_seconds: int = 1200, + poll_seconds: int = 30, +) -> dict | None: + """Wait for every check on the PR to conclude successfully. + + Returns ``{"head": sha}`` with the head SHA the checks ran against, or + ``{"merged": sha}`` when the PR merged while waiting (someone else landed + the bump; only the tag is still owed). Returns ``None`` if any check + fails, the PR is closed without merging, or the timeout elapses. Pending + checks (including a rollup with no registered checks) are never treated + as passing: the caller must not merge on ``None``. """ - print(f"Waiting for PR #{pr_number} to merge (timeout {timeout_seconds}s)...") + print(f"Waiting for checks on PR #{pr_number} (timeout {timeout_seconds}s)...") deadline = time.time() + timeout_seconds - last_state = None - direct_merge_attempts = 0 + last_verdict = None + + while True: + data = read_pr(pr_number, "state,headRefOid,statusCheckRollup,mergeCommit") + if data is not None: + state = data.get("state") + if state == "MERGED": + sha = (data.get("mergeCommit") or {}).get("oid") + if sha: + print( + f"PR #{pr_number} was merged by someone else while waiting " + f"on checks; adopting its merge commit {sha[:8]}." + ) + return {"merged": sha} + # mergeCommit can lag the state flip; keep polling for the oid. + print(f"PR #{pr_number} is MERGED; waiting for its merge commit oid...") + elif state != "OPEN": + print(f"PR #{pr_number} is {state}, not merging: {pr_url(repo, pr_number)}") + return None + else: + verdict, blocker = checks_verdict(data) + if verdict != last_verdict: + suffix = f" ({blocker})" if blocker else "" + print(f"PR #{pr_number} checks: {verdict}{suffix}") + last_verdict = verdict + if verdict == "FAILURE": + print( + f"Check {blocker!r} failed on PR #{pr_number}; not merging: " + f"{pr_url(repo, pr_number)}" + ) + return None + if verdict == "SUCCESS": + head_sha = data.get("headRefOid") + if not head_sha: + print("Checks passed but no headRefOid was returned.") + return None + return {"head": head_sha} + + if time.time() >= deadline: + break + time.sleep(poll_seconds) + + print(f"Timed out waiting for checks on PR #{pr_number}: {pr_url(repo, pr_number)}") + return None - while time.time() < deadline: - exit_code, stdout, stderr = run_command( - [ - "gh", - "pr", - "view", - str(pr_number), - "--json", - "state,mergeCommit,statusCheckRollup", - ] + +def get_live_branch_sha(repo: str, branch: str) -> str | None: + """Return the branch tip as the remote sees it right now (not the local clone).""" + exit_code, stdout, stderr = run_command( + ["gh", "api", f"repos/{repo}/git/ref/heads/{branch}", "-q", ".object.sha"] + ) + if exit_code != 0 or not stdout.strip(): + print(f"Failed to read live {branch} tip: {stderr}") + return None + return stdout.strip() + + +def files_changed_between(repo: str, base_sha: str, head_sha: str) -> list[str] | None: + exit_code, stdout, stderr = run_command( + ["gh", "api", f"repos/{repo}/compare/{base_sha}...{head_sha}", "-q", ".files[].filename"] + ) + if exit_code != 0: + print(f"Failed to compare {base_sha[:8]}...{head_sha[:8]}: {stderr}") + return None + return [line for line in stdout.splitlines() if line] + + +def rebase_bump_branch( + repo: str, + branch: str, + old_base_sha: str, + new_base_sha: str, + package_dir: str, + files: list[str], + headline: str, + body: str, +) -> bool: + """Rebuild the bump branch as a single signed commit on ``new_base_sha``. + + The bump commit is regenerated from the working-tree files rather than + rebased with git, because only ``createCommitOnBranch`` produces a commit + signed as the app. If the target branch's new commits touched the bumped + files, the regenerated content would silently clobber them; if they + touched anything else under ``package_dir``, the release would ship that + code under a version and changelog computed before it existed. Both cases + are refused and the job must be re-run to recompute the bump. + + Only the bump branch is ever force-moved here; the target branch is only + ever fast-forwarded. + """ + print( + f"Target moved from {old_base_sha[:8]} to {new_base_sha[:8]}; " + f"rebuilding {branch} on the new tip..." + ) + changed = files_changed_between(repo, old_base_sha, new_base_sha) + if changed is None: + return False + prefix = "" if package_dir in ("", ".") else package_dir.rstrip("/") + "/" + overlap = sorted( + path for path in changed if path in set(files) or path.startswith(prefix) + ) + if overlap: + print( + f"Target branch changed files this bump depends on ({overlap}) since " + "it was computed; refusing to rebuild. Re-run the job to recompute " + "the bump." ) - if exit_code != 0: - print(f"Failed to read PR status: {stderr}") - time.sleep(30) - continue + return False - try: - data = json.loads(stdout) - except json.JSONDecodeError: - print(f"Could not parse PR status JSON: {stdout}") - time.sleep(30) - continue + exit_code, _, stderr = run_command( + [ + "gh", + "api", + "-X", + "PATCH", + f"repos/{repo}/git/refs/heads/{branch}", + "-f", + f"sha={new_base_sha}", + "-F", + "force=true", + ] + ) + if exit_code != 0: + print(f"Failed to move {branch} to {new_base_sha[:8]}: {stderr}") + return False + + return create_signed_commit_on_branch( + repo, branch, new_base_sha, files, headline=headline, body=body + ) + + +def fast_forward_target(repo: str, target_branch: str, head_sha: str) -> bool: + """Fast-forward the target branch to ``head_sha`` with a strict ref update. + + ``force`` is deliberately absent: GitHub rejects a non-fast-forward + update with 422, so a target branch that moved underneath us can never + be overwritten. + """ + print(f"Fast-forwarding {target_branch} to {head_sha[:8]} via REST refs API...") + exit_code, _, stderr = run_command( + [ + "gh", + "api", + "-X", + "PATCH", + f"repos/{repo}/git/refs/heads/{target_branch}", + "-f", + f"sha={head_sha}", + ] + ) + if exit_code != 0: + print(f"Fast-forward of {target_branch} refused: {stderr.strip()[:300]}") + return False + print( + f"{target_branch} fast-forwarded to {head_sha[:8]} by the release app. " + "This direct ref update is the merge: the app is a bypass actor on the " + "require-review ruleset, and GitHub honours app bypass only on direct " + "ref updates, so no review appears on the PR." + ) + return True - state = data.get("state") - if state != last_state: - print(f"PR #{pr_number} state: {state}") - last_state = state - if state == "MERGED": - merge_commit = data.get("mergeCommit") or {} - sha = merge_commit.get("oid") - if not sha: - print("PR is MERGED but no mergeCommit oid was returned.") +def verify_pr_merged( + repo: str, + pr_number: int, + expected_sha: str, + timeout_seconds: int = 120, + poll_seconds: int = 5, +) -> str | None: + """Confirm GitHub now reports the PR as MERGED at ``expected_sha``. + + Returns the merge commit SHA, or ``None`` if the PR does not reach + MERGED within the window; the caller must not tag in that case. + """ + print(f"Verifying PR #{pr_number} reports merged...") + deadline = time.time() + timeout_seconds + while True: + data = read_pr(pr_number, "state,mergeCommit") + if data is not None: + state = data.get("state") + if state == "MERGED": + sha = (data.get("mergeCommit") or {}).get("oid") + if not sha: + print("PR is MERGED but no mergeCommit oid was returned.") + return None + if sha != expected_sha: + print( + f"PR #{pr_number} merged at {sha[:8]}, not the fast-forwarded " + f"head {expected_sha[:8]}; refusing to tag." + ) + return None + print(f"PR #{pr_number} merged at {sha[:8]}") + return sha + if state == "CLOSED": + print(f"PR #{pr_number} is CLOSED, not MERGED: {pr_url(repo, pr_number)}") return None - print(f"PR #{pr_number} merged at {sha[:8]}") - return sha + if time.time() >= deadline: + break + time.sleep(poll_seconds) + + print( + f"PR #{pr_number} did not report MERGED within {timeout_seconds}s after the " + f"fast-forward: {pr_url(repo, pr_number)}" + ) + return None - if state == "CLOSED": - print(f"PR #{pr_number} was closed without merging: {pr_url(repo, pr_number)}") + +def merge_bump_pr( + repo: str, + pr_number: int, + branch: str, + target_branch: str, + base_sha: str, + package_dir: str, + files: list[str], + headline: str, + body: str, + max_rounds: int = 3, +) -> str | None: + """Land the bump PR by fast-forwarding the target branch as the release app. + + Each round: wait for green checks, confirm the live target tip is still + the SHA the bump branch was built on, fast-forward, verify MERGED. If + the target moved (seen by the guard, or by the strict fast-forward being + refused inside the guard window), the bump branch is rebuilt on the new + tip and the round repeats. A PR that someone else merged mid-wait is + adopted as the merge. Returns the merged SHA, or ``None`` when the + caller must not tag. + """ + for round_number in range(1, max_rounds + 1): + outcome = wait_for_checks(repo, pr_number) + if outcome is None: return None + if "merged" in outcome: + return outcome["merged"] + head_sha = outcome["head"] - if state == "OPEN" and direct_merge_attempts < 3 and checks_green(data): - # Auto-merge waits for requirements the app is entitled to bypass - # (required reviews), so ask for the merge directly too. A refusal - # leaves auto-merge armed and the PR unmerged; it never escalates to - # a ref update, which would land the bump outside the branch policy. - direct_merge_attempts += 1 - exit_code, _, stderr = run_command( - ["gh", "pr", "merge", str(pr_number), "--squash"] + live_sha = get_live_branch_sha(repo, target_branch) + if live_sha is None: + return None + if live_sha != base_sha: + print( + f"Strict fast-forward guard: {target_branch} is at {live_sha[:8]} but " + f"the bump branch is based on {base_sha[:8]} (round {round_number}/{max_rounds})." ) - if exit_code == 0: - print(f"Merged PR #{pr_number} directly as the bypass actor.") - else: - print( - f"Merge attempt {direct_merge_attempts} refused " - f"({stderr.strip()[:200]}); auto-merge stays armed." - ) + if round_number == max_rounds: + print("Out of rebuild rounds; not merging.") + return None + if not rebase_bump_branch( + repo, branch, base_sha, live_sha, package_dir, files, headline, body + ): + return None + # The rebuilt head is seconds old. Wait for GitHub to settle it so + # the next rollup read reflects the new commit; checks_verdict + # treats an unregistered rollup as pending, so the rebuilt commit + # cannot merge before Actions reports on it either way. + if not wait_for_pr_stable(pr_number): + return None + base_sha = live_sha + continue + + if fast_forward_target(repo, target_branch, head_sha): + return verify_pr_merged(repo, pr_number, head_sha) - time.sleep(30) + # The strict (non-force) update was refused: the target moved inside + # the guard window. Re-read the live tip on the next round and rebuild + # there instead of aborting. + print( + f"Fast-forward refused mid-round; re-probing {target_branch} " + f"(round {round_number}/{max_rounds})." + ) - print(f"Timeout waiting for PR #{pr_number} to merge: {pr_url(repo, pr_number)}") + print("Out of merge rounds; not merging.") return None def create_and_push_tag(repo: str, tag: str, sha: str) -> bool: """Create the tag on the remote pointing at ``sha`` and push it. - ``sha`` must be a merge commit on the target branch: the tag publishes the - release, so it may only ever point at a bump that actually landed. + ``sha`` must be a merged commit on the target branch: the tag publishes + the release, so it may only ever point at a bump that actually landed. Uses the REST refs API rather than ``git push --tags`` so the operation works even if the runner's local main is behind (the workflow doesn't - re-fetch after the merge poll). + re-fetch after the fast-forward). """ print(f"Creating tag {tag} at {sha[:8]} via REST refs API...") exit_code, _, stderr = run_command( @@ -567,42 +874,48 @@ def bump_package( if not create_remote_branch(repo, branch, parent_sha): return False + headline = f"bump: {package_name} → {new_version}" + body = f"Auto-bump for {package_name}." if not create_signed_commit_on_branch( - repo, - branch, - parent_sha, - modified, - headline=f"bump: {package_name} → {new_version}", - body=f"Auto-bump for {package_name}.", + repo, branch, parent_sha, modified, headline=headline, body=body ): return False - pr_number = create_pr_with_automerge( - branch, target_branch, package_name, new_version - ) + pr_number = create_bump_pr(branch, target_branch, package_name, new_version) if pr_number is None: return False - merge_sha = wait_for_pr_merge(repo, pr_number, target_branch) + merge_sha = merge_bump_pr( + repo, + pr_number, + branch, + target_branch, + parent_sha, + package_dir, + modified, + headline, + body, + ) if merge_sha is None: print( f"Bump PR {pr_url(repo, pr_number)} did not merge, so no " - f"{tag} tag was created and nothing was released. Merge the PR " - "(or re-run this job once it can merge) to publish " - f"{package_name} {new_version}." + f"{tag} tag was created and nothing was released. Re-run this job " + f"once the PR can merge to publish {package_name} {new_version}; if " + "the PR has merged since, the re-run pushes the missing tag." ) return False if not create_and_push_tag(repo, tag, merge_sha): return False + delete_remote_branch(repo, branch) print(f"Successfully bumped {package_name} to {new_version}; tag {tag} pushed.") return True def main() -> None: parser = argparse.ArgumentParser( - description="Bump a package version via an auto-merging PR." + description="Bump a package version via a PR fast-forwarded by the release app." ) parser.add_argument("package_name", help="Package name (e.g. keycardai-oauth).") parser.add_argument("package_dir", help="Package directory (e.g. packages/oauth).") diff --git a/scripts/test_bump_package.py b/scripts/test_bump_package.py index 4c21ad6..bcf2355 100644 --- a/scripts/test_bump_package.py +++ b/scripts/test_bump_package.py @@ -78,12 +78,10 @@ class PrBaseBranchTests(unittest.TestCase): side_effect=[ # gh pr create (0, "https://github.com/keycardai/python-sdk/pull/999", ""), - # gh pr merge --auto --squash - (0, "", ""), ], ) def test_pr_targets_the_release_branch(self, run_command, _stable) -> None: - pr_number = bump_package.create_pr_with_automerge( + pr_number = bump_package.create_bump_pr( "bump/release-mcp-v1/keycardai-mcp-1.0.1", "release/mcp-v1", "keycardai-mcp", @@ -93,46 +91,365 @@ def test_pr_targets_the_release_branch(self, run_command, _stable) -> None: command = run_command.call_args_list[0][0][0] base_index = command.index("--base") self.assertEqual(command[base_index + 1], "release/mcp-v1") + for call in run_command.call_args_list: + self.assertNotIn("--auto", call[0][0]) + + +BASE = "a" * 40 +HEAD = "b" * 40 +MOVED = "c" * 40 + + +def refs_api_calls(run_command: mock.Mock) -> list[list[str]]: + return [ + call[0][0] + for call in run_command.call_args_list + if any(arg.startswith("repos/") and "git/refs" in arg for arg in call[0][0]) + ] -class MergeRefusalTests(unittest.TestCase): - """A refused merge must fail the run rather than force the release.""" +class ChecksGateTests(unittest.TestCase): + """The merge path only opens once every check has concluded successfully.""" - PR_STATUS = json.dumps( - { - "state": "OPEN", - "mergeCommit": None, - "statusCheckRollup": [{"status": "COMPLETED", "conclusion": "SUCCESS"}], + def test_verdict_prefers_failure_over_pending(self) -> None: + data = { + "statusCheckRollup": [ + {"name": "socket-scan", "status": "IN_PROGRESS", "conclusion": None}, + {"name": "lint-and-test", "status": "COMPLETED", "conclusion": "FAILURE"}, + ] } - ) + self.assertEqual( + bump_package.checks_verdict(data), ("FAILURE", "lint-and-test") + ) + + def test_verdict_is_pending_while_any_check_runs(self) -> None: + data = { + "statusCheckRollup": [ + {"name": "lint-and-test", "status": "COMPLETED", "conclusion": "SUCCESS"}, + {"name": "release-preview", "status": "QUEUED", "conclusion": None}, + ] + } + self.assertEqual( + bump_package.checks_verdict(data), ("PENDING", "release-preview") + ) + + def test_unregistered_rollup_is_pending_not_success(self) -> None: + # A fresh commit (a just-opened PR, or a just-rebuilt bump branch) has + # no registered check runs; that must read as pending, never passing. + self.assertEqual( + bump_package.checks_verdict({"statusCheckRollup": []}), ("PENDING", None) + ) + self.assertEqual( + bump_package.checks_verdict({"statusCheckRollup": None}), ("PENDING", None) + ) + + @mock.patch.object(bump_package.time, "sleep") + @mock.patch.object(bump_package.time, "time", return_value=0) + @mock.patch.object(bump_package, "run_command") + def test_failed_check_refuses_without_touching_refs( + self, run_command, _time, _sleep + ) -> None: + run_command.return_value = ( + 0, + json.dumps( + { + "state": "OPEN", + "headRefOid": HEAD, + "statusCheckRollup": [ + {"name": "lint-and-test", "status": "COMPLETED", "conclusion": "FAILURE"} + ], + } + ), + "", + ) + + outcome = bump_package.wait_for_checks("keycardai/python-sdk", 250) + + self.assertIsNone(outcome) + self.assertEqual(refs_api_calls(run_command), []) + + @mock.patch.object(bump_package.time, "sleep") + @mock.patch.object(bump_package.time, "time", side_effect=[0, 10_000]) + @mock.patch.object(bump_package, "run_command") + def test_pending_checks_time_out_instead_of_passing( + self, run_command, _time, _sleep + ) -> None: + run_command.return_value = ( + 0, + json.dumps( + { + "state": "OPEN", + "headRefOid": HEAD, + "statusCheckRollup": [ + {"name": "lint-and-test", "status": "IN_PROGRESS", "conclusion": None} + ], + } + ), + "", + ) + + outcome = bump_package.wait_for_checks( + "keycardai/python-sdk", 250, timeout_seconds=1 + ) + + self.assertIsNone(outcome) @mock.patch.object(bump_package.time, "sleep") - @mock.patch.object(bump_package.time, "time", side_effect=[0, 0, 10, 10_000]) + @mock.patch.object(bump_package.time, "time", side_effect=[0, 10_000]) @mock.patch.object(bump_package, "run_command") - def test_refused_merge_never_updates_the_target_ref( + def test_fresh_commit_with_no_registered_checks_times_out_instead_of_merging( self, run_command, _time, _sleep + ) -> None: + # The read that races Actions' check registration: a rebuilt bump + # branch (or a PR read too early) reports an empty rollup. + run_command.return_value = ( + 0, + json.dumps( + {"state": "OPEN", "headRefOid": HEAD, "statusCheckRollup": []} + ), + "", + ) + + outcome = bump_package.wait_for_checks( + "keycardai/python-sdk", 250, timeout_seconds=1 + ) + + self.assertIsNone(outcome) + self.assertEqual(refs_api_calls(run_command), []) + + @mock.patch.object(bump_package, "fast_forward_target") + @mock.patch.object(bump_package, "wait_for_checks", return_value=None) + def test_merge_is_unreachable_when_checks_gate_refuses( + self, _checks, fast_forward_target + ) -> None: + merge_sha = bump_package.merge_bump_pr( + "keycardai/python-sdk", 250, "bump/main/keycardai-mcp-2.2.0", "main", + BASE, "packages/mcp", ["pyproject.toml"], + "bump: keycardai-mcp → 2.2.0", "Auto-bump.", + ) + self.assertIsNone(merge_sha) + fast_forward_target.assert_not_called() + + +class StrictFastForwardTests(unittest.TestCase): + """The target ref is only ever fast-forwarded from the SHA the bump was built on.""" + + @mock.patch.object(bump_package, "verify_pr_merged") + @mock.patch.object(bump_package, "rebase_bump_branch", return_value=False) + @mock.patch.object(bump_package, "get_live_branch_sha", return_value=MOVED) + @mock.patch.object(bump_package, "wait_for_checks", return_value={"head": HEAD}) + @mock.patch.object(bump_package, "run_command") + def test_guard_refuses_when_target_moved( + self, run_command, _checks, _live, rebase_bump_branch, verify_pr_merged + ) -> None: + merge_sha = bump_package.merge_bump_pr( + "keycardai/python-sdk", 250, "bump/main/keycardai-mcp-2.2.0", "main", + BASE, "packages/mcp", ["pyproject.toml"], + "bump: keycardai-mcp → 2.2.0", "Auto-bump.", + ) + + self.assertIsNone(merge_sha) + self.assertEqual(refs_api_calls(run_command), []) + rebase_bump_branch.assert_called_once_with( + "keycardai/python-sdk", "bump/main/keycardai-mcp-2.2.0", BASE, MOVED, + "packages/mcp", ["pyproject.toml"], + "bump: keycardai-mcp → 2.2.0", "Auto-bump.", + ) + verify_pr_merged.assert_not_called() + + @mock.patch.object(bump_package, "verify_pr_merged", return_value=HEAD) + @mock.patch.object(bump_package, "get_live_branch_sha", return_value=BASE) + @mock.patch.object(bump_package, "wait_for_checks", return_value={"head": HEAD}) + @mock.patch.object(bump_package, "run_command", return_value=(0, "{}", "")) + def test_fast_forward_never_forces_the_target_ref( + self, run_command, _checks, _live, _verify + ) -> None: + merge_sha = bump_package.merge_bump_pr( + "keycardai/python-sdk", 250, "bump/main/keycardai-mcp-2.2.0", "main", + BASE, "packages/mcp", ["pyproject.toml"], + "bump: keycardai-mcp → 2.2.0", "Auto-bump.", + ) + + self.assertEqual(merge_sha, HEAD) + calls = refs_api_calls(run_command) + self.assertEqual(len(calls), 1) + command = calls[0] + self.assertIn("PATCH", command) + self.assertIn("repos/keycardai/python-sdk/git/refs/heads/main", command) + self.assertIn(f"sha={HEAD}", command) + self.assertFalse(any("force" in arg for arg in command), command) + + @mock.patch.object(bump_package, "create_signed_commit_on_branch", return_value=True) + @mock.patch.object(bump_package, "run_command") + def test_rebuild_refuses_when_target_changed_bumped_files( + self, run_command, create_signed_commit_on_branch + ) -> None: + run_command.return_value = (0, "packages/mcp/pyproject.toml\nREADME.md", "") + + rebuilt = bump_package.rebase_bump_branch( + "keycardai/python-sdk", "bump/main/keycardai-mcp-2.2.0", BASE, MOVED, + "packages/mcp", + ["packages/mcp/pyproject.toml", "packages/mcp/CHANGELOG.md"], + "bump: keycardai-mcp → 2.2.0", "Auto-bump.", + ) + + self.assertFalse(rebuilt) + self.assertEqual(refs_api_calls(run_command), []) + create_signed_commit_on_branch.assert_not_called() + + @mock.patch.object(bump_package, "create_signed_commit_on_branch", return_value=True) + @mock.patch.object(bump_package, "run_command") + def test_rebuild_refuses_when_target_changed_package_sources( + self, run_command, create_signed_commit_on_branch + ) -> None: + # A same-package commit that merged mid-run must never ship under the + # stale version and changelog, even when it left the bumped files alone. + run_command.return_value = (0, "packages/mcp/src/client.py\nREADME.md", "") + + rebuilt = bump_package.rebase_bump_branch( + "keycardai/python-sdk", "bump/main/keycardai-mcp-2.2.0", BASE, MOVED, + "packages/mcp", + ["packages/mcp/pyproject.toml", "packages/mcp/CHANGELOG.md"], + "bump: keycardai-mcp → 2.2.0", "Auto-bump.", + ) + + self.assertFalse(rebuilt) + self.assertEqual(refs_api_calls(run_command), []) + create_signed_commit_on_branch.assert_not_called() + + @mock.patch.object(bump_package, "create_signed_commit_on_branch", return_value=True) + @mock.patch.object(bump_package, "run_command") + def test_rebuild_proceeds_when_target_changes_are_unrelated( + self, run_command, create_signed_commit_on_branch ) -> None: run_command.side_effect = [ - (0, self.PR_STATUS, ""), - (1, "", "the base branch policy prohibits the merge"), + (0, "README.md\ndocs/guide.md", ""), # compare old base...new base + (0, "", ""), # force-move the bump branch ] - merge_sha = bump_package.wait_for_pr_merge( - "keycardai/python-sdk", 250, "main", timeout_seconds=1 + rebuilt = bump_package.rebase_bump_branch( + "keycardai/python-sdk", "bump/main/keycardai-mcp-2.2.0", BASE, MOVED, + "packages/mcp", + ["packages/mcp/pyproject.toml", "packages/mcp/CHANGELOG.md"], + "bump: keycardai-mcp → 2.2.0", "Auto-bump.", ) - self.assertIsNone(merge_sha) - for call in run_command.call_args_list: - command = call[0][0] - self.assertNotIn("PATCH", command) - self.assertFalse( - any(arg.startswith("repos/") and "git/refs" in arg for arg in command), - f"the refs API must not be touched: {command}", + self.assertTrue(rebuilt) + calls = refs_api_calls(run_command) + self.assertEqual(len(calls), 1) + command = calls[0] + self.assertIn( + "repos/keycardai/python-sdk/git/refs/heads/bump/main/keycardai-mcp-2.2.0", + command, + ) + self.assertIn("force=true", command) + create_signed_commit_on_branch.assert_called_once() + + @mock.patch.object(bump_package, "verify_pr_merged", return_value=HEAD) + @mock.patch.object(bump_package, "fast_forward_target", return_value=True) + @mock.patch.object(bump_package, "wait_for_pr_stable", return_value=True) + @mock.patch.object(bump_package, "rebase_bump_branch", return_value=True) + @mock.patch.object(bump_package, "get_live_branch_sha", side_effect=[MOVED, MOVED]) + @mock.patch.object(bump_package, "wait_for_checks", return_value={"head": HEAD}) + def test_rebuild_reenters_the_stable_wait_before_rereading_checks( + self, wait_for_checks, _live, _rebase, wait_for_pr_stable, _ff, _verify + ) -> None: + merge_sha = bump_package.merge_bump_pr( + "keycardai/python-sdk", 250, "bump/main/keycardai-mcp-2.2.0", "main", + BASE, "packages/mcp", ["pyproject.toml"], + "bump: keycardai-mcp → 2.2.0", "Auto-bump.", + ) + + self.assertEqual(merge_sha, HEAD) + wait_for_pr_stable.assert_called_once_with(250) + self.assertEqual(wait_for_checks.call_count, 2) + + @mock.patch.object(bump_package, "verify_pr_merged", return_value=HEAD) + @mock.patch.object(bump_package, "wait_for_pr_stable", return_value=True) + @mock.patch.object(bump_package, "rebase_bump_branch", return_value=True) + @mock.patch.object(bump_package, "fast_forward_target", side_effect=[False, True]) + @mock.patch.object( + bump_package, "get_live_branch_sha", side_effect=[BASE, MOVED, MOVED] + ) + @mock.patch.object(bump_package, "wait_for_checks", return_value={"head": HEAD}) + def test_ff_refusal_reprobes_the_live_tip_and_rebuilds( + self, _checks, _live, fast_forward_target, rebase_bump_branch, _stable, _verify + ) -> None: + # The guard window race: main moves between the live read and the + # PATCH, GitHub 422s the strict update, and the run recovers by + # rebuilding on the new tip instead of aborting. + merge_sha = bump_package.merge_bump_pr( + "keycardai/python-sdk", 250, "bump/main/keycardai-mcp-2.2.0", "main", + BASE, "packages/mcp", ["pyproject.toml"], + "bump: keycardai-mcp → 2.2.0", "Auto-bump.", + ) + + self.assertEqual(merge_sha, HEAD) + rebase_bump_branch.assert_called_once() + self.assertEqual(fast_forward_target.call_count, 2) + + +class MergeVerificationTests(unittest.TestCase): + """No tag is ever created unless GitHub reports the PR merged at the head.""" + + @mock.patch.object(bump_package.time, "sleep") + @mock.patch.object(bump_package.time, "time", side_effect=[0, 10_000]) + @mock.patch.object(bump_package, "run_command") + def test_unmerged_pr_after_fast_forward_returns_none( + self, run_command, _time, _sleep + ) -> None: + run_command.return_value = ( + 0, + json.dumps({"state": "OPEN", "mergeCommit": None}), + "", + ) + + self.assertIsNone( + bump_package.verify_pr_merged( + "keycardai/python-sdk", 250, HEAD, timeout_seconds=1 ) + ) + + @mock.patch.object(bump_package, "run_command") + def test_merge_at_unexpected_sha_returns_none(self, run_command) -> None: + run_command.return_value = ( + 0, + json.dumps({"state": "MERGED", "mergeCommit": {"oid": MOVED}}), + "", + ) + + self.assertIsNone( + bump_package.verify_pr_merged("keycardai/python-sdk", 250, HEAD) + ) + + @mock.patch.object(bump_package, "create_and_push_tag") + @mock.patch.object(bump_package, "verify_pr_merged", return_value=None) + @mock.patch.object(bump_package, "fast_forward_target", return_value=True) + @mock.patch.object(bump_package, "get_live_branch_sha", return_value=BASE) + @mock.patch.object(bump_package, "wait_for_checks", return_value={"head": HEAD}) + @mock.patch.object(bump_package, "create_bump_pr", return_value=250) + @mock.patch.object(bump_package, "create_signed_commit_on_branch", return_value=True) + @mock.patch.object(bump_package, "create_remote_branch", return_value=True) + @mock.patch.object(bump_package, "get_modified_files", return_value=["pyproject.toml"]) + @mock.patch.object(bump_package, "get_branch_sha", return_value=BASE) + @mock.patch.object(bump_package, "cz_bump_files_only", return_value="2.2.0") + @mock.patch.object(bump_package, "recover_untagged_bump", return_value=None) + @mock.patch.object(bump_package, "get_repo_slug", return_value="keycardai/python-sdk") + @mock.patch.object(bump_package, "pull_branch", return_value=True) + @mock.patch.object(bump_package, "configure_git") + def test_tag_is_unreachable_when_merge_verification_fails( + self, *mocks, **_kwargs + ) -> None: + create_and_push_tag = mocks[-1] + + self.assertFalse( + bump_package.bump_package("keycardai-mcp", "packages/mcp") + ) + create_and_push_tag.assert_not_called() @mock.patch.object(bump_package, "create_and_push_tag") - @mock.patch.object(bump_package, "wait_for_pr_merge", return_value=None) - @mock.patch.object(bump_package, "create_pr_with_automerge", return_value=250) + @mock.patch.object(bump_package, "merge_bump_pr", return_value=None) + @mock.patch.object(bump_package, "create_bump_pr", return_value=250) @mock.patch.object(bump_package, "create_signed_commit_on_branch", return_value=True) @mock.patch.object(bump_package, "create_remote_branch", return_value=True) @mock.patch.object(bump_package, "get_modified_files", return_value=["pyproject.toml"]) @@ -153,5 +470,145 @@ def test_unmerged_bump_pr_fails_the_run_without_tagging( create_and_push_tag.assert_not_called() +class ExternalMergeTests(unittest.TestCase): + """A PR merged by someone else mid-wait is adopted, not treated as failure.""" + + @mock.patch.object(bump_package, "run_command") + def test_human_merge_during_checks_wait_is_adopted(self, run_command) -> None: + run_command.return_value = ( + 0, + json.dumps( + { + "state": "MERGED", + "mergeCommit": {"oid": MOVED}, + "statusCheckRollup": [], + } + ), + "", + ) + + outcome = bump_package.wait_for_checks("keycardai/python-sdk", 250) + + self.assertEqual(outcome, {"merged": MOVED}) + + @mock.patch.object(bump_package, "verify_pr_merged") + @mock.patch.object(bump_package, "fast_forward_target") + @mock.patch.object(bump_package, "wait_for_checks", return_value={"merged": MOVED}) + def test_external_merge_skips_the_ref_update_and_returns_its_sha( + self, _checks, fast_forward_target, verify_pr_merged + ) -> None: + merge_sha = bump_package.merge_bump_pr( + "keycardai/python-sdk", 250, "bump/main/keycardai-mcp-2.2.0", "main", + BASE, "packages/mcp", ["pyproject.toml"], + "bump: keycardai-mcp → 2.2.0", "Auto-bump.", + ) + + self.assertEqual(merge_sha, MOVED) + fast_forward_target.assert_not_called() + verify_pr_merged.assert_not_called() + + @mock.patch.object(bump_package.time, "sleep") + @mock.patch.object(bump_package, "run_command") + def test_closed_pr_aborts_the_wait(self, run_command, _sleep) -> None: + run_command.return_value = (0, json.dumps({"state": "CLOSED"}), "") + + self.assertIsNone(bump_package.wait_for_checks("keycardai/python-sdk", 250)) + + +class BranchLifecycleTests(unittest.TestCase): + """A stale bump branch is reused, and a released one is deleted.""" + + @mock.patch.object(bump_package, "run_command") + def test_stale_branch_from_failed_run_is_force_moved_not_fatal( + self, run_command + ) -> None: + run_command.side_effect = [ + (1, "", "HTTP 422: Reference already exists"), + (0, "", ""), + ] + + self.assertTrue( + bump_package.create_remote_branch( + "keycardai/python-sdk", "bump/main/keycardai-mcp-2.2.0", BASE + ) + ) + recover = run_command.call_args_list[1][0][0] + self.assertIn("PATCH", recover) + self.assertIn( + "repos/keycardai/python-sdk/git/refs/heads/bump/main/keycardai-mcp-2.2.0", + recover, + ) + self.assertIn(f"sha={BASE}", recover) + self.assertIn("force=true", recover) + + @mock.patch.object( + bump_package, "run_command", return_value=(1, "", "HTTP 403 Forbidden") + ) + def test_other_branch_creation_errors_still_fail(self, run_command) -> None: + self.assertFalse( + bump_package.create_remote_branch( + "keycardai/python-sdk", "bump/main/keycardai-mcp-2.2.0", BASE + ) + ) + self.assertEqual(len(run_command.call_args_list), 1) + + @mock.patch.object(bump_package, "wait_for_pr_stable", return_value=True) + @mock.patch.object( + bump_package, + "run_command", + return_value=( + 1, + "", + 'a pull request for branch "bump/main/keycardai-mcp-2.2.0" into branch ' + '"main" already exists:\nhttps://github.com/keycardai/python-sdk/pull/311', + ), + ) + def test_existing_open_pr_is_reused(self, _run_command, _stable) -> None: + pr_number = bump_package.create_bump_pr( + "bump/main/keycardai-mcp-2.2.0", "main", "keycardai-mcp", "2.2.0" + ) + self.assertEqual(pr_number, 311) + + @mock.patch.object(bump_package, "delete_remote_branch") + @mock.patch.object(bump_package, "create_and_push_tag", return_value=True) + @mock.patch.object(bump_package, "merge_bump_pr", return_value=HEAD) + @mock.patch.object(bump_package, "create_bump_pr", return_value=250) + @mock.patch.object(bump_package, "create_signed_commit_on_branch", return_value=True) + @mock.patch.object(bump_package, "create_remote_branch", return_value=True) + @mock.patch.object(bump_package, "get_modified_files", return_value=["pyproject.toml"]) + @mock.patch.object(bump_package, "get_branch_sha", return_value=BASE) + @mock.patch.object(bump_package, "cz_bump_files_only", return_value="2.2.0") + @mock.patch.object(bump_package, "recover_untagged_bump", return_value=None) + @mock.patch.object(bump_package, "get_repo_slug", return_value="keycardai/python-sdk") + @mock.patch.object(bump_package, "pull_branch", return_value=True) + @mock.patch.object(bump_package, "configure_git") + def test_branch_is_deleted_after_verified_merge_and_tag(self, *mocks) -> None: + delete_remote_branch = mocks[-1] + + self.assertTrue(bump_package.bump_package("keycardai-mcp", "packages/mcp")) + delete_remote_branch.assert_called_once_with( + "keycardai/python-sdk", "bump/main/keycardai-mcp-2.2.0" + ) + + @mock.patch.object(bump_package, "delete_remote_branch") + @mock.patch.object(bump_package, "create_and_push_tag", return_value=False) + @mock.patch.object(bump_package, "merge_bump_pr", return_value=HEAD) + @mock.patch.object(bump_package, "create_bump_pr", return_value=250) + @mock.patch.object(bump_package, "create_signed_commit_on_branch", return_value=True) + @mock.patch.object(bump_package, "create_remote_branch", return_value=True) + @mock.patch.object(bump_package, "get_modified_files", return_value=["pyproject.toml"]) + @mock.patch.object(bump_package, "get_branch_sha", return_value=BASE) + @mock.patch.object(bump_package, "cz_bump_files_only", return_value="2.2.0") + @mock.patch.object(bump_package, "recover_untagged_bump", return_value=None) + @mock.patch.object(bump_package, "get_repo_slug", return_value="keycardai/python-sdk") + @mock.patch.object(bump_package, "pull_branch", return_value=True) + @mock.patch.object(bump_package, "configure_git") + def test_branch_survives_a_failed_tag_push(self, *mocks) -> None: + delete_remote_branch = mocks[-1] + + self.assertFalse(bump_package.bump_package("keycardai-mcp", "packages/mcp")) + delete_remote_branch.assert_not_called() + + if __name__ == "__main__": unittest.main()