Coverage for src / rhiza_tools / commands / rollback.py: 100%

273 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-04-28 02:21 +0000

1"""Command to rollback a release and/or version bump. 

2 

3This module implements rollback functionality that safely reverses release 

4and bump operations. It can delete local and remote tags, revert bump commits, 

5and restore the project to a previous version state. 

6 

7Example: 

8 Rollback the most recent release:: 

9 

10 from rhiza_tools.commands.rollback import rollback_command 

11 rollback_command() 

12 

13 Rollback a specific tag:: 

14 

15 rollback_command(tag="v1.2.3") 

16 

17 Dry run to preview rollback:: 

18 

19 rollback_command(dry_run=True) 

20""" 

21 

22from __future__ import annotations 

23 

24from dataclasses import dataclass 

25 

26import questionary as qs 

27import typer 

28from loguru import logger 

29 

30from rhiza_tools import console 

31from rhiza_tools.commands._shared import ( 

32 COOL_STYLE, 

33 NON_INTERACTIVE_ERRORS, 

34 run_git_command, 

35 validate_pyproject_exists, 

36) 

37from rhiza_tools.commands.release import check_tag_exists 

38 

39 

40@dataclass 

41class RollbackOptions: 

42 """Configuration options for the rollback command. 

43 

44 Attributes: 

45 tag: The tag to rollback (e.g., "v1.2.3"). None for interactive selection. 

46 revert_bump: If True, also revert the version bump commit. 

47 dry_run: If True, show what would change without actually changing anything. 

48 non_interactive: If True, skip all confirmation prompts. 

49 """ 

50 

51 tag: str | None = None 

52 revert_bump: bool = False 

53 dry_run: bool = False 

54 non_interactive: bool = False 

55 

56 

57def _get_recent_tags(limit: int = 10) -> list[str]: 

58 """Get recent version tags sorted by version descending. 

59 

60 Args: 

61 limit: Maximum number of tags to return. 

62 

63 Returns: 

64 List of tag names (e.g., ["v1.2.3", "v1.2.2", ...]). 

65 """ 

66 result = run_git_command( 

67 ["git", "tag", "--sort=-version:refname", "-l", "v*"], 

68 check=False, 

69 ) 

70 if result.returncode != 0 or not result.stdout.strip(): 

71 return [] 

72 

73 tags = [t.strip() for t in result.stdout.strip().split("\n") if t.strip()] 

74 return tags[:limit] 

75 

76 

77def _select_tag_interactively(tags: list[str]) -> str: 

78 """Prompt the user to select a tag to rollback. 

79 

80 Args: 

81 tags: List of available tags. 

82 

83 Returns: 

84 The selected tag name. 

85 

86 Raises: 

87 typer.Exit: If user cancels selection or no tags are available. 

88 """ 

89 if not tags: 

90 console.error("No version tags found in the repository.") 

91 console.error("Nothing to rollback.") 

92 raise typer.Exit(code=1) 

93 

94 # Annotate tags with local/remote info 

95 choices: list[str] = [] 

96 for tag in tags: 

97 exists_locally, exists_remotely = check_tag_exists(tag) 

98 markers = [] 

99 if exists_locally: 

100 markers.append("local") 

101 if exists_remotely: 

102 markers.append("remote") 

103 status = ", ".join(markers) if markers else "missing" 

104 choices.append(f"{tag} ({status})") 

105 

106 try: 

107 choice = qs.select( 

108 "Select tag to rollback:", 

109 choices=choices, 

110 style=COOL_STYLE, 

111 ).ask() 

112 except NON_INTERACTIVE_ERRORS: 

113 logger.debug("Running in non-interactive environment") 

114 raise typer.Exit(code=1) from None 

115 

116 if not choice: 

117 console.info("Rollback cancelled by user.") 

118 raise typer.Exit(code=0) 

119 

120 # Extract tag name from choice string "v1.2.3 (local, remote)" 

121 return str(choice).split(" (")[0] 

122 

123 

