Coverage for enderchest/cli.py: 84%

164 statements  

« prev     ^ index     » next       coverage.py v7.4.1, created at 2024-02-06 16:00 +0000

1"""Command-line interface""" 

2import argparse 

3import inspect 

4import logging 

5import os 

6import sys 

7from argparse import ArgumentParser, RawTextHelpFormatter 

8from pathlib import Path 

9from typing import Any, Iterable, Protocol, Sequence 

10 

11from . import craft, gather, loggers, place, remote, uninstall 

12from ._version import get_versions 

13 

14# mainly because I think I'm gonna forget what names are canonical (it's the first ones) 

15_create_aliases = ("craft", "create") 

16_instance_aliases = tuple( 

17 alias + plural for alias in ("instance", "minecraft") for plural in ("", "s") 

18) 

19_shulker_box_aliases = ("shulker_box", "shulkerbox", "shulker") 

20_remote_aliases = tuple( 

21 alias + plural for alias in ("enderchest", "remote") for plural in ("s", "") 

22) 

23_list_aliases = ("inventory", "list") 

24 

25 

26class Action(Protocol): # pragma: no cover 

27 """Common protocol for CLI actions""" 

28 

29 def __call__(self, minecraft_root: Path, /) -> Any: 

30 ... 

31 

32 

33def _place( 

34 minecraft_root: Path, 

35 errors: str = "prompt", 

36 keep_broken_links: bool = False, 

37 keep_stale_links: bool = False, 

38 keep_level: int = 0, 

39 stop_at_first_failure: bool = False, 

40 ignore_errors: bool = False, 

41 absolute: bool = False, 

42 relative: bool = False, 

43) -> None: 

44 """Wrapper sort through all the various argument groups""" 

45 

46 if stop_at_first_failure: 

47 errors = "abort" 

48 if ignore_errors: # elif? 

49 errors = "ignore" 

50 # else: errors = errors 

51 

52 if absolute is True: 

53 # technically we get this for free already 

54 relative = False 

55 

56 if keep_level > 0: 

57 keep_stale_links = True 

58 if keep_level > 1: 

59 keep_broken_links = True 

60 

61 place.cache_placements( 

62 minecraft_root, 

63 place.place_ender_chest( 

64 minecraft_root, 

65 keep_broken_links=keep_broken_links, 

66 keep_stale_links=keep_stale_links, 

67 error_handling=errors, 

68 relative=relative, 

69 ), 

70 ) 

71 

72 

73def _craft_shulker_box(minecraft_root: Path, name: str | None = None, **kwargs): 

74 """Wrapper to handle the fact that name is a required argument""" 

75 assert name # it's required by the parser, so this should be fine 

76 craft.craft_shulker_box(minecraft_root, name, **kwargs) 

77 

78 

79def _list_instance_boxes( 

80 minecraft_root: Path, 

81 instance_name: str | None = None, 

82 path: str | None = None, 

83 **kwargs, 

84): 

85 """Wrapper to route --path flag and instance_name arg""" 

86 if path is not None: 

87 place.list_placements( 

88 minecraft_root, pattern=path, instance_name=instance_name, **kwargs 

89 ) 

90 elif instance_name is not None: 

91 gather.get_shulker_boxes_matching_instance( 

92 minecraft_root, instance_name, **kwargs 

93 ) 

94 else: 

95 gather.load_shulker_boxes(minecraft_root, **kwargs) 

96 

97 

98def _list_shulker_box( 

99 minecraft_root: Path, shulker_box_name: str | None = None, **kwargs 

100): 

101 """Wrapper to handle the fact that name is a required argument""" 

102 assert shulker_box_name # it's required by the parser, so this should be fine 

103 gather.get_instances_matching_shulker_box( 

104 minecraft_root, shulker_box_name, **kwargs 

105 ) 

106 

107 

108def _update_ender_chest( 

109 minecraft_root: Path, 

110 official: bool | None = None, 

111 mmc: bool | None = None, 

112 **kwargs, 

113): 

114 """Wrapper to resolve the official vs. MultiMC flag""" 

115 if mmc: 

116 official = False 

117 gather.update_ender_chest(minecraft_root, official=official, **kwargs) 

118 

119 

