Coverage for frappe_manager / migration_manager / migration_error_handler.py: 17%

1""" 

2Migration error handling, reporting, and recovery. 

3 

4Handles failed migration scenarios including bench-specific failures, 

5archive operations, rollback coordination, and user-facing error reports. 

6""" 

7 

8import shutil 

9from typing import TYPE_CHECKING 

10 

11from frappe_manager import CLI_DIR, CLI_SITES_ARCHIVE 

12from frappe_manager.migration_manager.migration_exections import MigrationExceptionInBench 

13 

14if TYPE_CHECKING: 

15 from frappe_manager.migration_manager.migration_executor import MigrationExecutor 

16 

17 

18class MigrationErrorHandler: 

19 """ 

20 Handles migration failures, error reporting, and recovery operations. 

21 

22 Responsibilities: 

23 - Report failed/passed benches 

24 - Prompt user for archive vs rollback decision 

25 - Execute archiving of failed benches 

26 - Coordinate rollback operations 

27 - Update FM version after recovery 

28 """ 

29 

30 def __init__(self, executor: "MigrationExecutor"): 

31 self.executor = executor 

32 

33 def handle_bench_migration_failure(self, exception: MigrationExceptionInBench) -> bool: 

34 """ 

35 Handle MigrationExceptionInBench - partial bench migration failures. 

36 

37 Reports which benches passed/failed, prompts user for recovery action, 

38 and executes either archive or rollback. 

39 

40 Args: 

41 exception: The bench migration exception that was raised 

42 

43 Returns: 

44 bool: True if recovery succeeded, False if rollback was required 

45 """ 

46 if not self.executor.migrate_benches: 

47 return False 

48 

49 self._report_bench_results() 

50 

51 is_single_bench = self.executor.target_benches and len(self.executor.target_benches) == 1 

52 archive_decision = self._prompt_archive_or_rollback() 

53 

54 if archive_decision == "yes": 

55 self._archive_failed_benches() 

56 return True 

57 if archive_decision == "skip": 

58 return False 

59 self._rollback_all(show_cli_downgrade_instructions=not is_single_bench) 

60 return False 

61 

62 def handle_system_migration_failure(self, exception: Exception) -> bool: 

63 """ 

64 Handle system-level migration failure. 

65 

66 Reports error and triggers rollback. 

67 

68 Args: 

69 exception: The exception that caused system migration failure 

70 

71 Returns: 

72 bool: Always False (migration failed) 

73 """ 

74 self.executor.output.display_error(f"[red]Migration failed[red] : {exception}", emoji_code="") 

75 self._rollback_all() 

76 return False 

77 

78 def _report_bench_results(self): 

79 """Report which benches passed and which failed migration.""" 

80 passed_print_head = True 

81 

82 # Report passed benches 

83 for bench, bench_status in self.executor.migrate_benches.items(): 

84 if not bench_status["exception"]: 

85 if passed_print_head: 

86 self.executor.output.print("\n\n[bold green]Migration Passed Benches[/bold green]\n", emoji_code="") 

87 passed_print_head = False 

88 self.executor.output.print(f"[green]Bench[/green]: {bench}", emoji_code="") 

89 

90 failed_print_head = True 

91 

92 # Report failed benches 

93 for bench, bench_status in self.executor.migrate_benches.items(): 

94 if bench_status["exception"]: 

95 if failed_print_head: 

96 self.executor.output.print("\n[bold red]Migration Failed Benches[/bold red]\n", emoji_code="") 

97 failed_print_head = False 

98 

99 self.executor.output.display_error(f"[red]Bench[/red]: {bench}", emoji_code="") 

100 self.executor.output.display_error( 

101 f"[red]Failed Migration Version[/red]: {bench_status['last_migration_version']}", 

102 emoji_code="", 

103 ) 

104 self.executor.output.display_error( 

105 f"[red]Exception[/red]: {type(bench_status['exception']).__name__}", 

106 emoji_code="", 

107 ) 

108 self.executor.output.print(f" {bench_status['exception']}", emoji_code="") 

109 

110 self.executor.output.print(f"For error specifics, refer to {CLI_DIR}/logs/fm.log", emoji_code="") 

111 

112 # Print separator 

113 if not failed_print_head or not passed_print_head: 

114 self.executor.output.print("=" * 60, emoji_code="") 

115 

116 def _prompt_archive_or_rollback(self) -> str: 

117 """ 

118 Prompt user to choose between archiving failed benches or rolling back all. 

119 

120 Returns: 

121 str: "yes" to archive, "no" to rollback 

122 """ 

123 target_benches = self.executor.target_benches 

124 is_single_bench = target_benches is not None and len(target_benches) == 1 

125 

126 if is_single_bench and target_benches is not None: 

127 bench_name = target_benches[0] 

128 

129 if self.executor.on_failure == "archive": 