124def _get_tag_commit(tag: str) -> str | None: 

125 """Get the commit hash that a tag points to. 

126 

127 Args: 

128 tag: The tag name. 

129 

130 Returns: 

131 The commit hash, or None if the tag doesn't exist locally. 

132 """ 

133 result = run_git_command(["git", "rev-list", "-n", "1", tag], check=False) 

134 if result.returncode != 0: 

135 return None 

136 return result.stdout.strip() 

137 

138 

139def _get_tag_details(tag: str) -> dict[str, str]: 

140 """Get details about a tag. 

141 

142 Args: 

143 tag: The tag name. 

144 

145 Returns: 

146 Dictionary with commit hash, date, and message. 

147 """ 

148 details: dict[str, str] = {} 

149 result = run_git_command( 

150 ["git", "show", "-s", "--format=%H|%ci|%s", tag], 

151 check=False, 

152 ) 

153 if result.returncode == 0 and result.stdout.strip(): 

154 parts = result.stdout.strip().split("|") 

155 if len(parts) == 3: 

156 details["hash"] = parts[0] 

157 details["date"] = parts[1] 

158 details["message"] = parts[2] 

159 return details 

160 

161 

162def _is_bump_commit(tag: str) -> bool: 

163 """Check if the commit the tag points to looks like a bump commit. 

164 

165 Bump commits typically have messages like "Bump version: X.Y.Z → A.B.C" 

166 or contain version-related keywords. 

167 

168 Args: 

169 tag: The tag name. 

170 

171 Returns: 

172 True if the tag's commit appears to be a bump commit. 

173 """ 

174 result = run_git_command( 

175 ["git", "log", "-1", "--format=%s", tag], 

176 check=False, 

177 ) 

178 if result.returncode != 0: 

179 return False 

180 

181 message = result.stdout.strip().lower() 

182 bump_keywords = ["bump version", "bump:", "version bump", "release version", "chore: bump"] 

183 return any(keyword in message for keyword in bump_keywords) 

184 

185 

186def _get_previous_version_from_tags(current_tag: str) -> str | None: 

187 """Find the previous version tag before the given tag. 

188 

189 Args: 

190 current_tag: The current tag being rolled back. 

191 

192 Returns: 

193 The previous tag name, or None if no previous tag exists. 

194 """ 

195 result = run_git_command( 

196 ["git", "tag", "--sort=-version:refname", "-l", "v*"], 

197 check=False, 

198 ) 

199 if result.returncode != 0 or not result.stdout.strip(): 

200 return None 

201 

202 tags = [t.strip() for t in result.stdout.strip().split("\n") if t.strip()] 

203 

204 try: 

205 idx = tags.index(current_tag) 

206 if idx + 1 < len(tags): 

207 return tags[idx + 1] 

208 except ValueError: 

209 pass 

210 

211 return None 

212 

213 

214def _delete_local_tag(tag: str, dry_run: bool) -> bool: 

215 """Delete a tag from the local repository. 

216 

217 Args: 

218 tag: The tag name to delete. 

219 dry_run: If True, only simulate deletion. 

220 

221 Returns: 

222 True if deletion succeeded (or would succeed in dry-run). 

223 """ 

224 if dry_run: 

225 console.info(f"[DRY-RUN] Would delete local tag: {tag}") 

226 return True 

227 

228 result = run_git_command(["git", "tag", "-d", tag], check=False) 

229 if result.returncode == 0: 

230 console.success(f"Deleted local tag: {tag}") 

231 return True 

232 else: 

233 console.error(f"Failed to delete local tag: {tag}") 

234 console.error(f"Error: {result.stderr}") 

235 return False 

236 

237 

238def _delete_remote_tag(tag: str, dry_run: bool) -> bool: 

239 """Delete a tag from the remote repository. 

240 

241 Args: 

242 tag: The tag name to delete. 

243 dry_run: If True, only simulate deletion. 

244 

245 Returns: 

246 True if deletion succeeded (or would succeed in dry-run). 

247 """ 

