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

91 statements  

« prev     ^ index     » next       coverage.py v7.14.1, created at 2026-06-30 13:37 +0000

1"""Interactive UI and display helpers for the rollback command. 

2 

3These helpers handle user-facing interaction and display: selecting a tag 

4interactively, confirming the rollback, pushing the revert commit, and 

5printing the rollback plan. They contain no tag-deletion or commit-revert 

6git operations — those live in ``rollback/git.py``. 

7 

8All symbols defined here are re-exported by ``rollback.py`` so the public 

9import surface is unchanged. 

10""" 

11 

12from __future__ import annotations 

13 

14import questionary as qs 

15import typer 

16from loguru import logger 

17 

18from rhiza_tools import console 

19from rhiza_tools.commands._shared import ( 

20 COOL_STYLE, 

21 NON_INTERACTIVE_ERRORS, 

22 run_git_command, 

23) 

24from rhiza_tools.commands.release import check_tag_exists 

25 

26 

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

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

29 

30 Args: 

31 tags: List of available tags. 

32 

33 Returns: 

34 The selected tag name. 

35 

36 Raises: 

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

38 """ 

39 if not tags: 

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

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

42 raise typer.Exit(code=1) 

43 

44 # Annotate tags with local/remote info 

45 choices: list[str] = [] 

46 for tag in tags: 

47 exists_locally, exists_remotely = check_tag_exists(tag) 

48 markers = [] 

49 if exists_locally: 

50 markers.append("local") 

51 if exists_remotely: 

52 markers.append("remote") 

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

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

55 

56 try: 

57 choice = qs.select( 

58 "Select tag to rollback:", 

59 choices=choices, 

60 style=COOL_STYLE, 

61 ).ask() 

62 except NON_INTERACTIVE_ERRORS: 

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

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

65 

66 if not choice: 

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

68 raise typer.Exit(code=0) 

69 

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

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

72 

73 

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

75 """Push the revert commit to remote. 

76 

77 Args: 

78 dry_run: If True, only simulate the push. 

79 non_interactive: If True, skip confirmation prompt. 

80 

81 Returns: 

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

83 """ 

84 if dry_run: 

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

86 return True 

87 

88 should_push = non_interactive 

89 if not non_interactive: 

90 try: 

91 should_push = qs.confirm( 

92 "Push revert commit to remote?", 

93 default=True, 

94 style=COOL_STYLE, 

95 ).ask() 

96 except NON_INTERACTIVE_ERRORS: 

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

98 should_push = True 

99 

100 if not should_push: 

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

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

103 return True 

104 

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

106 if result.returncode == 0: 

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

108 return True 

109 else: 

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

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

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

113 return False 

114 

115 

116def _show_rollback_plan( 

117 tag: str, 

118 exists_locally: bool, 

119 exists_remotely: bool, 

120 revert_bump: bool, 

121 is_bump: bool, 

122 previous_tag: str | None, 

123 tag_details: dict[str, str], 

124) -> None: 

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

126 

127 Args: 

128 tag: Tag being rolled back. 

129 exists_locally: Whether the tag exists locally. 

130 exists_remotely: Whether the tag exists on remote. 

131 revert_bump: Whether to revert the bump commit. 

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

133 previous_tag: The previous version tag, if any. 

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

135 """ 

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

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

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

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

140 

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

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

143 

144 if tag_details: 

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

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

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

148 

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

150 

151 step = 1 

152 if exists_remotely: 

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

154 step += 1 

155 if exists_locally: 

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

157 step += 1 

158 if revert_bump and is_bump: 

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

160 step += 1 

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

162 step += 1 

163 

164 if previous_tag: 

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

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

167 else: 

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

169 

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

171 

172 

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

174 """Confirm rollback with the user. 

175 

176 Args: 

177 non_interactive: If True, skip confirmation. 

178 

179 Returns: 

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

181 """ 

182 if non_interactive: 

183 return True 

184 

185 try: 

186 return bool( 

187 qs.confirm( 

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

189 default=False, 

190 style=COOL_STYLE, 

191 ).ask() 

192 ) 

193 except NON_INTERACTIVE_ERRORS: 

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

195 return True