120def _open(minecraft_root: Path, verbosity: int = 0, **kwargs): 

121 """Router for open verb""" 

122 remote.sync_with_remotes(minecraft_root, "pull", verbosity=verbosity, **kwargs) 

123 

124 

125def _close(minecraft_root: Path, verbosity: int = 0, **kwargs): 

126 """Router for close verb""" 

127 remote.sync_with_remotes(minecraft_root, "push", verbosity=verbosity, **kwargs) 

128 

129 

130def _test( 

131 minecraft_root: Path, use_local_ssh: bool = False, pytest_args: Iterable[str] = () 

132): 

133 """Run the EnderChest test suite to ensure that it is running correctly on your 

134 system. Requires you to have installed GSB with the test extra 

135 (i.e. pipx install enderchest[test]).""" 

136 import pytest 

137 

138 from enderchest.test import plugin 

139 

140 if use_local_ssh: 

141 pytest_args = ("--use-local-ssh", *pytest_args) 

142 if exit_code := pytest.main( 

143 ["--pyargs", "enderchest.test", *pytest_args], 

144 plugins=(plugin,), 

145 ): 

146 raise SystemExit(f"Tests Failed with exit code: {exit_code}") 

147 

148 

149ACTIONS: tuple[tuple[tuple[str, ...], str, Action], ...] = ( 

150 # action names (first one is canonical), action description, action method 

151 ( 

152 sum(((verb, verb + " enderchest") for verb in _create_aliases), ()), 

153 "create and configure a new EnderChest installation", 

154 craft.craft_ender_chest, 

155 ), 

156 ( 

157 tuple( 

158 f"{verb} {alias}" 

159 for verb in _create_aliases 

160 for alias in _shulker_box_aliases 

161 ), 

162 "create and configure a new shulker box", 

163 _craft_shulker_box, 

164 ), 

165 ( 

166 ("place",), 

167 "link (or update the links) from your instances to your EnderChest", 

168 _place, 

169 ), 

170 ( 

171 tuple("gather " + alias for alias in _instance_aliases), 

172 "register (or update the registry of) a Minecraft installation", 

173 _update_ender_chest, 

174 ), 

175 ( 

176 tuple("gather " + alias for alias in _remote_aliases), 

177 "register (or update the registry of) a remote EnderChest", 

178 _update_ender_chest, 

179 ), 

180 ( 

181 # I freely admit this is ridiculous 

182 sum( 

183 ( 

184 ( 

185 verb, 

186 *( 

187 f"{verb} {alias}" 

188 # pluralization is hard 

189 for alias in ("shulker_boxes", "shulkerboxes", "shulkers") 

190 ), 

191 ) 

192 for verb in _list_aliases 

193 ), 

194 (), 

195 ), 

196 "list the shulker boxes inside your Enderchest", 

197 _list_instance_boxes, 

198 ), 

199 ( 

200 tuple( 

201 f"{verb} {alias}" 

202 for verb in _list_aliases 

203 for alias in _instance_aliases 

204 if alias.endswith("s") 

205 ), 

206 "list the minecraft instances registered with your Enderchest", 

207 gather.load_ender_chest_instances, 

208 ), 

209 ( 

210 tuple( 

211 f"{verb} {alias}" 

212 for verb in _list_aliases 

213 for alias in _instance_aliases 

214 if not alias.endswith("s") 

215 ), 

216 "list the shulker boxes that the specified instance links into", 

217 _list_instance_boxes, 

218 ), 

219 ( 

220 tuple( 

221 f"{verb} {alias}" 

222 for verb in _list_aliases 

223 for alias in _shulker_box_aliases 

224 ), 

225 "list the minecraft instances that match the specified shulker box", 

226 _list_shulker_box, 

227 ), 

228 ( 

229 tuple(f"{verb} {alias}" for verb in _list_aliases for alias in _remote_aliases), 

230 "list the other EnderChest installations registered with this EnderChest", 

231 gather.load_ender_chest_remotes, 

232 ), 

233 ( 

234 ("open",), 

235 "pull changes from other EnderChests", 

236 _open, 

237 ), 

238 ( 

239 ("close",), 

240 "push changes to other EnderChests", 

241 _close, 

242 ), 

243 ( 

244 ("break",), 

245 "uninstall EnderChest by copying all linked resources" 

246 " into its registered instances", 

247 uninstall.break_ender_chest, 

248 ), 

249 ( 

250 ("test",), 

251 "run the EnderChest test suite", 

252 _test, 

253 ), 

254) 