248 if dry_run: 

249 console.info(f"[DRY-RUN] Would delete remote tag: {tag}") 

250 return True 

251 

252 console.info(f"Deleting remote tag: {tag}...") 

253 result = run_git_command( 

254 ["git", "push", "origin", f":refs/tags/{tag}"], 

255 check=False, 

256 ) 

257 if result.returncode == 0: 

258 console.success(f"Deleted remote tag: {tag}") 

259 return True 

260 else: 

261 console.error(f"Failed to delete remote tag: {tag}") 

262 console.error(f"Error: {result.stderr}") 

263 return False 

264 

265 

266def _revert_bump_commit(commit_hash: str, dry_run: bool) -> bool: 

267 """Revert the version bump commit. 

268 

269 Creates a new revert commit rather than rewriting history, making 

270 this safe even when the commit has been pushed to remote. 

271 

272 Args: 

273 commit_hash: The commit hash to revert. 

274 dry_run: If True, only simulate the revert. 

275 

276 Returns: 

277 True if revert succeeded (or would succeed in dry-run). 

278 """ 

279 if dry_run: 

280 result = run_git_command( 

281 ["git", "log", "-1", "--format=%s", commit_hash], 

282 check=False, 

283 ) 

284 commit_msg = result.stdout.strip() if result.returncode == 0 else "unknown" 

285 console.info(f"[DRY-RUN] Would revert commit {commit_hash[:8]}: {commit_msg}") 

286 return True 

287 

288 console.info(f"Reverting bump commit {commit_hash[:8]}...") 

289 result = run_git_command( 

290 ["git", "revert", "--no-edit", commit_hash], 

291 check=False, 

292 ) 

293 if result.returncode == 0: 

294 console.success(f"Reverted bump commit: {commit_hash[:8]}") 

295 return True 

296 else: 

297 console.error(f"Failed to revert commit {commit_hash[:8]}") 

298 console.error(f"Error: {result.stderr}") 

299 console.error("You may need to resolve conflicts manually:") 

300 console.error(f" git revert {commit_hash[:8]}") 

301 return False 

302 

303 

304def _push_revert(dry_run: bool, non_interactive: bool) -> bool: 

305 """Push the revert commit to remote. 

306 

307 Args: 

308 dry_run: If True, only simulate the push. 

309 non_interactive: If True, skip confirmation prompt. 

310 

311 Returns: 

312 True if push succeeded (or would succeed in dry-run). 

313 """ 

314 if dry_run: 

315 console.info("[DRY-RUN] Would push revert commit to remote") 

316 return True 

317 

318 should_push = non_interactive 

319 if not non_interactive: 

320 try: 

321 should_push = qs.confirm( 

322 "Push revert commit to remote?", 

323 default=True, 

324 style=COOL_STYLE, 

325 ).ask() 

326 except NON_INTERACTIVE_ERRORS: 

327 logger.debug("Running in non-interactive environment, proceeding with push") 

328 should_push = True 

329 

330 if not should_push: 

331 console.info("Revert commit created locally but not pushed.") 

332 console.info("Push manually when ready: git push") 

333 return True 

334 

335 result = run_git_command(["git", "push"], check=False) 

336 if result.returncode == 0: 

337 console.success("Revert commit pushed to remote.") 

338 return True 

339 else: 

340 console.error("Failed to push revert commit.") 

341 console.error(f"Error: {result.stderr}") 

342 console.error("Push manually: git push") 

343 return False 

344 

345 

346def _show_rollback_plan( 

347 tag: str, 

348 exists_locally: bool, 

349 exists_remotely: bool, 

350 revert_bump: bool, 

351 is_bump: bool, 

352 previous_tag: str | None, 

353 tag_details: dict[str, str], 

354) -> None: 

