Coverage for src/ezcompiler/services/updater_service.py: 82.72%

1# /////////////////////////////////////////////////////////////// 

2# UPDATER_SERVICE - Update client generation service 

3# Project: ezcompiler 

4# /////////////////////////////////////////////////////////////// 

5 

6""" 

7Updater service - Generates auto-update client files for compiled apps. 

8 

9Produces update.py, settings.py, and copies root.json from the local 

10TUF repository. Files are meant to be embedded in the compiled bundle 

11via include_files. 

12""" 

13 

14from __future__ import annotations 

15 

16import shutil 

17from pathlib import Path 

18 

19from ..shared import CompilerConfig 

20from ..shared._constants import UPDATE_SUBDIR 

21from ..shared.exceptions import UpdaterConfigError, UpdaterGenerationError 

22 

23_TEMPLATES_DIR = Path(__file__).parent.parent / "assets" / "templates" / "updater" 

24_ROOT_JSON = "root.json" 

25 

26 

27class UpdaterService: 

28 """Service that generates the tufup client update files.""" 

29 

30 @staticmethod 

31 def generate(config: CompilerConfig, output_dir: Path) -> list[Path]: 

32 """Generate update.py, settings.py, and copy root.json. 

33 

34 Args: 

35 config: Compiler configuration (must have tuf_enabled=True). 

36 output_dir: Directory where files are written (created if absent). 

37 

38 Returns: 

39 List of generated file paths [settings.py, update.py, root.json]. 

40 

41 Raises: 

42 UpdaterConfigError: If config is invalid (tuf_enabled=False, 

43 root.json absent, repo_public_url missing for non-disk). 

44 UpdaterGenerationError: If writing files fails. 

45 """ 

46 UpdaterService._validate(config) 

47 update_url = UpdaterService._resolve_update_url(config) 

48 

49 try: 

50 output_dir.mkdir(parents=True, exist_ok=True) 

51 generated: list[Path] = [] 

52 generated.append( 

53 UpdaterService._render_template( 

54 "settings", config, update_url, output_dir 

55 ) 

56 ) 

57 generated.append( 

58 UpdaterService._render_template( 

59 "update", config, update_url, output_dir 

60 ) 

61 ) 

62 generated.append(UpdaterService._copy_root_json(config, output_dir)) 

63 return generated 

64 except (UpdaterConfigError, UpdaterGenerationError): 

65 raise 

66 except Exception as exc: 

67 raise UpdaterGenerationError( 

68 f"Failed to generate updater files in {output_dir}: {exc}" 

69 ) from exc 

70 

71 @staticmethod 

72 def _validate(config: CompilerConfig) -> None: 

73 if not config.tuf_enabled: 

74 raise UpdaterConfigError( 

75 "tuf_enabled must be True to generate updater files. " 

76 "Set tuf_enabled=True in your config." 

77 ) 

78 if config.tuf_repo_dir is None: 78 ↛ 79line 78 didn't jump to line 79 because the condition on line 78 was never true

79 raise UpdaterConfigError("tuf_repo_dir is required to locate root.json.") 

80 root_json = config.tuf_repo_dir / "metadata" / _ROOT_JSON 

81 if not root_json.exists(): 

82 raise UpdaterConfigError( 

83 f"root.json not found at {root_json}. " 

84 "Run 'ezcompiler release init' first to initialise the TUF repository." 

85 ) 

86 

87 @staticmethod 

88 def _resolve_update_url(config: CompilerConfig) -> str: 

89 # The client URL must mirror where UploaderService writes the TUF 

90 # tree: disk/server place it under the UPDATE_SUBDIR subdir, while r2 

91 # uploads straight to the bucket prefix. Sharing the constant keeps 

92 # both sides in sync. 

93 if config.repo_destination == "disk": 

94 endpoint = config.repo_endpoint.rstrip("/").replace("\\", "/") 

95 if not endpoint.startswith("/"): 95 ↛ 96line 95 didn't jump to line 96 because the condition on line 95 was never true

96 endpoint = f"/{endpoint}" 

97 return f"file://{endpoint}/{UPDATE_SUBDIR}" 

98 if config.repo_destination == "r2": 

99 return config.repo_public_url.rstrip("/") 

100 # server 

101 return config.repo_public_url.rstrip("/") + f"/{UPDATE_SUBDIR}" 

102 

103 @staticmethod 

104 def _render_template( 

105 name: str, 

106 config: CompilerConfig, 

107 update_url: str, 

108 output_dir: Path, 

109 ) -> Path: 

110 tmpl_path = _TEMPLATES_DIR / f"{name}.py.template" 

111 try: 

112 content = tmpl_path.read_text(encoding="utf-8") 

113 except OSError as exc: 

114 raise UpdaterGenerationError( 

115 f"Cannot read template {tmpl_path}: {exc}" 

116 ) from exc 

117 

118 content = content.replace("#APP_NAME#", config.project_name) 

119 content = content.replace("#VERSION#", config.version) 

120 content = content.replace("#UPDATE_URL#", update_url) 

121 

122 dest = output_dir / f"{name}.py" 

123 try: 

124 dest.write_text(content, encoding="utf-8") 

125 except OSError as exc: 

126 raise UpdaterGenerationError(f"Cannot write {dest}: {exc}") from exc 

127 return dest 

128 

129 @staticmethod 

130 def _copy_root_json(config: CompilerConfig, output_dir: Path) -> Path: 

131 assert config.tuf_repo_dir is not None # guaranteed by _validate 

132 src = config.tuf_repo_dir / "metadata" / _ROOT_JSON 

133 dst = output_dir / _ROOT_JSON 

134 try: 

135 shutil.copy2(src, dst) 

136 except OSError as exc: 

137 raise UpdaterGenerationError( 

138 f"Cannot copy root.json from {src} to {dst}: {exc}" 

139 ) from exc 

140 return dst