255 

256 

257def generate_parsers() -> tuple[ArgumentParser, dict[str, ArgumentParser]]: 

258 """Generate the command-line parsers 

259 

260 Returns 

261 ------- 

262 enderchest_parser : ArgumentParser 

263 The top-level argument parser responsible for routing arguments to 

264 specific action parsers 

265 action_parsers : dict of str to ArgumentParser 

266 The verb-specific argument parsers 

267 """ 

268 descriptions: dict[str, str] = {} 

269 root_description: str = "" 

270 for commands, description, _ in ACTIONS: 

271 descriptions[commands[0]] = description 

272 root_description += f"\n\t{commands[0]}\n\t\tto {description}" 

273 

274 enderchest_parser = ArgumentParser( 

275 prog="enderchest", 

276 description=( 

277 f"v{get_versions()['version']}\n" 

278 "\nsyncing and linking for all your Minecraft instances" 

279 ), 

280 formatter_class=RawTextHelpFormatter, 

281 ) 

282 

283 enderchest_parser.add_argument( 

284 "-v", # don't worry--this doesn't actually conflict with --verbose 

285 "-V", 

286 "--version", 

287 action="version", 

288 version=f"%(prog)s v{get_versions()['version']}", 

289 ) 

290 

291 # these are really just for the sake of --help 

292 # (the parsed args aren't actually used) 

293 enderchest_parser.add_argument( 

294 "action", 

295 help=f"The action to perform. Options are:{root_description}", 

296 type=str, 

297 ) 

298 enderchest_parser.add_argument( 

299 "arguments", 

300 nargs="*", 

301 help="Any additional arguments for the specific action." 

302 " To learn more, try: enderchest {action} -h", 

303 ) 

304 

305 action_parsers: dict[str, ArgumentParser] = {} 

306 for verb, description in descriptions.items(): 

307 parser = ArgumentParser( 

308 prog=f"enderchest {verb}", 

309 description=description, 

310 ) 

311 if verb != "test": 

312 root = parser.add_mutually_exclusive_group() 

313 root.add_argument( 

314 "root", 

315 nargs="?", 

316 help=( 

317 "Optionally specify your root minecraft directory." 

318 " If no path is given, the current working directory will be used." 

319 ), 

320 type=Path, 

321 ) 

322 root.add_argument( 

323 "--root", 

324 dest="root_flag", 

325 help="specify your root minecraft directory", 

326 type=Path, 

327 ) 

328 

329 # I'm actually okay with -vvqvqqv hilarity 

330 parser.add_argument( 

331 "--verbose", 

332 "-v", 

333 action="count", 

334 default=0, 

335 help="increase the amount of information that's printed", 

336 ) 

337 parser.add_argument( 

338 "--quiet", 

339 "-q", 

340 action="count", 

341 default=0, 

342 help="decrease the amount of information that's printed", 

343 ) 

344 action_parsers[verb] = parser 

345 

346 # craft options 

347 craft_parser = action_parsers[_create_aliases[0]] 

348 craft_parser.add_argument( 

349 "--from", 

350 dest="copy_from", 

351 help=( 

352 "provide the URI (e.g. rsync://deck@my-steam-deck/home/deck/) of a" 

353 " remote EnderChest installation that can be used" 

354 " to boostrap the creation of this one." 

355 ), 

356 ) 

357 craft_parser.add_argument( 

358 "-r", 

359 "--remote", 

360 dest="remotes", 

361 action="append", 

362 help=( 

363 "provide the URI (e.g. rsync://deck@my-steam-deck/home/deck/) of a" 

364 " remote EnderChest installation to register with this one" 

365 ), 

366 ) 

367 craft_parser.add_argument( 

368 "-i", 

369 "--instance", 

370 dest="instance_search_paths", 

371 action="append", 

372 type=Path, 

373 help="specify a folder to search for Minecraft installations in", 

374 ) 