355 """Display the rollback plan to the user. 

356 

357 Args: 

358 tag: Tag being rolled back. 

359 exists_locally: Whether the tag exists locally. 

360 exists_remotely: Whether the tag exists on remote. 

361 revert_bump: Whether to revert the bump commit. 

362 is_bump: Whether the tagged commit appears to be a bump commit. 

363 previous_tag: The previous version tag, if any. 

364 tag_details: Details about the tag (hash, date, message). 

365 """ 

366 header = typer.style("Rollback Plan", fg=typer.colors.YELLOW, bold=True) 

367 console.info(f"\n{'─' * 50}") 

368 console.info(f" {header}") 

369 console.info(f"{'─' * 50}") 

370 

371 tag_styled = typer.style(tag, fg=typer.colors.RED, bold=True) 

372 console.info(f"\n Tag to rollback: {tag_styled}") 

373 

374 if tag_details: 

375 console.info(f" Commit: {tag_details.get('hash', 'unknown')[:8]}") 

376 console.info(f" Date: {tag_details.get('date', 'unknown')}") 

377 console.info(f" Message: {tag_details.get('message', 'unknown')}") 

378 

379 console.info(f"\n {typer.style('Actions:', fg=typer.colors.CYAN, bold=True)}") 

380 

381 step = 1 

382 if exists_remotely: 

383 console.info(f" {step}. Delete remote tag: git push origin :refs/tags/{tag}") 

384 step += 1 

385 if exists_locally: 

386 console.info(f" {step}. Delete local tag: git tag -d {tag}") 

387 step += 1 

388 if revert_bump and is_bump: 

389 console.info(f" {step}. Revert bump commit (creates a new revert commit)") 

390 step += 1 

391 console.info(f" {step}. Push revert commit to remote") 

392 step += 1 

393 

394 if previous_tag: 

395 prev_styled = typer.style(previous_tag, fg=typer.colors.GREEN, bold=True) 

396 console.info(f"\n Previous version: {prev_styled}") 

397 else: 

398 console.info("\n No previous version tag found.") 

399 

400 console.info(f"\n{'─' * 50}") 

401 

402 

403def _confirm_rollback(non_interactive: bool) -> bool: 

404 """Confirm rollback with the user. 

405 

406 Args: 

407 non_interactive: If True, skip confirmation. 

408 

409 Returns: 

410 True if user confirms (or non-interactive mode). 

411 """ 

412 if non_interactive: 

413 return True 

414 

415 try: 

416 return bool( 

417 qs.confirm( 

418 "Proceed with rollback? This action cannot be undone.", 

419 default=False, 

420 style=COOL_STYLE, 

421 ).ask() 

422 ) 

423 except NON_INTERACTIVE_ERRORS: 

424 logger.debug("Running in non-interactive environment, proceeding") 

425 return True 

426 

427 

428def _validate_rollback_preconditions(tag: str) -> tuple[bool, bool]: 

429 """Validate that the tag exists somewhere before attempting rollback. 

430 

431 Args: 

432 tag: The tag to validate. 

433 

434 Returns: 

435 Tuple of (exists_locally, exists_remotely). 

436 

437 Raises: 

438 typer.Exit: If the tag doesn't exist anywhere. 

439 """ 

440 exists_locally, exists_remotely = check_tag_exists(tag) 

441 

442 if not exists_locally and not exists_remotely: 

443 console.error(f"Tag '{tag}' does not exist locally or on remote.") 

444 console.error("Nothing to rollback.") 

445 raise typer.Exit(code=1) 

446 

447 return exists_locally, exists_remotely 

448 

449 

450def _resolve_tag(options: RollbackOptions) -> str: 

451 """Determine which tag to rollback from options or interactively. 

452 

453 Args: 

454 options: Rollback configuration options. 

455 

456 Returns: 

457 The resolved tag name with 'v' prefix. 

458 

459 Raises: 

460 typer.Exit: If no tags are found in non-interactive mode. 

461 """ 

462 tag = options.tag 

463 if not tag: 

464 if options.non_interactive: 

465 recent_tags = _get_recent_tags(limit=1) 

466 if not recent_tags: 

