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

273 statements  

« prev     ^ index     » next       coverage.py v7.13.4, created at 2026-02-23 01:10 +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 run_git_command, 

34 validate_pyproject_exists, 

35) 

36from rhiza_tools.commands.release import check_tag_exists 

37 

38 

39@dataclass 

40class RollbackOptions: 

41 """Configuration options for the rollback command. 

42 

43 Attributes: 

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

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

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

47 non_interactive: If True, skip all confirmation prompts. 

48 """ 

49 

50 tag: str | None = None 

51 revert_bump: bool = False 

52 dry_run: bool = False 

53 non_interactive: bool = False 

54 

55 

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

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

58 

59 Args: 

60 limit: Maximum number of tags to return. 

61 

62 Returns: 

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

64 """ 

65 result = run_git_command( 

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

67 check=False, 

68 ) 

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

70 return [] 

71 

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

73 return tags[:limit] 

74 

75 

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

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

78 

79 Args: 

80 tags: List of available tags. 

81 

82 Returns: 

83 The selected tag name. 

84 

85 Raises: 

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

87 """ 

88 if not tags: 

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

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

91 raise typer.Exit(code=1) 

92 

93 # Annotate tags with local/remote info 

94 choices: list[str] = [] 

95 for tag in tags: 

96 exists_locally, exists_remotely = check_tag_exists(tag) 

97 markers = [] 

98 if exists_locally: 

99 markers.append("local") 

100 if exists_remotely: 

101 markers.append("remote") 

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

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

104 

105 try: 

106 choice = qs.select( 

107 "Select tag to rollback:", 

108 choices=choices, 

109 style=COOL_STYLE, 

110 ).ask() 

111 except EOFError: 

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

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

114 

115 if not choice: 

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

117 raise typer.Exit(code=0) 

118 

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

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

121 

122 

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

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

125 

126 Args: 

127 tag: The tag name. 

128 

129 Returns: 

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

131 """ 

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

133 if result.returncode != 0: 

134 return None 

135 return result.stdout.strip() 

136 

137 

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

139 """Get details about a tag. 

140 

141 Args: 

142 tag: The tag name. 

143 

144 Returns: 

145 Dictionary with commit hash, date, and message. 

146 """ 

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

148 result = run_git_command( 

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

150 check=False, 

151 ) 

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

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

154 if len(parts) == 3: 

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

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

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

158 return details 

159 

160 

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

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

163 

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

165 or contain version-related keywords. 

166 

167 Args: 

168 tag: The tag name. 

169 

170 Returns: 

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

172 """ 

173 result = run_git_command( 

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

175 check=False, 

176 ) 

177 if result.returncode != 0: 

178 return False 

179 

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

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

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

183 

184 

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

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

187 

188 Args: 

189 current_tag: The current tag being rolled back. 

190 

191 Returns: 

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

193 """ 

194 result = run_git_command( 

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

196 check=False, 

197 ) 

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

199 return None 

200 

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

202 

203 try: 

204 idx = tags.index(current_tag) 

205 if idx + 1 < len(tags): 

206 return tags[idx + 1] 

207 except ValueError: 

208 pass 

209 

210 return None 

211 

212 

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

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

215 

216 Args: 

217 tag: The tag name to delete. 

218 dry_run: If True, only simulate deletion. 

219 

220 Returns: 

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

222 """ 

223 if dry_run: 

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

225 return True 

226 

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

228 if result.returncode == 0: 

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

230 return True 

231 else: 

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

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

234 return False 

235 

236 

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

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

239 

240 Args: 

241 tag: The tag name to delete. 

242 dry_run: If True, only simulate deletion. 

243 

244 Returns: 

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

246 """ 

247 if dry_run: 

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

249 return True 

250 

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

252 result = run_git_command( 

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

254 check=False, 

255 ) 

256 if result.returncode == 0: 

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

258 return True 

259 else: 

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

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

262 return False 

263 

264 

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

266 """Revert the version bump commit. 

