feat(release): add retry mode to re-deploy existing release without re-tagging

Recovers from a release whose tag+commit landed cleanly but whose
post-tag steps (git push, latest-tag bump, twine upload) failed
mid-flight. pkgmgr release --retry skips the version bump, file
rewrites, commit, and tag-creation steps and re-runs only the
idempotent tail: re-push the existing HEAD tag, re-align the floating
latest tag, and (unless --no-publish) re-invoke publish.

The retry logic lives in its own module pkgmgr.actions.release.retry
so the workflow.py orchestrator stays focused on the forward path.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-05-27 20:52:06 +02:00
parent a6c40451fe
commit 43fbcfb227
5 changed files with 259 additions and 3 deletions

View File

@@ -0,0 +1,67 @@
"""Re-deploy an existing release without modifying files or creating a new tag.
The release workflow normally bumps versions, rewrites packaging
manifests, commits, tags, pushes, and uploads to PyPI in one shot.
When a post-tag step fails mid-flight (typical examples: `git push`
rejected, `twine upload` aborted by a broken venv, `update-latest`
rejected by branch protection) the local tag still exists on HEAD but
the side effects downstream are incomplete.
`retry_release` re-runs the idempotent tail of that flow so a botched
release can be re-pushed without touching code or recreating tags:
* `git push origin <branch> <tag>` for the existing HEAD tag
* re-align the floating `latest` tag if HEAD tag is the highest
Publishing (PyPI etc.) stays the caller's responsibility — the publish
workflow is already idempotent (twine rejects duplicates per spec) and
can be invoked independently via the `publish` subcommand.
"""
from __future__ import annotations
import os
from pkgmgr.actions.publish.git_tags import head_semver_tags
from pkgmgr.core.git import GitRunError, run
from pkgmgr.core.git.queries import get_current_branch
from pkgmgr.core.version.semver import SemVer
from .git_ops import is_highest_version_tag, update_latest_tag
def retry_release(
pyproject_path: str = "pyproject.toml",
preview: bool = False,
) -> None:
"""Re-push the HEAD release without re-tagging or modifying any files."""
try:
branch = get_current_branch() or "main"
except GitRunError:
branch = "main"
print(f"Retrying release push on branch: {branch}")
tags = head_semver_tags(cwd=os.path.dirname(os.path.abspath(pyproject_path)))
if not tags:
raise RuntimeError(
"No version tag on HEAD. Nothing to retry — "
"run `pkgmgr release <type>` first to create a release."
)
tag = max(tags, key=SemVer.parse)
print(f"Re-pushing existing tag: {tag}")
run(["push", "origin", branch, tag], preview=preview)
try:
if is_highest_version_tag(tag):
update_latest_tag(tag, preview=preview)
else:
print(f"[INFO] Skipping 'latest' update (tag {tag} is not the highest).")
except GitRunError as exc:
print(f"[WARN] Failed to update floating 'latest' tag for {tag}: {exc}")
if preview:
print(f"[PREVIEW] Retry push for {tag} would now complete.")
return
print(f"Retry push completed for {tag}.")

View File

