Coverage for frappe_manager / site_manager / exceptions.py: 30%

1from builtins import len 

2from pathlib import Path 

3 

4from rich.box import Box 

5 

6from frappe_manager.docker.subprocess_output import SubprocessOutput 

7from frappe_manager.utils import helpers 

8 

9 

10class BenchException(Exception): 

11 """Base exception for all bench-related errors.""" 

12 

13 def __init__( 

14 self, 

15 bench_name: str, 

16 message: str, 

17 prefix_bench_name: bool = True, 

18 ): 

19 self.message = message 

20 

21 if prefix_bench_name: 

22 self.message = f"[blue][bold]{bench_name} :[/bold][/blue] {message}" 

23 

24 super().__init__(self.message) 

25 

26 

27class BenchDockerComposeFileNotFound(BenchException): 

28 """Raised when docker-compose.yml file is not found.""" 

29 

30 def __init__( 

31 self, 

32 bench_name: str, 

33 path: Path, 

34 message: str = "Compose file not found at {}. Aborting operation.", 

35 ): 

36 self.bench_name = bench_name 

37 self.path = path 

38 self.message = message.format(self.path) 

39 super().__init__(self.bench_name, self.message) 

40 

41 

42class BenchServiceNotRunning(BenchException): 

43 """Raised when a required bench service is not running.""" 

44 

45 def __init__( 

46 self, 

47 bench_name: str, 

48 service: str, 

49 message: str = "Service {} not running.", 

50 ): 

51 self.bench_name = bench_name 

52 self.service = service 

53 self.message = message.format(self.service) 

54 super().__init__(self.bench_name, self.message) 

55 

56 

57class BenchNotFoundError(FileNotFoundError, BenchException): 

58 """Raised when bench directory is not found at expected location.""" 

59 

60 def __init__( 

61 self, 

62 bench_name: str, 

63 path: Path, 

64 message: str = "Bench not found at {}.", 

65 ): 

66 self.bench_name = bench_name 

67 self.path = path 

68 self.message = message.format(self.path) 

69 super().__init__(self.bench_name, self.message) 

70 

71 

72class BenchRemoveDirectoryError(BenchException): 

73 """Raised when bench directory removal fails.""" 

74 

75 def __init__( 

76 self, 

77 bench_name: str, 

78 path: Path, 

79 message: str = "Remove dirs failed at {}.", 

80 ): 

81 self.bench_name = bench_name 

82 self.path = path 

83 self.message = message.format(self.path) 

84 super().__init__(self.bench_name, self.message) 

85 

86 

87class BenchLogFileNotFoundError(BenchException): 

88 """Raised when bench log file is not found.""" 

89 

90 def __init__( 

91 self, 

92 bench_name: str, 

93 path: Path, 

94 message: str = "Log file not found at {}.", 

95 ): 

96 self.bench_name = bench_name 

97 self.path = path 

98 self.message = message.format(self.path) 

99 super().__init__(self.bench_name, self.message) 

100 

101 

102class BenchWorkersStartError(BenchException): 

103 """Raised when bench workers fail to start.""" 

104 

105 def __init__( 

106 self, 

107 bench_name: str, 

108 message: str = "Workers not able to start.", 

109 ): 

110 self.bench_name = bench_name 

111 self.message = message 

112 super().__init__(self.bench_name, self.message) 

113 

114 

115class BenchWorkersSupervisorConfigurtionGenerateError(BenchException): 

116 """Raised when supervisor worker configuration generation fails.""" 

117 

118 def __init__( 

119 self, 

120 bench_name: str, 

121 message: str = "Failed to configure workers.", 

122 ): 

123 self.bench_name = bench_name 

124 self.message = message 

125 super().__init__(self.bench_name, self.message) 

126 

127 

128class BenchWorkersSupervisorConfigurtionNotFoundError(BenchException): 

129 """Raised when supervisor worker configuration file is not found.""" 

130 

131 def __init__( 

132 self, 

133 bench_name: str, 

134 config_dir: str, 

135 message: str = "Superviosrd workers configuration not found in {}.", 

136 ): 

137 self.bench_name = bench_name 

138 self.config_dir = config_dir 

139 self.message = message.format(self.config_dir) 

140 super().__init__(self.bench_name, self.message) 

141 

142 

143class BenchConfigFileNotFound(BenchException): 

144 """Raised when bench configuration file (bench_config.toml) is not found.""" 

145 

146 def __init__(self, bench_name, config_path, message="Config file not found at {}."): 

147 self.bench_name = bench_name 

148 self.config_path = config_path 

149 self.message = message.format(config_path) 

150 super().__init__(self.bench_name, self.message) 

151 

152 

153class BenchConfigValidationError(BenchException): 

154 """Raised when bench configuration validation fails.""" 

155 

156 def __init__(self, bench_name, config_path, message="FM bench config not valid at {}"): 

157 self.bench_name = bench_name 

158 self.config_path = config_path 

159 self.message = message.format(self.config_path) 

160 super().__init__(self.bench_name, self.message) 

161 

162 

163class AdminToolsFailedToStart(BenchException): 

