Coverage for b4_backup/cli/tools.py: 100%

1"""Helpful cli commands, that are too specific to be in the main part of the code. 

2 

3Some parts here may be untested. 

4""" 

5 

6import io 

7import json 

8import logging 

9 

10import rich 

11import typer 

12from omegaconf import OmegaConf 

13from rich.syntax import Syntax 

14 

15from b4_backup import config_schema 

16from b4_backup.cli import utils 

17from b4_backup.main import backup_target_host, dataclass 

18from b4_backup.main.b4_backup import B4Backup 

19 

20log = logging.getLogger("b4_backup.cli") 

21 

22app = typer.Typer(help=__doc__) 

23 

24 

25@app.command() 

26def dump_config(ctx: typer.Context): 

27 """Return the fully interpolated configuration. For debugging.""" 

28 config: config_schema.BaseConfig = ctx.obj 

29 

30 rich.print(Syntax(OmegaConf.to_yaml(config), "yaml", line_numbers=True)) 

31 

32 

33@app.command() 

34def update_config( # pragma: no cover 

35 ctx: typer.Context, 

36 new_targets: str = typer.Argument(..., help="JSON object containing 'target_name: source'"), 

37 dry_run: bool = typer.Option( 

38 False, help="Just print the new config instead of actually updating the config file" 

39 ), 

40 delete_source: bool = typer.Option(False, help="Delete all backups on source side"), 

41): 

42 """Updates the b4 config based on the new targets list parameter. 

43 

44 Targets that are not mentioned in the new targets list, aren't removed, 

45 but the source attribute will become None, until every backup is eventually deleted or cleaned. 

46 

47 You need to provide the config updates in a format like this: 

48 {"localhost/test": "/new_source", "example.com/test": "ssh://root@example.com/test"} 

49 

50 ruaml.yaml required. 

51 """ 

52 from ruamel.yaml import YAML 

53 

54 yaml = YAML() 

55 

56 config: config_schema.BaseConfig = ctx.obj 

57 b4_backup = B4Backup(config.timezone) 

58 

59 with utils.error_handler(): 

60 config_yaml = yaml.load(config.config_path) 

61 new_target_objs = json.loads(new_targets) 

62 

63 target_choice = dataclass.ChoiceSelector(list(config.backup_targets)) 

64 passive_targets = { 

65 dst_host.name: len(dst_host.snapshots()) 

66 for _none, dst_host in backup_target_host.host_generator( 

67 target_choice, config.backup_targets, use_source=False 

68 ) 

69 if dst_host 

70 } 

71 

72 for target_name, source in new_target_objs.items(): 

73 if target_name in passive_targets: 

74 config_yaml["backup_targets"][target_name]["source"] = source 

75 else: 

76 config_yaml["backup_targets"][target_name] = {"source": source} 

77 

78 old_targets = list(passive_targets.keys() - new_target_objs.keys()) 

79 old_targets_choice = dataclass.ChoiceSelector(old_targets) 

80 for src_host, dst_host in backup_target_host.host_generator( 

81 old_targets_choice, config.backup_targets 

82 ): 

83 if not dst_host: 

84 continue 

85 

86 if delete_source and src_host: 

87 log.info("Removing all backups on source side") 

88 

89 if not dry_run: 

90 b4_backup.delete_all(src_host) 

91 

92 if passive_targets[dst_host.name] > 0: 

93 config_yaml["backup_targets"][dst_host.name]["source"] = None 

94 else: 

95 del config_yaml["backup_targets"][dst_host.name] 

96 

97 config_yaml["default_targets"] = list(new_target_objs) 

98 

99 strio = io.StringIO() 

100 yaml.dump(config_yaml, strio) 

101 strio.seek(0) 

102 new_config_yaml = strio.read() 

103 

104 if dry_run: 

105 rich.print(Syntax(new_config_yaml, "yaml", line_numbers=True)) 

106 else: 

107 config.config_path.write_text(new_config_yaml)