375 craft_parser.add_argument( 

376 "--overwrite", 

377 action="store_true", 

378 help=( 

379 "if there's already an EnderChest installation in this location," 

380 " overwrite its configuration" 

381 ), 

382 ) 

383 

384 # shulker box craft options 

385 shulker_craft_parser = action_parsers[ 

386 f"{_create_aliases[0]} {_shulker_box_aliases[0]}" 

387 ] 

388 shulker_craft_parser.add_argument( 

389 "name", 

390 help="specify the name for this shulker box", 

391 ) 

392 shulker_craft_parser.add_argument( 

393 "--priority", 

394 "-p", 

395 help="specify the link priority for this shulker box (higher = linked later)", 

396 ) 

397 shulker_craft_parser.add_argument( 

398 "-i", 

399 "--instance", 

400 dest="instances", 

401 action="append", 

402 help="only link instances with one of the provided names to this shulker box", 

403 ) 

404 shulker_craft_parser.add_argument( 

405 "-t", 

406 "--tag", 

407 dest="tags", 

408 action="append", 

409 help="only link instances with one of the provided tags to this shulker box", 

410 ) 

411 shulker_craft_parser.add_argument( 

412 "-e", 

413 "--enderchest", 

414 dest="hosts", 

415 action="append", 

416 help=( 

417 "only link instances registered to one of the provided EnderChest" 

418 " installations with this shulker box" 

419 ), 

420 ) 

421 shulker_craft_parser.add_argument( 

422 "-l", 

423 "--folder", 

424 dest="link_folders", 

425 action="append", 

426 help=( 

427 "specify the name of a folder inside this shulker box" 

428 " that should be linked completely" 

429 ), 

430 ) 

431 shulker_craft_parser.add_argument( 

432 "--overwrite", 

433 action="store_true", 

434 help=( 

435 "if there's already a shulker box with the specified name," 

436 " overwrite its configuration" 

437 ), 

438 ) 

439 

440 # place options 

441 place_parser = action_parsers["place"] 

442 cleanup = place_parser.add_argument_group() 

443 cleanup.add_argument( 

444 "--keep-broken-links", 

445 action="store_true", 

446 help="do not remove broken links from instances", 

447 ) 

448 cleanup.add_argument( 

449 "--keep-stale-links", 

450 action="store_true", 

451 help=( 

452 "do not remove existing links into the EnderChest," 

453 " even if the shulker box or instance spec has changed" 

454 ), 

455 ) 

456 cleanup.add_argument( 

457 "-k", 

458 dest="keep_level", 

459 action="count", 

460 default=0, 

461 help=( 

462 "shorthand for the above cleanup options:" 

463 " -k will --keep-stale-links," 

464 " and -kk will --keep-broken-links as well" 

465 ), 

466 ) 

467 error_handling = place_parser.add_argument_group( 

468 title="error handling" 

469 ).add_mutually_exclusive_group() 

470 error_handling.add_argument( 

471 "--stop-at-first-failure", 

472 "-x", 

473 action="store_true", 

474 help="stop linking at the first issue", 

475 ) 

476 error_handling.add_argument( 

477 "--ignore-errors", action="store_true", help="ignore any linking errors" 

478 ) 

479 error_handling.add_argument( 

480 "--errors", 

481 "-e", 

482 choices=( 

483 "prompt", 

484 "ignore", 

485 "skip", 

486 "skip-instance", 

487 "skip-shulker-box", 

488 "abort", 

489 ), 

490 default="prompt", 

491 help=( 

492 "specify how to handle linking errors" 

493 " (default behavior is to prompt after every error)" 

494 ), 

495 ) 

496 link_type = place_parser.add_mutually_exclusive_group() 

497 link_type.add_argument( 

498 "--absolute", 

499 "-a", 

500 action="store_true", 

501 help="use absolute paths for all link targets", 

502 ) 

503 link_type.add_argument( 

504 "--relative", 

505 "-r", 

506 action="store_true", 

507 help="use relative paths for all link targets", 

508 ) 

509 

510 # gather instance options 

511 gather_instance_parser = action_parsers[f"gather {_instance_aliases[0]}"] 

512 gather_instance_parser.add_argument( 

513 "search_paths", 

514 nargs="+", 

515 action="extend", 

516 type=Path, 

517 help="specify a folder or folders to search for Minecraft installations", 

518 ) 