@@ -26,6 +26,7 @@ from .git_ops import (
) )
from .package_name import resolve_package_name from .package_name import resolve_package_name
from .prompts import confirm_proceed_release, should_delete_branch from .prompts import confirm_proceed_release, should_delete_branch
from .retry import retry_release
from .versioning import bump_semver, determine_current_version from .versioning import bump_semver, determine_current_version
@@ -199,7 +200,12 @@ def release(
preview: bool = False, preview: bool = False,
force: bool = False, force: bool = False,
close: bool = False, close: bool = False,
retry: bool = False,
) -> None: ) -> None:
if retry:
retry_release(pyproject_path=pyproject_path, preview=preview)
return
if preview: if preview:
_release_impl( _release_impl(
pyproject_path=pyproject_path, pyproject_path=pyproject_path,

View File

@@ -49,7 +49,19 @@ def handle_release(
print(f"[WARN] Skipping repository {identifier}: directory missing.") print(f"[WARN] Skipping repository {identifier}: directory missing.")
continue continue
print(f"[pkgmgr] Running release for repository {identifier}...") retry = bool(getattr(args, "retry", False))
if retry and args.release_type:
print(
f"[WARN] Ignoring release_type '{args.release_type}' for {identifier} — --retry skips version bumps."
)
if not retry and not args.release_type:
print(
f"[WARN] Skipping {identifier}: release_type is required unless --retry is set."
)
continue
action_label = "retry-release" if retry else "release"
print(f"[pkgmgr] Running {action_label} for repository {identifier}...")
cwd_before = os.getcwd() cwd_before = os.getcwd()
try: try:
@@ -58,11 +70,12 @@ def handle_release(
run_release( run_release(
pyproject_path="pyproject.toml", pyproject_path="pyproject.toml",
changelog_path="CHANGELOG.md", changelog_path="CHANGELOG.md",
release_type=args.release_type, release_type=args.release_type or "patch",
message=args.message or None, message=args.message or None,
preview=getattr(args, "preview", False), preview=getattr(args, "preview", False),
force=getattr(args, "force", False), force=getattr(args, "force", False),
close=getattr(args, "close", False), close=getattr(args, "close", False),
retry=retry,
) )
if not getattr(args, "no_publish", False): if not getattr(args, "no_publish", False):

View File

@@ -24,8 +24,12 @@ def add_release_subparser(
release_parser.add_argument( release_parser.add_argument(
"release_type", "release_type",
nargs="?",
choices=["major", "minor", "patch"], choices=["major", "minor", "patch"],
help="Type of version increment for the release (major, minor, patch).", help=(
"Type of version increment for the release (major, minor, patch). "
"Omit when `--retry` is set."
),
) )
release_parser.add_argument( release_parser.add_argument(
@@ -61,3 +65,15 @@ def add_release_subparser(
action="store_true", action="store_true",
help="Do not run publish automatically after a successful release.", help="Do not run publish automatically after a successful release.",
) )
release_parser.add_argument(
"--retry",
action="store_true",
help=(
"Re-deploy the existing HEAD release without re-tagging or "
"modifying any files: re-push the existing version tag, "
"re-align the floating `latest` tag, and (unless --no-publish) "
"re-run publish. Use this to recover from a release whose "
"post-tag push or PyPI upload failed mid-flight."
),
)

View File

@@ -0,0 +1,154 @@
"""Unit tests for the `release(retry=True)` re-deploy entry point.
`TestReleaseRetry` exercises the routing decision in `release()` —
that `--retry` skips the full `_release_impl` flow and dispatches to
`retry_release` instead.
`TestRetryRelease` exercises the standalone module `pkgmgr.actions.release.retry`:
HEAD-tag discovery, idempotent push, latest-tag re-alignment, and
fallback behaviour when the current branch cannot be detected.
"""
from __future__ import annotations
import unittest
from unittest.mock import patch
from pkgmgr.actions.release.retry import retry_release
from pkgmgr.actions.release.workflow import release
from pkgmgr.core.git import GitRunError
class TestReleaseRetry(unittest.TestCase):
@patch("pkgmgr.actions.release.workflow.retry_release")
@patch("pkgmgr.actions.release.workflow._release_impl")
def test_retry_routes_to_retry_release_only(
self, mock_release_impl, mock_retry_release
) -> None:
release(retry=True, preview=False)
mock_retry_release.assert_called_once()
mock_release_impl.assert_not_called()
self.assertFalse(mock_retry_release.call_args.kwargs["preview"])
@patch("pkgmgr.actions.release.workflow.retry_release")
@patch("pkgmgr.actions.release.workflow._release_impl")
def test_retry_preview_passthrough(
self, mock_release_impl, mock_retry_release
) -> None:
release(retry=True, preview=True)
mock_retry_release.assert_called_once()
mock_release_impl.assert_not_called()
self.assertTrue(mock_retry_release.call_args.kwargs["preview"])
@patch("pkgmgr.actions.release.workflow.retry_release")
@patch("pkgmgr.actions.release.workflow._release_impl")
def test_retry_ignores_release_args(
self, mock_release_impl, mock_retry_release
) -> None:
# release_type/message/close/force must NOT trigger the full release flow
# when retry=True.
release(
retry=True,
release_type="major",
message="ignored body",
preview=False,
force=True,
close=True,
)
mock_retry_release.assert_called_once()
mock_release_impl.assert_not_called()
class TestRetryRelease(unittest.TestCase):
@patch("pkgmgr.actions.release.retry.update_latest_tag")
@patch("pkgmgr.actions.release.retry.is_highest_version_tag", return_value=True)
@patch("pkgmgr.actions.release.retry.run")
@patch("pkgmgr.actions.release.retry.head_semver_tags")
@patch(
"pkgmgr.actions.release.retry.get_current_branch",
return_value="release-fix",
)
def test_retry_pushes_existing_tag_and_updates_latest(
self,
_mock_branch,
mock_head_tags,
mock_run,
_mock_highest,
mock_update_latest,
) -> None:
mock_head_tags.return_value = ["v1.13.4"]
retry_release(pyproject_path="pyproject.toml", preview=False)
mock_run.assert_called_once_with(
["push", "origin", "release-fix", "v1.13.4"], preview=False
)
mock_update_latest.assert_called_once_with("v1.13.4", preview=False)
@patch("pkgmgr.actions.release.retry.update_latest_tag")
@patch("pkgmgr.actions.release.retry.is_highest_version_tag", return_value=False)
@patch("pkgmgr.actions.release.retry.run")
@patch("pkgmgr.actions.release.retry.head_semver_tags")
@patch(
"pkgmgr.actions.release.retry.get_current_branch",
return_value="release-fix",
)
def test_retry_skips_latest_for_non_highest_tag(
self,
_mock_branch,
mock_head_tags,
_mock_run,
_mock_highest,
mock_update_latest,
) -> None:
mock_head_tags.return_value = ["v1.0.0"]
retry_release(pyproject_path="pyproject.toml", preview=False)
mock_update_latest.assert_not_called()
@patch(
"pkgmgr.actions.release.retry.head_semver_tags",
return_value=[],
)
@patch(
"pkgmgr.actions.release.retry.get_current_branch",
return_value="main",
)
def test_retry_raises_when_head_has_no_tag(
self, _mock_branch, _mock_head_tags
) -> None:
with self.assertRaises(RuntimeError) as cm:
retry_release(pyproject_path="pyproject.toml", preview=False)
self.assertIn("No version tag on HEAD", str(cm.exception))
@patch("pkgmgr.actions.release.retry.update_latest_tag")
@patch("pkgmgr.actions.release.retry.is_highest_version_tag", return_value=True)
@patch("pkgmgr.actions.release.retry.run")
@patch("pkgmgr.actions.release.retry.head_semver_tags")
@patch(
"pkgmgr.actions.release.retry.get_current_branch",
side_effect=GitRunError("detached"),
)
def test_retry_falls_back_to_main_branch_when_detection_fails(
self,
_mock_branch,
mock_head_tags,
mock_run,
_mock_highest,
_mock_update_latest,
) -> None:
mock_head_tags.return_value = ["v2.0.0"]
retry_release(pyproject_path="pyproject.toml", preview=False)
mock_run.assert_called_once_with(
["push", "origin", "main", "v2.0.0"], preview=False
)
if __name__ == "__main__":
unittest.main()