467 console.error("No version tags found in the repository.") 

468 raise typer.Exit(code=1) 

469 tag = recent_tags[0] 

470 console.info(f"Non-interactive mode: rolling back most recent tag: {tag}") 

471 else: 

472 recent_tags = _get_recent_tags() 

473 tag = _select_tag_interactively(recent_tags) 

474 

475 if not tag.startswith("v"): 

476 tag = f"v{tag}" 

477 

478 return tag 

479 

480 

481def _should_revert_bump(options: RollbackOptions, exists_locally: bool, is_bump: bool) -> bool: 

482 """Determine whether the bump commit should be reverted. 

483 

484 If ``--revert-bump`` was passed, returns True immediately. Otherwise, 

485 prompts the user interactively when the tagged commit looks like a bump. 

486 

487 Args: 

488 options: Rollback configuration options. 

489 exists_locally: Whether the tag exists locally. 

490 is_bump: Whether the tagged commit is a bump commit. 

491 

492 Returns: 

493 True if the bump commit should be reverted. 

494 """ 

495 if options.revert_bump: 

496 return True 

497 

498 if options.non_interactive or options.dry_run or not exists_locally or not is_bump: 

499 return False 

500 

501 try: 

502 return bool( 

503 qs.confirm( 

504 "The tagged commit appears to be a version bump. Revert it too?", 

505 default=True, 

506 style=COOL_STYLE, 

507 ).ask() 

508 ) 

509 except NON_INTERACTIVE_ERRORS: 

510 logger.debug("Running in non-interactive environment") 

511 return False 

512 

513 

514def _execute_rollback( 

515 tag: str, 

516 exists_locally: bool, 

517 exists_remotely: bool, 

518 revert_bump: bool, 

519 is_bump: bool, 

520 dry_run: bool, 

521 non_interactive: bool, 

522) -> bool: 

523 """Execute the rollback steps: delete tags, revert bump, push. 

524 

525 Args: 

526 tag: The tag to rollback. 

527 exists_locally: Whether the tag exists locally. 

528 exists_remotely: Whether the tag exists on remote. 

529 revert_bump: Whether to revert the bump commit. 

530 is_bump: Whether the tagged commit is a bump commit. 

531 dry_run: If True, only simulate changes. 

532 non_interactive: If True, skip confirmation prompts. 

533 

534 Returns: 

535 True if all steps succeeded. 

536 

537 Raises: 

538 typer.Exit: If the remote tag deletion fails (non-dry-run). 

539 """ 

540 success = True 

541 

542 # Get commit hash BEFORE deleting tags (needed for revert) 

543 tag_commit = None 

544 if revert_bump and is_bump and exists_locally: 

545 tag_commit = _get_tag_commit(tag) 

546 if not tag_commit: 

547 console.error(f"Could not find commit for tag: {tag}") 

548 console.error("Skipping bump revert but proceeding with tag deletion.") 

549 revert_bump = False 

550 

551 # Delete remote tag first to stop any in-progress release 

552 if exists_remotely and not _delete_remote_tag(tag, dry_run): 

553 success = False 

554 if not dry_run: 

555 console.error("Failed to delete remote tag. Aborting remaining steps.") 

556 console.error("You can retry or manually delete with:") 

557 console.error(f" git push origin :refs/tags/{tag}") 

558 raise typer.Exit(code=1) 

559 

560 # Delete local tag 

561 if exists_locally and not _delete_local_tag(tag, dry_run): 

562 success = False 

563 if not dry_run: 

564 console.warning(f"Failed to delete local tag. Delete manually: git tag -d {tag}") 

565 

566 # Revert bump commit and push (if requested) 

567 if revert_bump and is_bump and tag_commit: 

568 if not _revert_bump_commit(tag_commit, dry_run): 

569 success = False 

570 if not dry_run: 

571 console.warning("Bump revert failed. Tags were still deleted.") 

572 console.warning("You may need to manually revert the bump commit.") 

573 else: 