130 self.executor.output.print( 

131 "Warning: --on-failure=archive not supported for single bench migrations. Using rollback.", 

132 emoji_code="", 

133 ) 

134 

135 if self.executor.on_failure == "rollback": 

136 self.executor.output.print("Rolling back bench (--on-failure=rollback)", emoji_code="") 

137 return "no" 

138 

139 rollback_msg = [ 

140 "Migration failed for bench.", 

141 "", 

142 "Available options:", 

143 "[blue][yes][/blue] Rollback bench : Restore the bench to its last working version before migration.", 

144 f"[blue][no][/blue] Skip rollback : Leave bench in current state. You can manually fix or retry with: fm migrate {bench_name}", 

145 "", 

146 "Do you want to rollback the bench?", 

147 ] 

148 

149 rollback_decision = self.executor.output.prompt_ask( 

150 prompt="\n".join(rollback_msg), 

151 choices=["yes", "no"], 

152 required_flag="--on-failure", 

153 ) 

154 

155 if rollback_decision == "yes": 

156 return "no" 

157 self.executor.output.print("\nSkipping rollback. Bench remains in current state.", emoji_code="") 

158 self.executor.output.print( 

159 f"You can manually fix the bench or retry with: fm migrate {bench_name}", 

160 emoji_code="", 

161 ) 

162 return "skip" 

163 

164 if self.executor.on_failure == "archive": 

165 self.executor.output.print("Archiving failed benches (--on-failure=archive)", emoji_code="") 

166 return "yes" 

167 if self.executor.on_failure == "rollback": 

168 self.executor.output.print("Rolling back all benches (--on-failure=rollback)", emoji_code="") 

169 return "no" 

170 

171 archive_msg = [ 

172 "Available options after migrations failure :", 

173 rf"[blue][yes][/blue] Archive failed benches : Benches that have failed will be rolled back to there last successfully completed migration version and stored in '{CLI_SITES_ARCHIVE}'.", 

174 r"[blue][no][/blue] Revert migration : Restore the FM CLI and FM environment to the last successfully completed migration version for all benches.", 

175 "\nDo you wish to archive all benches that failed during migration ?", 

176 ] 

177 

178 return self.executor.output.prompt_ask( 

179 prompt="\n".join(archive_msg), 

180 choices=["yes", "no"], 

181 required_flag="--on-failure", 

182 ) 

183 

184 def _archive_failed_benches(self): 

185 """ 

186 Archive all failed benches to CLI_SITES_ARCHIVE. 

187 

188 Updates executor's prev_version and moves failed bench directories. 

189 """ 

190 self.executor.prev_version = self.executor.undo_stack[-1].version 

191 

192 for bench, bench_info in self.executor.migrate_benches.items(): 

193 if bench_info["exception"]: 

194 archive_bench_path = CLI_SITES_ARCHIVE / bench 

195 CLI_SITES_ARCHIVE.mkdir(exist_ok=True, parents=True) 

196 shutil.move(bench_info["object"].path, archive_bench_path) 

197 self.executor.output.print(f"[bold]Archived bench :[/bold] [yellow]{bench}[/yellow]", emoji_code="") 

198 

199 def _rollback_all(self, show_cli_downgrade_instructions: bool = True): 

200 """ 

201 Execute full rollback and update FM config version. 

202 

203 Calls orchestrator's rollback, updates FM version, and optionally provides 

204 user instructions for CLI version rollback. 

205 

206 Args: 

207 show_cli_downgrade_instructions: Whether to show pip install command 

208 """ 

209 from frappe_manager.migration_manager.migration_orchestrator import MigrationOrchestrator 

210 

211 orchestrator = MigrationOrchestrator(self.executor) 

212 orchestrator.undo_stack = self.executor.undo_stack 

213 orchestrator.rollback_migrations() 

214 

215 self.executor.fm_config_manager.version = self.executor.rollback_version 

216 self.executor.fm_config_manager.export_to_toml() 

217 

218 self.executor.output.print("", emoji_code="") 

219 self.executor.output.print("Rollback complete.", emoji_code="") 

220 

221 if show_cli_downgrade_instructions: 

222 self.executor.output.print("", emoji_code="") 

223 self.executor.output.print( 

224 f"To revert FM CLI to v{self.executor.rollback_version.version!s}, run:", 

225 emoji_code="", 

226 ) 

227 self.executor.output.print( 

228 f" uv tool install frappe-manager=={self.executor.rollback_version.version!s}", 

229 emoji_code="", 

230 ) 

231 

232 self.executor.output.print("", emoji_code="") 

233 

234 def finalize_success(self): 

235 """ 

236 Update FM config version after successful migration. 

237 

238 Called when all migrations complete successfully. 

239 """ 

240 self.executor.fm_config_manager.version = self.executor.current_version 

241 self.executor.fm_config_manager.export_to_toml()