519 instance_type = gather_instance_parser.add_mutually_exclusive_group() 

520 instance_type.add_argument( 

521 "--official", 

522 "-o", 

523 action="store_true", 

524 help="specify that these are instances managed by the official launcher", 

525 ) 

526 instance_type.add_argument( 

527 "--mmc", 

528 "-m", 

529 action="store_true", 

530 help="specify that these are MultiMC-like instances", 

531 ) 

532 

533 # gather remote options 

534 gather_remote_parser = action_parsers[f"gather {_remote_aliases[0]}"] 

535 gather_remote_parser.add_argument( 

536 "remotes", 

537 nargs="+", 

538 action="extend", 

539 help=( 

540 "Provide URIs (e.g. rsync://deck@my-steam-deck/home/deck/) of any" 

541 " remote EnderChest installation to register with this one." 

542 "Note: you should not use this method if the alias (name) of the" 

543 "remote does not match the remote's hostname (in this example," 

544 '"my-steam-deck").' 

545 ), 

546 ) 

547 

548 # list shulker box options 

549 

550 # list [instance] boxes options 

551 list_boxes_parser = action_parsers[f"{_list_aliases[0]}"] 

552 list_instance_boxes_parser = action_parsers[ 

553 f"{_list_aliases[0]} {_instance_aliases[0]}" 

554 ] 

555 

556 instance_name_docs = "The name of the minecraft instance to query" 

557 list_boxes_parser.add_argument( 

558 "--instance", "-i", dest="instance_name", help=instance_name_docs 

559 ) 

560 list_instance_boxes_parser.add_argument("instance_name", help=instance_name_docs) 

561 

562 for parser in (list_boxes_parser, list_instance_boxes_parser): 

563 parser.add_argument( 

564 "--path", 

565 "-p", 

566 help=( 

567 "optionally, specify a specific path" 

568 " (absolute, relative, filename or glob pattern" 

569 " to get a report of the shulker box(es) that provide that resource" 

570 ), 

571 ) 

572 

573 # list shulker options 

574 list_shulker_box_parser = action_parsers[ 

575 f"{_list_aliases[0]} {_shulker_box_aliases[0]}" 

576 ] 

577 list_shulker_box_parser.add_argument( 

578 "shulker_box_name", help="the name of the shulker box to query" 

579 ) 

580 

581 # open / close options 

582 for action in ("open", "close"): 

583 sync_parser = action_parsers[action] 

584 

585 sync_parser.add_argument( 

586 "--dry-run", 

587 action="store_true", 

588 help=( 

589 "perform a dry run of the sync operation," 

590 " reporting the operations that will be performed" 

591 " but not actually carrying them out" 

592 ), 

593 ) 

594 sync_parser.add_argument( 

595 "--exclude", 

596 "-e", 

597 action="extend", 

598 nargs="+", 

599 help="Provide any file patterns you would like to skip syncing", 

600 ) 

601 sync_parser.add_argument( 

602 "--timeout", 

603 "-t", 

604 type=int, 

605 help=( 

606 "set a maximum number of seconds to try to sync to a remote chest" 

607 " before giving up and going on to the next one" 

608 ), 

609 ) 

610 sync_confirm_wait = sync_parser.add_argument_group( 

611 title="sync confirmation control", 

612 description=( 

613 "The default behavior when syncing EnderChests is to first perform a" 

614 " dry run of every sync operation and then wait 5 seconds before" 

615 " proceeding with the real sync. The idea is to give you time to" 

616 " interrupt the sync if the dry run looks wrong. You can raise or" 

617 " lower that wait time through these flags. You can also modify it" 

618 " by editing the enderchest.cfg file." 

619 ), 

620 ).add_mutually_exclusive_group() 

621 sync_confirm_wait.add_argument( 

622 "--wait", 

623 "-w", 

624 dest="sync_confirm_wait", 

625 type=int, 

626 help="set the time in seconds to wait after performing a dry run" 

627 " before the real sync is performed", 

628 ) 

629 sync_confirm_wait.add_argument( 

630 "--confirm", 

631 "-c", 

632 dest="sync_confirm_wait", 

633 action="store_true", 

634 help="after performing the dry run, explicitly ask for confirmation" 

635 " before performing the real sync", 

636 ) 