164 """Raised when admin tools (mailpit, adminer, redis-queue-dashboard) fail to start.""" 

165 

166 def __init__(self, bench_name, message="Failed to start admin tools."): 

167 self.bench_name = bench_name 

168 self.message = message 

169 super().__init__(self.bench_name, self.message) 

170 

171 

172class BenchSSLCertificateAlreadyIssued(BenchException): 

173 """Raised when attempting to issue an SSL certificate that already exists.""" 

174 

175 def __init__(self, bench_name, message="SSL Certificate already issued."): 

176 self.bench_name = bench_name 

177 self.message = message 

178 super().__init__(self.bench_name, self.message) 

179 

180 

181class BenchSSLCertificateNotIssued(BenchException): 

182 """Raised when SSL certificate operation requires an issued certificate but none exists.""" 

183 

184 def __init__(self, bench_name, message="No SSL Certificate issued."): 

185 self.bench_name = bench_name 

186 self.message = message 

187 super().__init__(self.bench_name, self.message) 

188 

189 

190class BenchAttachTocontainerFailed(BenchException): 

191 """Raised when attaching to a container fails.""" 

192 

193 def __init__(self, bench_name, service_name, message="Attach to {} service container failed."): 

194 self.bench_name = bench_name 

195 self.service_name = service_name 

196 self.message = message.format(self.service_name) 

197 super().__init__(self.bench_name, self.message) 

198 

199 

200class BenchNotRunning(BenchException): 

201 """Raised when bench services are required to be running but are not.""" 

202 

203 def __init__(self, bench_name, message="Bench services not running."): 

204 self.bench_name = bench_name 

205 self.message = message 

206 super().__init__(self.bench_name, self.message) 

207 

208 

209class BenchFailedToRemoveDevPackages(BenchException): 

210 """Raised when pip uninstall of development packages fails.""" 

211 

212 def __init__(self, bench_name, message="Not able pip uninstall dev packages."): 

213 self.bench_name = bench_name 

214 self.message = message 

215 super().__init__(self.bench_name, self.message) 

216 

217 

218class BenchFrappeServiceSupervisorNotRunning(BenchException): 

219 """Raised when supervisorctl is not running in the frappe service container.""" 

220 

221 def __init__(self, bench_name, message="Supervisorctl is not running in frappe service"): 

222 self.bench_name = bench_name 

223 self.message = message 

224 super().__init__(self.bench_name, self.message) 

225 

226 

227class BenchOperationException(BenchException): 

228 """Base exception for bench operations that may include subprocess output.""" 

229 

230 def __init__( 

231 self, 

232 bench_name, 

233 message: str, 

234 print_combined: bool = True, 

235 print_stdout: bool = False, 

236 print_stderr: bool = False, 

237 ): 

238 self.bench_name = bench_name 

239 self.message = message 

240 self.print_stdout = print_stdout 

241 self.print_stderr = print_stderr 

242 self.print_combined = print_combined 

243 self.output = None 

244 super().__init__(self.bench_name, self.message) 

245 

246 def set_output(self, output: SubprocessOutput): 

247 self.output = output 

248 from rich.panel import Panel 

249 

250 to_print = [] 

251 

252 box: Box = Box("╭ \n \n ── \n│ \n \n \n | \n \n", ascii=True) 

253 

254 if self.print_stdout: 

255 panel = Panel.fit( 

256 "\n".join(self.output.stdout), 

257 box=box, 

258 padding=(0, 1), 

259 border_style="dim", 

260 title="Error command stdout", 

261 title_align="left", 

262 ) 

263 to_print.append(helpers.rich_object_to_string(panel)) 

264 

265 if self.print_combined: 

266 panel = Panel.fit( 

267 "\n".join(self.output.combined), 

268 box=box, 

269 padding=(0, 1), 

270 border_style="dim", 

271 title="Error command output", 

272 title_align="left", 

273 ) 

274 to_print.append(helpers.rich_object_to_string(panel)) 

275 

276 if self.print_stderr: 

277 panel = Panel.fit( 

278 "\n".join(self.output.stderr), 

279 box=box, 

280 padding=(0, 1), 

281 border_style="dim", 

282 title="Error command stderr", 

283 title_align="left", 

284 ) 

285 to_print.append(helpers.rich_object_to_string(panel)) 

286 

287 self.message = self.message + "\n" + "\n".join(to_print) 

288 

289 super().__init__(self.bench_name, self.message, prefix_bench_name=False) 

290 

291 

292class BenchOperationFrappeBranchChangeFailed(BenchOperationException): 

293 """Raised when changing a Frappe app branch fails.""" 

294 

295 def __init__(self, bench_name, app: str, branch: str, message: str = "Failed to change {} app branch to {}."): 

296 self.app = app 

297 self.branch = branch 

298 formatted_message = message.format(app, branch) 

299 super().__init__(bench_name, formatted_message) 

300 

301 

302class BenchOperationRequiredDockerImagesNotAvailable(BenchException): 

303 """Raised when required Docker images are not available locally.""" 

304 

