Coverage for src/ezcompiler/services/uploader_service.py: 74.36%
122 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-16 16:37 +0000
1# ///////////////////////////////////////////////////////////////
2# UPLOADER_SERVICE - Upload orchestration service
3# Project: ezcompiler
4# ///////////////////////////////////////////////////////////////
6"""
7Uploader service - Upload orchestration service for EzCompiler.
9This module provides the UploaderService class that orchestrates file
10and directory uploads using different upload backends (disk, server).
12Services layer can use WARNING and ERROR log levels.
13"""
15from __future__ import annotations
17# ///////////////////////////////////////////////////////////////
18# IMPORTS
19# ///////////////////////////////////////////////////////////////
20# Standard library imports
21import shutil
22import tempfile
23from collections.abc import Callable, Iterator
24from contextlib import contextmanager
25from pathlib import Path
26from typing import TYPE_CHECKING, Any, Literal
28# Local imports
29from ..adapters import UploaderFactory
30from ..shared._constants import RELEASE_SUBDIR, TUF_PUBLIC_DIRS, UPDATE_SUBDIR
31from ..shared.exceptions import UploadError
32from ..utils.validators import validate_upload_structure
33from .tuf_service import WITHDRAWN_FILE, TufService
35if TYPE_CHECKING: 35 ↛ 36line 35 didn't jump to line 36 because the condition on line 35 was never true
36 from .._types import UploaderPort
37 from ..shared._compiler_config import CompilerConfig
39# ///////////////////////////////////////////////////////////////
40# TYPE ALIASES
41# ///////////////////////////////////////////////////////////////
43UploadType = Literal["disk", "server", "r2"]
45# ///////////////////////////////////////////////////////////////
46# CONSTANTS
47# ///////////////////////////////////////////////////////////////
49# Default keystore name, excluded from staging whatever its position.
50DEFAULT_KEYSTORE_DIRNAME = "keystore"
52# ///////////////////////////////////////////////////////////////
53# CLASSES
54# ///////////////////////////////////////////////////////////////
57class UploaderService:
58 """
59 Upload orchestration service.
61 Orchestrates file and directory uploads using different upload backends.
62 Handles upload type selection, validation, and execution.
64 Example:
65 >>> service = UploaderService()
66 >>> service.upload(
67 ... source_path=Path("dist.zip"),
68 ... upload_type="disk",
69 ... destination="releases/",
70 ... upload_config={"overwrite": True}
71 ... )
72 """
74 # ////////////////////////////////////////////////
75 # UPLOAD METHODS
76 # ////////////////////////////////////////////////
78 @staticmethod
79 def upload(
80 source_path: Path,
81 upload_type: UploadType,
82 destination: str,
83 upload_config: dict[str, Any] | None = None,
84 ) -> None:
85 """
86 Upload a file or directory to the specified destination.
88 Args:
89 source_path: Path to the source file or directory
90 upload_type: Type of upload ("disk" or "server")
91 destination: Destination path or URL
92 upload_config: Additional uploader configuration options
94 Raises:
95 UploadError: If upload fails or upload type is invalid
97 Example:
98 >>> UploaderService.upload(
99 ... Path("dist.zip"),
100 ... "disk",
101 ... "releases/",
102 ... {"overwrite": True}
103 ... )
104 """
105 try:
106 # Validate upload type
107 if not validate_upload_structure(upload_type): 107 ↛ 108line 107 didn't jump to line 108 because the condition on line 107 was never true
108 raise UploadError(f"Invalid upload type: {upload_type}")
110 # Copy: the caller's dict is reused for the following uploads,
111 # an in-place mutation would contaminate the next backend.
112 config = dict(upload_config or {})
113 if upload_type == "disk":
114 config["destination_path"] = destination
115 elif upload_type == "server": 115 ↛ 120line 115 didn't jump to line 120 because the condition on line 115 was always true
116 config["server_url"] = destination
117 # r2: bucket comes from upload_config, destination = object prefix
119 # Create uploader and perform upload
120 uploader: UploaderPort = UploaderFactory.create_uploader(
121 upload_type, config
122 )
123 uploader.upload(source_path=source_path, destination=destination)
124 except UploadError:
125 raise
126 except Exception as e:
127 raise UploadError(f"Upload failed: {str(e)}") from e
129 @staticmethod
130 def download(
131 remote_source: str,
132 upload_type: UploadType,
133 destination_local: Path,
134 upload_config: dict[str, Any] | None = None,
135 ) -> None:
136 """Download a remote tree into ``destination_local`` via the backend.
138 Args:
139 remote_source: Remote source (path, URL or prefix) to fetch.
140 upload_type: Backend type ("disk", "server" or "r2").
141 destination_local: Local directory to populate.
142 upload_config: Additional uploader configuration options.
144 Raises:
145 UploadError: If the download fails or the type is invalid.
146 """
147 try:
148 uploader: UploaderPort = UploaderFactory.create_uploader(
149 upload_type, upload_config or {}
150 )
151 uploader.download(remote_source, destination_local)
152 except UploadError:
153 raise
154 except Exception as e:
155 raise UploadError(f"Download failed: {str(e)}") from e
157 @staticmethod
158 def upload_release(
159 config: CompilerConfig,
160 repo_dir: Path,
161 release_root: Path | None,
162 destination: str | None = None,
163 repo_destination: str | None = None,
164 release_destination: str | None = None,
165 upload_config: dict[str, Any] | None = None,
166 ) -> None:
167 """Sequential double upload: TUF tree then installer zip.
169 Step 1 — upload the TUF tree to ``<dest>/update/`` (or the R2 prefix).
170 Step 2 — upload the installer zip to ``<dest>/release/`` (skipped on R2).
172 Note: The double upload is not atomic. If the zip upload fails, the
173 TUF repo is already online. On failure, run upload() again to
174 resume.
176 Args:
177 config: CompilerConfig holding the destinations and R2 options.
178 repo_dir: Local directory of the TUF repo.
179 release_root: Local directory of the installer zip (None on R2).
180 destination: Shared override for both destinations.
181 repo_destination: Override for ``config.repo_destination``.
182 release_destination: Override for ``config.release_destination``.
183 upload_config: Extra options passed to the uploaders.
185 Raises:
186 UploadError: If an upload fails.
187 """
188 repo_dest = repo_destination or config.repo_destination
189 rel_dest = release_destination or config.release_destination
191 UploaderService.upload_tuf_repo(
192 config, repo_dir, repo_dest, destination, upload_config
193 )
195 if release_root is not None:
196 UploaderService.upload_release_zip(
197 config, release_root, rel_dest, destination, upload_config
198 )
200 @staticmethod
201 def upload_tuf_repo(
202 config: CompilerConfig,
203 repo_dir: Path,
204 repo_dest: str,
205 destination: str | None,
206 upload_config: dict[str, Any] | None,
207 ) -> None:
208 """Upload the public part of the TUF tree to the configured destination.
210 Only ``metadata/``, ``targets/`` and ``withdrawn.json`` are published,
211 from a temporary copy (see ``staged_tuf_tree``): the private keys
212 stored under ``repo_dir`` are never transferred.
213 """
214 try:
215 with UploaderService.staged_tuf_tree(
216 repo_dir, keys_dir=TufService.keys_dir(config)
217 ) as staged:
218 UploaderService._upload_tuf_tree(
219 config, staged, repo_dest, destination, upload_config
220 )
221 except UploadError as e:
222 raise UploadError(f"TUF repo upload failed: {e}") from e
224 @staticmethod
225 @contextmanager
226 def staged_tuf_tree(
227 repo_dir: Path, *, keys_dir: Path | None = None
228 ) -> Iterator[Path]:
229 """Copy the public files of the TUF tree into a temporary directory.
231 The layout is preserved (``metadata/``, ``targets/``,
232 ``withdrawn.json`` at the root); the directory is removed on exit.
233 The keystore is always excluded from the copy, even when it sits
234 under ``metadata/`` or ``targets/``.
236 Args:
237 repo_dir: Root of the local TUF tree.
238 keys_dir: Private keystore to exclude, on top of any directory
239 named ``keystore``.
241 Raises:
242 UploadError: If ``repo_dir/metadata`` is missing (publishing an
243 empty tree would overwrite the remote channel) or if the copy
244 fails.
245 """
246 if not (repo_dir / "metadata").is_dir():
247 raise UploadError(
248 f"TUF tree missing: {repo_dir / 'metadata'} not found. "
249 "Run `ezcompiler tuf init` first, then the build."
250 )
251 with tempfile.TemporaryDirectory(prefix="ezcompiler-tuf-") as tmp:
252 staging = Path(tmp)
253 try:
254 ignore = UploaderService._ignore_private_keys(keys_dir)
255 for name in TUF_PUBLIC_DIRS:
256 if (repo_dir / name).is_dir():
257 shutil.copytree(repo_dir / name, staging / name, ignore=ignore)
258 withdrawn = repo_dir / WITHDRAWN_FILE
259 if withdrawn.is_file(): 259 ↛ 260line 259 didn't jump to line 260 because the condition on line 259 was never true
260 shutil.copy2(withdrawn, staging / WITHDRAWN_FILE)
261 except OSError as e:
262 raise UploadError(f"Copie de l'arbre TUF impossible : {e}") from e
263 yield staging
265 @staticmethod
266 def _ignore_private_keys(
267 keys_dir: Path | None,
268 ) -> Callable[[str, list[str]], set[str]]:
269 """Build the ``ignore`` filter for ``shutil.copytree``.
271 Args:
272 keys_dir: Configured private keystore, when known.
274 Returns:
275 Callable: Filter excluding the configured keystore as well as any
276 directory named ``keystore``.
277 """
278 resolved = keys_dir.expanduser().resolve() if keys_dir else None
280 def ignore(src: str, names: list[str]) -> set[str]:
281 excluded: set[str] = set()
282 for name in names:
283 if name == DEFAULT_KEYSTORE_DIRNAME: 283 ↛ 284line 283 didn't jump to line 284 because the condition on line 283 was never true
284 excluded.add(name)
285 continue
286 if resolved is None:
287 continue
288 candidate = (Path(src) / name).resolve()
289 if candidate == resolved or resolved in candidate.parents: 289 ↛ 290line 289 didn't jump to line 290 because the condition on line 289 was never true
290 excluded.add(name)
291 return excluded
293 return ignore
295 @staticmethod
296 def _upload_tuf_tree(
297 config: CompilerConfig,
298 staged_dir: Path,
299 repo_dest: str,
300 destination: str | None,
301 upload_config: dict[str, Any] | None,
302 ) -> None:
303 """Route an already-filtered TUF tree to the configured backend."""
304 if repo_dest == "r2":
305 endpoint = config.repo_endpoint
306 bucket, _, prefix = endpoint.partition("/")
307 UploaderService.upload(
308 source_path=staged_dir,
309 upload_type="r2",
310 destination=prefix,
311 upload_config={"bucket": bucket},
312 )
313 elif repo_dest == "server": 313 ↛ 314line 313 didn't jump to line 314 because the condition on line 313 was never true
314 base = destination or config.resolved_repo_destination or ""
315 UploaderService.upload(
316 source_path=staged_dir,
317 upload_type="server",
318 destination=base.rstrip("/") + f"/{UPDATE_SUBDIR}",
319 upload_config=upload_config,
320 )
321 else: # disk (default)
322 base = destination or config.resolved_repo_destination or ""
323 UploaderService.upload(
324 source_path=staged_dir,
325 upload_type="disk",
326 destination=str(Path(base) / UPDATE_SUBDIR),
327 upload_config=upload_config,
328 )
330 @staticmethod
331 def upload_release_zip(
332 config: CompilerConfig,
333 release_root: Path,
334 rel_dest: str,
335 destination: str | None,
336 upload_config: dict[str, Any] | None,
337 ) -> None:
338 """Upload the installer zip to the configured destination."""
339 try:
340 if rel_dest == "r2": 340 ↛ 341line 340 didn't jump to line 341 because the condition on line 340 was never true
341 endpoint = config.release_endpoint
342 bucket, _, prefix = endpoint.partition("/")
343 UploaderService.upload(
344 source_path=release_root,
345 upload_type="r2",
346 destination=prefix,
347 upload_config={"bucket": bucket},
348 )
349 elif rel_dest == "server": 349 ↛ 350line 349 didn't jump to line 350 because the condition on line 349 was never true
350 base = destination or config.resolved_release_destination or ""
351 UploaderService.upload(
352 source_path=release_root,
353 upload_type="server",
354 destination=base.rstrip("/") + f"/{RELEASE_SUBDIR}",
355 upload_config=upload_config,
356 )
357 else: # disk (default)
358 base = destination or config.resolved_release_destination or ""
359 UploaderService.upload(
360 source_path=release_root,
361 upload_type="disk",
362 destination=str(Path(base) / RELEASE_SUBDIR),
363 upload_config=upload_config,
364 )
365 except UploadError as e:
366 raise UploadError(f"Release zip upload failed: {e}") from e
368 # ////////////////////////////////////////////////
369 # UTILITY METHODS
370 # ////////////////////////////////////////////////
372 @staticmethod
373 def get_supported_types() -> list[str]:
374 """
375 Get list of supported upload types.
377 Returns:
378 list[str]: List of supported upload type names
380 Example:
381 >>> types = UploaderService.get_supported_types()
382 >>> print(types)
383 ['disk', 'server']
384 """
385 return UploaderFactory.get_supported_types()
387 @staticmethod
388 def validate_upload_config(
389 upload_type: UploadType, config: dict[str, Any] | None = None
390 ) -> bool:
391 """
392 Validate configuration for a specific upload type.
394 Args:
395 upload_type: Type of uploader to validate
396 config: Configuration to validate (default: None)
398 Returns:
399 bool: True if configuration is valid, False otherwise
401 Example:
402 >>> is_valid = UploaderService.validate_upload_config(
403 ... "disk", {"overwrite": True}
404 ... )
405 """
406 return UploaderFactory.validate_config(upload_type, config)