Coverage for frappe_manager / commands / __init__.py: 55%

1import os 

2import secrets 

3import shutil 

4import sys 

5import uuid 

6from pathlib import Path 

7from typing import Annotated, List, Optional, cast 

8 

9import typer 

10from typer_examples import install 

11from rich.panel import Panel 

12 

13from frappe_manager import ( 

14 CLI_BENCH_CONFIG_FILE_NAME, 

15 CLI_BENCHES_DIRECTORY, 

16 CLI_DIR, 

17 CLI_FM_CONFIG_PATH, 

18 DEFAULT_EXTENSIONS, 

19 STABLE_APP_BRANCH_MAPPING_LIST, 

20 EnableDisableOptionsEnum, 

21 SiteServicesEnum, 

22) 

23from frappe_manager.commands.self import self_app 

24from frappe_manager.commands.services import services_app 

25from frappe_manager.commands.ssl import ssl_app 

26from frappe_manager.docker import ComposeFile, DockerClient 

27from frappe_manager.logger import log 

28from frappe_manager.logger.context import LoggerContext 

29from frappe_manager.metadata_manager import FMConfigManager 

30from frappe_manager.migration_manager.bench_migration_state import ( 

31 bench_needs_migration, 

32 get_bench_migration_version, 

33 set_bench_migration_version, 

34) 

35from frappe_manager.migration_manager.migration_executor import ( 

36 MigrationExecutor, 

37 get_benches_needing_migration, 

38 needs_fm_infrastructure_migration, 

39 needs_migration, 

40) 

41from frappe_manager.migration_manager.version import Version 

42from frappe_manager.ngrok import create_tunnel 

43from frappe_manager.output_manager import OutputHandler, get_global_output_handler, spinner, temporary_stop 

44from frappe_manager.output_manager.logging_output import LoggingOutputHandler 

45from frappe_manager.services_manager.services import ServicesManager 

46from frappe_manager.services_manager.services_exceptions import ServicesNotCreated 

47from frappe_manager.site_manager.bench_config import AppConfig, BenchConfig, FMBenchEnvType, RestartPolicyEnum 

48from frappe_manager.site_manager.bench_service import BenchService 

49from frappe_manager.site_manager.domain_conflict import DomainConflictError, validate_domains_unique 

50from frappe_manager.site_manager.exceptions import BenchNotRunning 

51from frappe_manager.site_manager.modules.app_cloner import AppCloner 

52from frappe_manager.site_manager.site import Bench 

53from frappe_manager.utils.callbacks import ( 

54 alias_domains_validation_callback, 

55 apps_list_validation_callback, 

56 code_command_extensions_callback, 

57 create_command_sitename_callback, 

58 sitename_callback, 

59 sites_autocompletion_callback, 

60 version_callback, 

61) 

62from frappe_manager.utils.helpers import ( 

63 get_current_fm_version, 

64 is_cli_help_called, 

65) 

66from frappe_manager.utils.site import pull_docker_images, validate_sitename 

67 

68# Helper functions 

69 

70 

71def get_bench_arg_from_context(ctx: typer.Context) -> str | None: 

72 """ 

73 Extract bench/site name from command context. 

74 Commands use different parameter names (benchname, sitename, bench_name). 

75 """ 

76 return ctx.params.get("benchname") or ctx.params.get("sitename") or ctx.params.get("bench_name") 

77 

78 

79def check_bench_migration_required(bench_name: str | None) -> None: 

80 from frappe_manager.migration_manager.bench_migration_state import bench_needs_migration 

81 

82 if not bench_name: 

83 return 

84 

85 bench_path = CLI_BENCHES_DIRECTORY / bench_name 

86 

87 if not bench_path.exists(): 

88 return 

89 

90 current_version = Version(get_current_fm_version()) 

91 

92 if bench_needs_migration(bench_path, current_version): 

93 output = get_global_output_handler() 

94 output.stop() 

95 

96 bench_path = CLI_BENCHES_DIRECTORY / bench_name 

97 from frappe_manager.migration_manager.bench_migration_state import get_bench_migration_version 

98 

99 bench_version = get_bench_migration_version(bench_path) 

100 fm_version = Version(get_current_fm_version()) 

101 

102 output.warning(f"Bench migration required: {bench_name} (v{bench_version} → v{fm_version})\n", emoji_code="") 

103 output.print("Bench migration updates configuration and applies necessary changes.\n", emoji_code="") 