574 if not _push_revert(dry_run, non_interactive): 

575 success = False 

576 

577 return success 

578 

579 

580def _print_rollback_summary(dry_run: bool, success: bool, previous_tag: str | None) -> None: 

581 """Print a summary after rollback execution. 

582 

583 Args: 

584 dry_run: Whether this was a dry run. 

585 success: Whether all rollback steps succeeded. 

586 previous_tag: The previous version tag, if any. 

587 """ 

588 if dry_run: 

589 console.info("\n[DRY-RUN] Rollback preview complete (no changes made)") 

590 return 

591 

592 if not success: 

593 console.warning("\nRollback completed with warnings. Review the output above.") 

594 return 

595 

596 success_msg = typer.style("✓", fg=typer.colors.GREEN, bold=True) 

597 console.success(f"\n{success_msg} Rollback completed successfully!") 

598 

599 if previous_tag: 

600 console.info(f"Previous version was: {previous_tag}") 

601 console.info("To re-release at the previous version, run:") 

602 console.info(" rhiza-tools release") 

603 console.info("To bump to a new version instead:") 

604 console.info(" rhiza-tools bump") 

605 else: 

606 console.info("No previous version tag found.") 

607 console.info("To set a new version:") 

608 console.info(" rhiza-tools bump <version>") 

609 

610 

611def rollback_command(options: RollbackOptions) -> None: 

612 """Rollback a release and/or version bump. 

613 

614 This command safely reverses release and bump operations by: 

615 

616 1. Deleting the release tag from remote (stops/prevents the release workflow) 

617 2. Deleting the release tag locally 

618 3. Optionally reverting the version bump commit (with ``--revert-bump``) 

619 4. Optionally pushing the revert commit to remote 

620 

621 The command uses ``git revert`` rather than ``git reset`` to create a new 

622 revert commit, making it safe even when changes have been pushed to remote. 

623 

624 Args: 

625 options: Configuration options for the rollback. 

626 

627 Raises: 

628 typer.Exit: If the tag doesn't exist, pyproject.toml is missing, 

629 or any git operations fail. 

630 

631 Example: 

632 Rollback the most recent release:: 

633 

634 rollback_command(RollbackOptions()) 

635 

636 Preview rollback:: 

637 

638 rollback_command(RollbackOptions(dry_run=True)) 

639 

640 Rollback a specific tag with bump revert:: 

641 

642 rollback_command(RollbackOptions(tag="v1.2.3", revert_bump=True)) 

643 

644 Non-interactive rollback:: 

645 

646 rollback_command(RollbackOptions( 

647 tag="v1.2.3", 

648 revert_bump=True, 

649 non_interactive=True, 

650 )) 

651 """ 

652 validate_pyproject_exists() 

653 

654 result = run_git_command(["git", "rev-parse", "--abbrev-ref", "HEAD"]) 

655 current_branch = result.stdout.strip() 

656 console.info(f"Current branch: {typer.style(current_branch, fg=typer.colors.CYAN, bold=True)}") 

657 

658 tag = _resolve_tag(options) 

659 exists_locally, exists_remotely = _validate_rollback_preconditions(tag) 

660 

661 tag_details = _get_tag_details(tag) if exists_locally else {} 

662 is_bump = _is_bump_commit(tag) if exists_locally else False 

663 previous_tag = _get_previous_version_from_tags(tag) 

664 revert_bump = _should_revert_bump(options, exists_locally, is_bump) 

665 

666 _show_rollback_plan(tag, exists_locally, exists_remotely, revert_bump, is_bump, previous_tag, tag_details) 

667 

668 if not options.dry_run and not _confirm_rollback(options.non_interactive): 

669 console.info("Rollback cancelled by user.") 

670 raise typer.Exit(code=0) 

671 

672 success = _execute_rollback( 

673 tag, exists_locally, exists_remotely, revert_bump, is_bump, options.dry_run, options.non_interactive 

674 ) 

675 _print_rollback_summary(options.dry_run, success, previous_tag)