305 def __init__( 

306 self, 

307 bench_name, 

308 pull_command, 

309 message: str = "Required docker images not available. Pull all required images using command '{}'.", 

310 ): 

311 self.bench_name = bench_name 

312 self.message = message.format(pull_command) 

313 super().__init__(self.bench_name, self.message) 

314 

315 

316class BenchOperationWaitForRequiredServiceFailed(BenchOperationException): 

317 """Raised when waiting for a required service to become available times out.""" 

318 

319 def __init__( 

320 self, 

321 bench_name, 

322 host: str, 

323 port: str, 

324 timeout: int, 

325 message: str = "Waiting for service {}:{} timed out. {}", 

326 print_combined: bool = True, 

327 print_stdout: bool = False, 

328 print_stderr: bool = False, 

329 ): 

330 self.bench_name = bench_name 

331 self.host = host 

332 self.port = port 

333 self.timeout = timeout 

334 self.print_stdout = print_stdout 

335 self.print_stderr = print_stderr 

336 self.print_combined = print_combined 

337 self.message = message.format(host, port, timeout) 

338 

339 super().__init__(self.bench_name, self.message, self.print_combined, self.print_stdout, self.print_stderr) 

340 

341 

342class BenchOperationBenchSiteCreateFailed(BenchOperationException): 

343 """Raised when bench site creation fails.""" 

344 

345 def __init__( 

346 self, 

347 bench_name, 

348 print_combined: bool = True, 

349 print_stdout: bool = False, 

350 print_stderr: bool = False, 

351 message: str = "Failed to create site {}.", 

352 ): 

353 self.bench_name = bench_name 

354 self.message = message.format(bench_name) 

355 self.print_stdout = print_stdout 

356 self.print_stderr = print_stderr 

357 self.print_combined = print_combined 

358 super().__init__(self.bench_name, self.message, self.print_combined, self.print_stdout, self.print_stderr) 

359 

360 

361class BenchOperationBenchInstallAppInPythonEnvFailed(BenchOperationException): 

362 """Raised when installing an app in the Python environment fails.""" 

363 

364 def __init__( 

365 self, 

366 bench_name, 

367 app_name: str, 

368 message: str = "Failed to install app {} in python env.", 

369 print_combined: bool = True, 

370 print_stdout: bool = False, 

371 print_stderr: bool = False, 

372 ): 

373 self.bench_name = bench_name 

374 self.app_name = app_name 

375 self.message = message.format(app_name) 

376 self.print_stdout = print_stdout 

377 self.print_stderr = print_stderr 

378 self.print_combined = print_combined 

379 

380 super().__init__(self.bench_name, self.message, self.print_combined, self.print_stdout, self.print_stderr) 

381 

382 

383class BenchOperationBenchRemoveAppFromPythonEnvFailed(BenchOperationException): 

384 """Raised when removing an app from the Python environment fails.""" 

385 

386 def __init__( 

387 self, 

388 bench_name, 

389 app_name: str, 

390 message: str = "Failed to remove app {} from python env.", 

391 print_combined: bool = True, 

392 print_stdout: bool = False, 

393 print_stderr: bool = False, 

394 ): 

395 self.bench_name = bench_name 

396 self.app_name = app_name 

397 self.message = message.format(app_name) 

398 self.print_stdout = print_stdout 

399 self.print_stderr = print_stderr 

400 self.print_combined = print_combined 

401 

402 super().__init__(self.bench_name, self.message, self.print_combined, self.print_stdout, self.print_stderr) 

403 

404 

405class BenchOperationBenchAppInSiteFailed(BenchOperationException): 

406 """Raised when installing an app in a site fails.""" 

407 

408 def __init__( 

409 self, 

410 bench_name, 

411 app_name: str, 

412 message: str = "Failed to install app {} in site {}.", 

413 print_combined: bool = True, 

414 print_stdout: bool = False, 

415 print_stderr: bool = False, 

416 ): 

417 self.bench_name = bench_name 

418 self.app_name = app_name 

419 self.message = message.format(app_name, self.bench_name) 

420 self.print_stdout = print_stdout 

421 self.print_stderr = print_stderr 

422 self.print_combined = print_combined 

423 super().__init__(self.bench_name, self.message, self.print_combined, self.print_stdout, self.print_stderr) 

424 

425 

426class BenchOperationBenchBuildFailed(BenchOperationException): 

427 """Raised when bench build operation fails.""" 

428 

429 def __init__( 

430 self, 

431 bench_name, 

432 apps: list[str] | None = None, 

433 message: str = "Failed to build", 

434 print_combined: bool = True, 

435 print_stdout: bool = False, 

436 print_stderr: bool = False, 

437 ): 

438 self.bench_name = bench_name 

439 self.apps = apps 

440 if apps: 

441 message = message + " app" 

442 if len(apps) > 1: 

443 message = message + " apps" 

444 for app in apps: 

445 message += f" {app}" 

446 self.message = message 

447 self.print_stdout = print_stdout 

448 self.print_stderr = print_stderr 

449 self.print_combined = print_combined 

450 super().__init__(self.bench_name, self.message, self.print_combined, self.print_stdout, self.print_stderr)