104 output.print(f"Run: [cyan]fm migrate {bench_name}[/cyan]\n", emoji_code="") 

105 raise typer.Exit(0) 

106 

107 

108# Create main Typer app 

109app = typer.Typer(no_args_is_help=True, rich_markup_mode="rich") 

110# Activate typer-examples for the main Typer app 

111install(app) 

112 

113# Register subcommands 

114app.add_typer(services_app, name="services", help="Handle global services.") 

115app.add_typer(self_app, name="self", help="Perform operations related to the [bold][blue]fm[/bold][/blue] itself.") 

116app.add_typer(ssl_app, name="ssl", help="Perform operations related to ssl.") 

117 

118 

119# App callback (runs before all commands) 

120@app.callback() 

121def app_callback( 

122 ctx: typer.Context, 

123 verbose: Annotated[bool, typer.Option("--verbose", "-v", help="Enable verbose output (info level)")] = False, 

124 log_level: Annotated[ 

125 str | None, 

126 typer.Option("--log-level", help="Set log level explicitly (debug|info|warning|error)"), 

127 ] = None, 

128 non_interactive: Annotated[ 

129 bool, 

130 typer.Option( 

131 "--non-interactive", 

132 "-n", 

133 help="Run without interactive prompts. All prompts will error with suggestions for required flags.", 

134 ), 

135 ] = False, 

136 version: Annotated[ 

137 bool | None, 

138 typer.Option("--version", "-V", help="Show Version.", callback=version_callback), 

139 ] = None, 

140): 

141 """ 

142 Docker Compose based CLI for managing Frappe benches. 

143 

144 Create, manage, and develop isolated Frappe environments using containers. 

145 Each bench runs independently with its own apps, database, and configuration. 

146 """ 

147 ctx.obj = {} 

148 

149 correlation_id = str(uuid.uuid4()) 

150 ctx.obj["correlation_id"] = correlation_id 

151 

152 # Import early for validation error reporting 

153 from frappe_manager.output_manager import get_global_output_handler, set_global_output_handler 

154 

155 # Determine effective log level 

156 if log_level: 

157 # Explicit --log-level takes precedence 

158 level_name = log_level.upper() 

159 

160 # Validate log level 

161 valid_levels = ["DEBUG", "INFO", "WARNING", "ERROR"] 

162 if level_name not in valid_levels: 

163 output = get_global_output_handler() 

164 output.display_error(f"Invalid log level: {log_level}. Must be one of: {', '.join(valid_levels).lower()}") 

165 raise typer.Exit(1) 

166 elif verbose: 

167 # -v flag sets INFO level 

168 level_name = "INFO" 

169 else: 

170 # Default: WARNING 

171 level_name = "WARNING" 

172 

173 # Store in context for commands 

174 ctx.obj["log_level"] = level_name 

175 ctx.obj["verbose"] = verbose or level_name in ["INFO", "DEBUG"] 

176 ctx.obj["non_interactive"] = non_interactive 

177 

178 # Upgrade global output handler to LoggingOutputHandler now that we have CLI args 

179 from frappe_manager.logger import ContextualLogger 

180 

181 basic_handler = get_global_output_handler() 

182 logger_context = LoggerContext(correlation_id=correlation_id) 

183 contextual_logger = ContextualLogger(log.get_logger(file_level="DEBUG"), context=logger_context) 

184 upgraded_handler = LoggingOutputHandler(basic_handler, contextual_logger) 

185 set_global_output_handler(upgraded_handler) 

186 

187 ctx.obj["logger"] = contextual_logger 

188 

189 output = get_global_output_handler() 

190 output.set_interactive_mode(non_interactive_flag=non_interactive) 

191 

192 help_called = is_cli_help_called(ctx) 

193 ctx.obj["is_help_called"] = help_called 

194 

195 if not help_called: 

196 output = get_global_output_handler() 

197 with spinner(output, "Working"): 

198 if not CLI_DIR.exists(): 

199 CLI_DIR.mkdir(parents=True, exist_ok=True) 

200 CLI_BENCHES_DIRECTORY.mkdir(parents=True, exist_ok=True) 

201 output.print(f"fm directory doesn't exists! Created at -> {CLI_DIR!s}") 

202 elif not CLI_DIR.is_dir(): 

203 output.exit("Sites directory is not a directory! Aborting!") 

204 

205 global logger 

206 console_level = level_name if ctx.obj["verbose"] else None 

207 

208 fm_config_manager: FMConfigManager = FMConfigManager.import_from_toml() 

