Coverage for src/ezcompiler/services/release_service.py: 86.36%
36 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-16 16:37 +0000
1# ///////////////////////////////////////////////////////////////
2# RELEASE_SERVICE - Secure-release orchestration service
3# Project: ezcompiler
4# ///////////////////////////////////////////////////////////////
6"""
7Release service - Orchestrates secure-release packaging.
9Builds a signed TUF repository locally via a releaser adapter, then OPTIONALLY
10delegates the remote transfer of that repository tree to the existing
11``UploaderService`` (disk/server). Release packaging and transfer stay
12separate concerns.
13"""
15from __future__ import annotations
17# ///////////////////////////////////////////////////////////////
18# IMPORTS
19# ///////////////////////////////////////////////////////////////
20from pathlib import Path
21from typing import TYPE_CHECKING, Any, Literal, cast
23from ..adapters import ReleaserFactory
24from ..shared.exceptions import ReleaseError
25from .tuf_service import TufService
26from .uploader_service import UploaderService
28if TYPE_CHECKING: 28 ↛ 29line 28 didn't jump to line 29 because the condition on line 28 was never true
29 from .._types import ReleaserPort
31# ///////////////////////////////////////////////////////////////
32# CLASSES
33# ///////////////////////////////////////////////////////////////
36class ReleaseService:
37 """Service orchestrating secure-release packaging and publication."""
39 # ------------------------------------------------
40 # RELEASE METHODS
41 # ------------------------------------------------
43 @staticmethod
44 def release_and_publish(
45 bundle_dir: Path,
46 app_name: str,
47 version: str,
48 repo_dir: Path,
49 *,
50 release_type: str = "tufup",
51 publish: bool = False,
52 pull_before: bool = False,
53 upload_type: str | None = None,
54 destination: str | None = None,
55 releaser_config: dict[str, Any] | None = None,
56 upload_config: dict[str, Any] | None = None,
57 required: bool = False,
58 ) -> Path:
59 """Build the local TUF repo, then optionally publish it.
61 Args:
62 bundle_dir: Directory containing the compiled application.
63 app_name: Application name (used by tufup to name bundles).
64 version: Application version string.
65 repo_dir: Root directory for the local TUF repository tree.
66 release_type: Release backend to use (default: "tufup").
67 publish: When True, transfer the repository/ tree via an uploader.
68 pull_before: When True, download the current remote tree into
69 ``repo_dir`` before releasing (R2 source-of-truth cycle).
70 Requires upload_type and destination.
71 upload_type: Upload backend ("disk" or "server"). Required when publish=True.
72 destination: Upload destination path or URL. Required when publish=True.
73 releaser_config: Extra config forwarded to the releaser adapter.
74 upload_config: Extra config forwarded to the uploader adapter.
75 required: Mark the version as mandatory for clients.
77 Returns:
78 Path: The local ``repository/`` tree path.
80 Raises:
81 ValueError: When publish=True but upload_type or destination is missing.
82 ReleaseError: When the version is not above a withdrawn version, or
83 when release packaging or publishing fails.
84 """
85 if pull_before and upload_type and destination:
86 UploaderService.download(
87 remote_source=destination,
88 upload_type=cast(Literal["disk", "server", "r2"], upload_type),
89 destination_local=repo_dir,
90 upload_config=upload_config,
91 )
93 # After the pull: withdrawn.json travels with the remote tree.
94 TufService.ensure_releasable(repo_dir, version)
95 # First release after a withdrawal: full archive only, a patch
96 # would start from an archive the withdrawn clients do not have.
97 patch = not TufService.needs_full_archive(repo_dir, app_name)
99 releaser: ReleaserPort = ReleaserFactory.create_releaser(
100 release_type, releaser_config
101 )
102 repository_path = releaser.release(
103 bundle_dir=bundle_dir,
104 app_name=app_name,
105 version=version,
106 repo_dir=repo_dir,
107 patch=patch,
108 required=required,
109 )
111 if not publish:
112 return repository_path
114 if not upload_type or not destination:
115 raise ValueError("publish=True requires both upload_type and destination")
117 try:
118 # Never the raw tree: the default keystore lives under repo_dir.
119 with UploaderService.staged_tuf_tree(
120 repository_path, keys_dir=(releaser_config or {}).get("keys_dir")
121 ) as staged:
122 UploaderService.upload(
123 source_path=staged,
124 upload_type=cast(Literal["disk", "server", "r2"], upload_type),
125 destination=destination,
126 upload_config=upload_config,
127 )
128 except Exception as exc:
129 raise ReleaseError(f"Publishing release repository failed: {exc}") from exc
131 return repository_path
133 @staticmethod
134 def init_release(
135 app_name: str,
136 repo_dir: Path,
137 keys_dir: Path,
138 *,
139 release_type: str = "tufup",
140 releaser_config: dict[str, Any] | None = None,
141 ) -> bool:
142 """Create the releaser through the factory and delegate to init_keys.
144 Returns True if the init ran, False if it was already present (skip).
145 """
146 releaser: ReleaserPort = ReleaserFactory.create_releaser(
147 release_type, releaser_config
148 )
149 return releaser.init_keys(
150 app_name=app_name, repo_dir=repo_dir, keys_dir=keys_dir
151 )
153 @staticmethod
154 def refresh_expiration(
155 app_name: str,
156 repo_dir: Path,
157 keys_dir: Path,
158 *,
159 roles: tuple[str, ...] = ("targets", "snapshot", "timestamp"),
160 days: int | None = None,
161 release_type: str = "tufup",
162 releaser_config: dict[str, Any] | None = None,
163 ) -> Path:
164 """Re-sign the metadata to push back expiration without a new release.
166 Native tufup keep-alive for projects updated irregularly.
168 Returns the local TUF repo directory.
169 """
170 releaser: ReleaserPort = ReleaserFactory.create_releaser(
171 release_type, releaser_config
172 )
173 return releaser.refresh_expiration(
174 app_name=app_name,
175 repo_dir=repo_dir,
176 keys_dir=keys_dir,
177 roles=roles,
178 days=days,
179 )