267 

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

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

270 

271 Args: 

272 commit_hash: The commit hash to revert. 

273 dry_run: If True, only simulate the revert. 

274 

275 Returns: 

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

277 """ 

278 if dry_run: 

279 result = run_git_command( 

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

281 check=False, 

282 ) 

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

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

285 return True 

286 

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

288 result = run_git_command( 

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

290 check=False, 

291 ) 

292 if result.returncode == 0: 

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

294 return True 

295 else: 

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

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

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

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

300 return False 

301 

302 

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

304 """Push the revert commit to remote. 

305 

306 Args: 

307 dry_run: If True, only simulate the push. 

308 non_interactive: If True, skip confirmation prompt. 

309 

310 Returns: 

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

312 """ 

313 if dry_run: 

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

315 return True 

316 

317 should_push = non_interactive 

318 if not non_interactive: 

319 try: 

320 should_push = qs.confirm( 

321 "Push revert commit to remote?", 

322 default=True, 

323 style=COOL_STYLE, 

324 ).ask() 

325 except EOFError: 

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

327 should_push = True 

328 

329 if not should_push: 

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

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

332 return True 

333 

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

335 if result.returncode == 0: 

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

337 return True 

338 else: 

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

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

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

342 return False 

343 

344 

345def _show_rollback_plan( 

346 tag: str, 

347 exists_locally: bool, 

348 exists_remotely: bool, 

349 revert_bump: bool, 

350 is_bump: bool, 

351 previous_tag: str | None, 

352 tag_details: dict[str, str], 

353) -> None: 

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

355 

356 Args: 

357 tag: Tag being rolled back. 

358 exists_locally: Whether the tag exists locally. 

359 exists_remotely: Whether the tag exists on remote. 

360 revert_bump: Whether to revert the bump commit. 

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

362 previous_tag: The previous version tag, if any. 

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

364 """ 

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

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

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

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

369 

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

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

372 

373 if tag_details: 

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

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

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

377 

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

379 

380 step = 1 

381 if exists_remotely: 

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

383 step += 1 

384 if exists_locally: 

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

386 step += 1 

387 if revert_bump and is_bump: 

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

389 step += 1 

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

391 step += 1 

392 

393 if previous_tag: 

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

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

396 else: 

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

398 

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

400 

401 

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

403 """Confirm rollback with the user. 

404 

405 Args: 

406 non_interactive: If True, skip confirmation. 

407 

408 Returns: 

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

410 """ 

411 if non_interactive: 

412 return True 

413 

414 try: 

415 return bool( 

416 qs.confirm( 

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

418 default=False, 

419 style=COOL_STYLE, 

420 ).ask() 

421 ) 

422 except EOFError: 

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

424 return True 

425 

426 

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

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

429 

430 Args: 

431 tag: The tag to validate. 

432 

433 Returns: 

434 Tuple of (exists_locally, exists_remotely). 

435 

436 Raises: 

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

438 """ 

439 exists_locally, exists_remotely = check_tag_exists(tag) 

440 

441 if not exists_locally and not exists_remotely: 

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

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

444 raise typer.Exit(code=1) 

445 

446 return exists_locally, exists_remotely 

447 

448 

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

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

451 

452 Args: 

453 options: Rollback configuration options. 

454 

455 Returns: 

456 The resolved tag name with 'v' prefix. 

457 

458 Raises: 

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

460 """ 

461 tag = options.tag 

462 if not tag: 

463 if options.non_interactive: 

464 recent_tags = _get_recent_tags(limit=1) 

465 if not recent_tags: 

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

467 raise typer.Exit(code=1) 

468 tag = recent_tags[0] 

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

470 else: 

471 recent_tags = _get_recent_tags() 

472 tag = _select_tag_interactively(recent_tags) 

473 

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

475 tag = f"v{tag}" 

476 

477 return tag 

478 

479 

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

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

482 

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

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

485 

486 Args: 

487 options: Rollback configuration options. 