209 file_level = fm_config_manager.logs.file_level 

210 

211 logger = log.get_logger(console_level=console_level, file_level=file_level) 

212 

213 contextual_logger.logger = logger 

214 

215 logger.info("") 

216 logger.info(f"{':' * 20}FM Invoked{':' * 20}") 

217 logger.info("") 

218 

219 logger.info(f"RUNNING COMMAND: {' '.join(sys.argv[1:])}") 

220 logger.info(f"LOG LEVEL: {level_name}") 

221 logger.info("-" * 20) 

222 

223 if not DockerClient().server_running(): 

224 output.exit("Docker daemon not running. Please start docker service") 

225 

226 if not CLI_FM_CONFIG_PATH.exists(): 

227 output.print("First installation detected. Pulling docker images...️", "🔍") 

228 

229 completed_status = pull_docker_images() 

230 

231 if not completed_status: 

232 if CLI_DIR.exists(): 

233 shutil.rmtree(CLI_DIR) 

234 output.exit("Aborting. Not able to pull all required Docker images") 

235 

236 current_version = Version(get_current_fm_version()) 

237 fm_config_manager.version = current_version 

238 fm_config_manager.export_to_toml() 

239 

240 invoked_command = ctx.invoked_subcommand or "no-command" 

241 

242 from frappe_manager.migration_manager.migration_constants import ( 

243 MIGRATION_CHECK_WHITELIST_BENCH_COMMANDS, 

244 MIGRATION_CHECK_WHITELIST_COMMANDS, 

245 ) 

246 

247 def get_full_command_path() -> str: 

248 """ 

249 Build full command path from sys.argv for multi-level commands. 

250 

251 Multi-level commands (self, ssl, services) have subcommands and return paths like "ssl add". 

252 Single-level commands take arguments (start, stop, create) and return just the base command. 

253 Stops parsing at flags (--) or path-like arguments (/ or ~). Limits depth to 2 levels max. 

254 """ 

255 # Commands that have subcommands (multi-level structure) 

256 MULTI_LEVEL_COMMANDS = {"self", "ssl", "services"} 

257 

258 if len(sys.argv) < 2: 

259 return invoked_command 

260 

261 first_command = sys.argv[1] if len(sys.argv) > 1 else invoked_command 

262 

263 # If it's not a multi-level command, return just the first command 

264 if first_command not in MULTI_LEVEL_COMMANDS: 

265 return first_command if not first_command.startswith("-") else invoked_command 

266 

267 # For multi-level commands, build the full path (max 2 levels) 

268 command_parts = [] 

269 for arg in sys.argv[1:]: 

270 # Stop at flags 

271 if arg.startswith("-"): 

272 break 

273 # Stop at paths (likely bench names like /path or ~/path) 

274 if arg.startswith("/") or arg.startswith("~"): 

275 break 

276 # Limit to max 2 command levels (e.g., "self compose") 

277 if len(command_parts) >= 2: 

278 break 

279 

280 command_parts.append(arg) 

281 

282 return " ".join(command_parts) if command_parts else invoked_command 

283 

284 full_command = get_full_command_path() 

285 

286 commands_skip_migration_check = MIGRATION_CHECK_WHITELIST_COMMANDS 

287 

288 commands_skip_bench_migration = ["stop", "delete"] + MIGRATION_CHECK_WHITELIST_BENCH_COMMANDS 

289 

290 # Get bench argument if present 

291 bench_arg = get_bench_arg_from_context(ctx) 

292 bench_path = CLI_BENCHES_DIRECTORY / bench_arg if bench_arg else None 

293 

294 # Check migration states 

295 fm_infrastructure_version = fm_config_manager.get_system_migration_version() 

296 current_version = Version(get_current_fm_version()) 

297 infra_needs_migration = fm_infrastructure_version < current_version 

298 

299 bench_needs_migration_flag = False 

300 bench_version = None 

301 if bench_path and bench_path.exists() and invoked_command not in commands_skip_bench_migration: 

302 bench_needs_migration_flag = bench_needs_migration(bench_path, current_version) 

303 if bench_needs_migration_flag: 

304 bench_version = get_bench_migration_version(bench_path) 

305 

306 should_check_migration = ( 

307 invoked_command not in commands_skip_migration_check 

308 and full_command not in commands_skip_migration_check 

309 ) 

310 

311 if should_check_migration: 

312 output = get_global_output_handler() 

313 

314 # Scenario 1: Infra needs migration 