637 

638 # test pass-through 

639 test_parser = action_parsers["test"] 

640 test_parser.add_argument( 

641 "--use-local-ssh", 

642 action="store_true", 

643 dest="use_local_ssh", 

644 help=( 

645 "By default, tests of SSH functionality will be run against a mock" 

646 " SSH server. If you are running EnderChest on a machine you can SSH" 

647 " into locally (by running `ssh localhost`) without requiring a password," 

648 " running the tests with this flag will produce more accurate results." 

649 ), 

650 ) 

651 test_parser.add_argument( 

652 "pytest_args", 

653 nargs=argparse.REMAINDER, 

654 help="any additional arguments to pass through to py.test", 

655 ) 

656 

657 return enderchest_parser, action_parsers 

658 

659 

660def parse_args(argv: Sequence[str]) -> tuple[Action, Path, int, dict[str, Any]]: 

661 """Parse the provided command-line options to determine the action to perform and 

662 the arguments to pass to the action 

663 

664 Parameters 

665 ---------- 

666 argv : list-like of str (sys.argv) 

667 The options passed into the command line 

668 

669 Returns 

670 ------- 

671 Callable 

672 The action method that will be called 

673 str 

674 The root of the minecraft folder (parent of the EnderChest) 

675 where the action will be performed 

676 int 

677 The verbosity level of the operation (in terms of log levels) 

678 dict 

679 Any additional options that will be given to the action method 

680 

681 """ 

682 actions: dict[str, Action] = {} 

683 aliases: dict[str, str] = {} 

684 for commands, _, method in ACTIONS: 

685 for command in commands: 

686 aliases[command] = commands[0] 

687 actions[commands[0]] = method 

688 

689 enderchest_parser, action_parsers = generate_parsers() 

690 

691 _ = enderchest_parser.parse_args(argv[1:2]) # check for --help and --version 

692 

693 for command in sorted(aliases.keys(), key=lambda x: -len(x)): # longest first 

694 if " ".join((*argv[1:], "")).startswith(command + " "): 

695 if command == "test": 

696 parsed, extra = action_parsers["test"].parse_known_args(argv[2:]) 

697 return ( 

698 actions["test"], 

699 Path(), 

700 0, 

701 { 

702 "use_local_ssh": parsed.use_local_ssh, 

703 "pytest_args": [*parsed.pytest_args, *extra], 

704 }, 

705 ) 

706 action_kwargs = vars( 

707 action_parsers[aliases[command]].parse_args( 

708 argv[1 + len(command.split()) :] 

709 ) 

710 ) 

711 

712 action = actions[aliases[command]] 

713 

714 root_arg = action_kwargs.pop("root") 

715 root_flag = action_kwargs.pop("root_flag") 

716 

717 verbosity = action_kwargs.pop("verbose") - action_kwargs.pop("quiet") 

718 

719 argspec = inspect.getfullargspec(action) 

720 if "verbosity" in argspec.args + argspec.kwonlyargs: 

721 action_kwargs["verbosity"] = verbosity 

722 

723 log_level = loggers.verbosity_to_log_level(verbosity) 

724 

725 MINECRAFT_ROOT = os.getenv("MINECRAFT_ROOT") 

726 

727 return ( 

728 actions[aliases[command]], 

729 Path(root_arg or root_flag or MINECRAFT_ROOT or os.getcwd()), 

730 log_level, 

731 action_kwargs, 

732 ) 

733 

734 enderchest_parser.print_help(sys.stderr) 

735 sys.exit(1) 

736 

737 

738def main(): 

739 """CLI Entrypoint""" 

740 logger = logging.getLogger(__package__) 

741 cli_handler = logging.StreamHandler() 

742 cli_handler.setFormatter(loggers.CLIFormatter()) 

743 logger.addHandler(cli_handler) 

744 

745 action, root, log_level, kwargs = parse_args(sys.argv) 

746 

747 # TODO: set log levels per logger based on the command 

748 cli_handler.setLevel(log_level) 

749 

750 # TODO: when we add log files, set this to minimum log level across all handlers 

751 logger.setLevel(log_level) 

752 

753 action(root, **kwargs)