488 exists_locally: Whether the tag exists locally. 

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

490 

491 Returns: 

492 True if the bump commit should be reverted. 

493 """ 

494 if options.revert_bump: 

495 return True 

496 

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

498 return False 

499 

500 try: 

501 return bool( 

502 qs.confirm( 

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

504 default=True, 

505 style=COOL_STYLE, 

506 ).ask() 

507 ) 

508 except EOFError: 

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

510 return False 

511 

512 

513def _execute_rollback( 

514 tag: str, 

515 exists_locally: bool, 

516 exists_remotely: bool, 

517 revert_bump: bool, 

518 is_bump: bool, 

519 dry_run: bool, 

520 non_interactive: bool, 

521) -> bool: 

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

523 

524 Args: 

525 tag: The tag to rollback. 

526 exists_locally: Whether the tag exists locally. 

527 exists_remotely: Whether the tag exists on remote. 

528 revert_bump: Whether to revert the bump commit. 

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

530 dry_run: If True, only simulate changes. 

531 non_interactive: If True, skip confirmation prompts. 

532 

533 Returns: 

534 True if all steps succeeded. 

535 

536 Raises: 

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

538 """ 

539 success = True 

540 

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

542 tag_commit = None 

543 if revert_bump and is_bump and exists_locally: 

544 tag_commit = _get_tag_commit(tag) 

545 if not tag_commit: 

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

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

548 revert_bump = False 

549 

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

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

552 success = False 

553 if not dry_run: 

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

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

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

557 raise typer.Exit(code=1) 

558 

559 # Delete local tag 

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

561 success = False 

562 if not dry_run: 

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

564 

565 # Revert bump commit and push (if requested) 

566 if revert_bump and is_bump and tag_commit: 

567 if not _revert_bump_commit(tag_commit, dry_run): 

568 success = False 

569 if not dry_run: 

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

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

572 else: 

573 if not _push_revert(dry_run, non_interactive): 

574 success = False 

575 

576 return success 

577 

578 

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

580 """Print a summary after rollback execution. 

581 

582 Args: 

583 dry_run: Whether this was a dry run. 

584 success: Whether all rollback steps succeeded. 

585 previous_tag: The previous version tag, if any. 

586 """ 

587 if dry_run: 

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

589 return 

590 

591 if not success: 

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

593 return 

594 

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

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

597 

598 if previous_tag: 

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

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

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

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

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

604 else: 

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

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

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

608 

609 

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

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

612 

613 This command safely reverses release and bump operations by: 

614 

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

616 2. Deleting the release tag locally 

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

618 4. Optionally pushing the revert commit to remote 

619 

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

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

622 

623 Args: 

624 options: Configuration options for the rollback. 

625 

626 Raises: 

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

628 or any git operations fail. 

629 

630 Example: 

631 Rollback the most recent release:: 

632 

633 rollback_command(RollbackOptions()) 

634 

635 Preview rollback:: 

636 

637 rollback_command(RollbackOptions(dry_run=True)) 

638 

639 Rollback a specific tag with bump revert:: 

640 

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

642 

643 Non-interactive rollback:: 

644 

645 rollback_command(RollbackOptions( 

646 tag="v1.2.3", 

647 revert_bump=True, 

648 non_interactive=True, 

649 )) 

650 """ 

651 validate_pyproject_exists() 

652 

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

654 current_branch = result.stdout.strip() 

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

656 

657 tag = _resolve_tag(options) 

658 exists_locally, exists_remotely = _validate_rollback_preconditions(tag) 

659 

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

661 is_bump = _is_bump_commit(tag) if exists_locally else False 

662 previous_tag = _get_previous_version_from_tags(tag) 

663 revert_bump = _should_revert_bump(options, exists_locally, is_bump) 

664 

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

666 

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

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

669 raise typer.Exit(code=0) 

670 

671 success = _execute_rollback( 

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

673 ) 

674 _print_rollback_summary(options.dry_run, success, previous_tag)