315 if infra_needs_migration: 

316 output.warning( 

317 f"FM infrastructure needs update: v{fm_infrastructure_version} -> v{current_version}", 

318 ) 

319 output.print("This updates CLI config and global services", emoji_code=" ") 

320 output.print("", emoji_code="") 

321 

322 with temporary_stop(output): 

323 infra_choice = output.prompt_ask( 

324 prompt="How would you like to proceed?", 

325 choices=[ 

326 {"name": "Update now (recommended)", "value": "update"}, 

327 {"name": "Update later (run 'fm migrate' when ready)", "value": "skip"}, 

328 ], 

329 default="update", 

330 required_flag="'fm migrate' (run migration explicitly)", 

331 ) 

332 

333 if infra_choice == "update": 

334 output.print("\n🔄 Updating FM infrastructure...\n", emoji_code="") 

335 

336 migrations = MigrationExecutor( 

337 fm_config_manager, 

338 migrate_fm_infrastructure=True, 

339 auto_proceed=True, 

340 on_failure="rollback", 

341 output_handler=output, 

342 ) 

343 

344 with temporary_stop(output): 

345 migration_status = migrations.execute() 

346 

347 if not migration_status: 

348 output.display_error("FM infrastructure update failed") 

349 output.print("Please run 'fm migrate' manually to fix.", emoji_code="") 

350 raise typer.Exit(1) 

351 

352 fm_config_manager.set_system_migration_version(current_version) 

353 fm_config_manager.export_to_toml() 

354 

355 output.print(f"FM infrastructure updated to v{current_version}\n", emoji_code="✅ ") 

356 

357 # Now check bench migration if bench arg present 

358 if bench_needs_migration_flag and bench_arg and bench_version: 

359 output.warning( 

360 f"Bench '{bench_arg}' needs migration: v{bench_version} -> v{current_version}", 

361 ) 

362 output.print("This may modify bench configuration and services.", emoji_code="") 

363 output.print("", emoji_code="") 

364 

365 with temporary_stop(output): 

366 bench_choice = output.prompt_ask( 

367 prompt=f"Migrate bench '{bench_arg}' now?", 

368 choices=[ 

369 {"name": "Update now", "value": "update"}, 

370 { 

371 "name": f"Update later (run 'fm migrate {bench_arg}' when ready)", 

372 "value": "skip", 

373 }, 

374 ], 

375 default="update", 

376 required_flag=f"'fm migrate {bench_arg}' (run migration explicitly)", 

377 ) 

378 

379 if bench_choice == "update": 

380 output.print(f"\nMigrating bench '{bench_arg}'...\n", emoji_code="🔄 ") 

381 

382 bench_migrations = MigrationExecutor( 

383 fm_config_manager, 

384 target_benches=[bench_arg], 

385 auto_proceed=True, 

386 on_failure="rollback", 

387 output_handler=output, 

388 ) 

389 

390 with temporary_stop(output): 

391 bench_status = bench_migrations.execute() 

392 

393 if not bench_status: 

394 output.display_error(f"Bench migration failed for '{bench_arg}'") 

395 output.print(f"Please run 'fm migrate {bench_arg}' manually.", emoji_code="") 

396 raise typer.Exit(1) 

397 

398 set_bench_migration_version(bench_path, current_version) # type: ignore[arg-type] 

399 output.print(f"Bench '{bench_arg}' migrated to v{current_version}\n", emoji_code="✅ ") 

400 else: 

401 output.print("", emoji_code="") 

402 output.warning(f"Skipped bench migration. Run 'fm migrate {bench_arg}' when ready.") 

403 output.print("Note: Bench may not work correctly until migrated.", emoji_code="") 

404 output.print("", emoji_code="") 

405 output.display_error(f"Cannot {invoked_command} '{bench_arg}' - migration required") 

406 output.print(f"Run 'fm migrate {bench_arg}' first", emoji_code="") 

407 raise typer.Exit(1) 

408 

409 else: 

410 output.display_error("Cannot proceed - FM infrastructure migration required") 

411 output.print("Run 'fm migrate' when ready", emoji_code="") 

412 raise typer.Exit(1) 

413 

414 # Scenario 2: Only bench needs migration (infra already up-to-date) 

415 elif bench_needs_migration_flag and bench_arg and bench_version: 

416 output.warning(f"Bench '{bench_arg}' needs migration: v{bench_version} -> v{current_version}") 

417 output.print("This may modify bench configuration and services.", emoji_code="") 

418 output.print("", emoji_code="") 

419 

420 with temporary_stop(output): 

421 bench_choice = output.prompt_ask( 

422 prompt=f"Migrate bench '{bench_arg}' now?", 

423 choices=[ 

424 {"name": "Update now", "value": "update"}, 

425 {"name": f"Update later (run 'fm migrate {bench_arg}' when ready)", "value": "skip"}, 

426 ], 

427 default="update", 

428 required_flag=f"'fm migrate {bench_arg}' (run migration explicitly)", 

429 ) 

430 

431 if bench_choice == "update": 

432 output.print(f"\nMigrating bench '{bench_arg}'...\n", emoji_code="🔄 ") 

433 

434 bench_migrations = MigrationExecutor( 

435 fm_config_manager, 

436 target_benches=[bench_arg], 

437 auto_proceed=True, 

438 on_failure="rollback", 

439 output_handler=output, 

440 ) 

441 

442 with temporary_stop(output): 

443 bench_status = bench_migrations.execute() 

444 

445 if not bench_status: 

446 output.display_error(f"Bench migration failed for '{bench_arg}'") 

447 output.print(f"Please run 'fm migrate {bench_arg}' manually.", emoji_code="") 

448 raise typer.Exit(1) 

449 

450 set_bench_migration_version(bench_path, current_version) # type: ignore[arg-type] 

451 output.print(f"Bench '{bench_arg}' migrated to v{current_version}\n", emoji_code="✅ ") 

452 else: 

453 output.print("", emoji_code="") 

454 output.warning(f"Skipped bench migration. Run 'fm migrate {bench_arg}' when ready.") 

455 output.print("Note: Bench may not work correctly until migrated.", emoji_code="") 

456 output.print("", emoji_code="") 

457 output.display_error(f"Cannot {invoked_command} '{bench_arg}' - migration required") 

458 output.print(f"Run 'fm migrate {bench_arg}' first", emoji_code="") 

459 raise typer.Exit(1) 

460 

461 services_manager: ServicesManager = ServicesManager( 

462 verbose=ctx.obj["verbose"], 

463 invoked_subcommand=ctx.invoked_subcommand, 

464 ) 

465 

466 services_manager.init() 

467 

468 # Don't start services for migrate command (migration handles its own service lifecycle) 

469 should_start_services = invoked_command != "migrate" 

470 

471 try: 

472 services_manager.entrypoint_checks(start=should_start_services) 

473 except ServicesNotCreated as e: 

474 services_manager.remove_itself() 

475 output.exit(f"Not able to create services. {e}") 

476 

477 ctx.obj["services"] = services_manager 

478 ctx.obj["fm_config_manager"] = fm_config_manager 

479 

480 

481# Import extracted read-only commands (Step 3) 

482# Import extracted remaining commands (Step 6) 

483from frappe_manager.commands.code import code 

484 

485# Import extracted complex commands (Step 5) 

486from frappe_manager.commands.create import create 

487from frappe_manager.commands.delete import delete 

488from frappe_manager.commands.info import info 

489from frappe_manager.commands.list import list as list_benches 

490from frappe_manager.commands.logs import logs 

491from frappe_manager.commands.migrate import migrate 

492from frappe_manager.commands.ngrok import ngrok 

493from frappe_manager.commands.reset import reset 

494from frappe_manager.commands.restart import restart 

495from frappe_manager.commands.shell import shell 

496 

497# Import extracted lifecycle commands (Step 4) 

498from frappe_manager.commands.start import start 

499from frappe_manager.commands.stop import stop 

500from frappe_manager.commands.update import update 

501 

502# Register all commands with the app 

503app.command(name="create", no_args_is_help=True)(create) 

504app.command(name="delete")(delete) 

505app.command(name="list")(list_benches) 

506app.command(name="start")(start) 

507app.command(name="stop")(stop) 

508app.command(name="code")(code) 

509app.command(name="logs")(logs) 

510app.command(name="shell", context_settings={"allow_extra_args": True, "ignore_unknown_options": True})(shell) 

511app.command(name="info")(info) 

512app.command(name="update", no_args_is_help=True)(update) 

513app.command(name="reset")(reset) 

514app.command(name="restart")(restart) 

515app.command(name="ngrok")(ngrok) 

516app.command(name="migrate")(migrate) 

517 

518# Export app and helpers for backward compatibility 

519__all__ = ["app", "app_callback", "check_bench_migration_required", "get_bench_arg_from_context"]