Coverage for src/ezcompiler/services/pipeline_service.py: 87.78%
68 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-04 11:22 +0000
1# ///////////////////////////////////////////////////////////////
2# PIPELINE_SERVICE - Build pipeline orchestration helpers
3# Project: ezcompiler
4# ///////////////////////////////////////////////////////////////
6"""
7Pipeline service - Compilation, ZIP and upload orchestration.
9This service extracts the compile->zip->upload workflow from interfaces
10so the orchestration logic remains reusable and testable.
11"""
13from __future__ import annotations
15# ///////////////////////////////////////////////////////////////
16# IMPORTS
17# ///////////////////////////////////////////////////////////////
18# Standard library imports
19import shutil
20from collections.abc import Callable
21from pathlib import Path
22from typing import Any, Literal, cast
24# Local imports
25from ..shared import CompilationResult, CompilerConfig
26from .compiler_service import CompilerService
27from .installer_service import InstallerService
28from .release_service import ReleaseService
29from .uploader_service import UploaderService
31# ///////////////////////////////////////////////////////////////
32# CLASSES
33# ///////////////////////////////////////////////////////////////
36class PipelineService:
37 """Service that coordinates compile, zip and upload stages."""
39 def __init__(
40 self,
41 compiler_service_factory: (
42 Callable[[CompilerConfig], CompilerService] | None
43 ) = None,
44 ) -> None:
45 """Initialise the pipeline service.
47 Args:
48 compiler_service_factory: Optional factory to create a CompilerService
49 from a CompilerConfig. Defaults to ``CompilerService`` constructor.
50 Inject a custom factory in tests to avoid triggering real compilation.
51 """
52 self._compiler_service_factory: Callable[[CompilerConfig], CompilerService] = (
53 compiler_service_factory or CompilerService
54 )
56 def compile_project(
57 self,
58 config: CompilerConfig,
59 console: bool = True,
60 compiler: str | None = None,
61 ) -> tuple[CompilerService, CompilationResult]:
62 """Compile a project and return service + result."""
63 compiler_service = self._compiler_service_factory(config)
64 compilation_result = compiler_service.compile(
65 console=console,
66 compiler=cast(
67 Literal["Cx_Freeze", "PyInstaller", "Nuitka"] | None,
68 compiler,
69 ),
70 )
71 return compiler_service, compilation_result
73 def zip_artifact(
74 self,
75 config: CompilerConfig,
76 compiler_service: CompilerService,
77 compilation_result: CompilationResult | None,
78 progress_callback: Callable[[str, int], None] | None = None,
79 ) -> bool:
80 """Create ZIP artifact when required and return True when created."""
81 zip_needed = compilation_result.zip_needed if compilation_result else True
82 if not zip_needed:
83 return False
85 compiler_service._zip_artifact(
86 output_path=str(config.zip_file_path),
87 progress_callback=progress_callback,
88 )
89 return True
91 @staticmethod
92 def build_stages(
93 config: CompilerConfig,
94 should_zip: bool = False,
95 should_upload: bool = False,
96 should_release: bool = False,
97 should_installer: bool = False,
98 ) -> list[dict[str, Any]]:
99 """
100 Build the stage list for dynamic_layered_progress.
102 Args:
103 config: Compiler configuration (used for display labels)
104 should_zip: Whether a ZIP stage should be included
105 should_upload: Whether an upload stage should be included
107 Returns:
108 list[dict]: Stage configuration list ready for dynamic_layered_progress
109 """
110 stages: list[dict[str, Any]] = [
111 {
112 "name": "main",
113 "type": "main",
114 "description": f"Building {config.project_name} v{config.version}",
115 },
116 {
117 "name": "version",
118 "type": "spinner",
119 "description": "Generating version file",
120 },
121 {
122 "name": "compile",
123 "type": "spinner",
124 "description": f"Compiling with {config.compiler}",
125 },
126 ]
127 if should_zip:
128 stages.append(
129 {
130 "name": "zip",
131 "type": "progress",
132 "description": "Creating ZIP archive",
133 "total": 100,
134 }
135 )
136 if should_installer:
137 stages.append(
138 {
139 "name": "installer",
140 "type": "spinner",
141 "description": "Building installer",
142 }
143 )
144 if should_release:
145 stages.append(
146 {
147 "name": "release",
148 "type": "spinner",
149 "description": "Building TUF release",
150 }
151 )
152 if should_upload:
153 stages.append(
154 {
155 "name": "upload",
156 "type": "spinner",
157 "description": "Uploading artifacts",
158 }
159 )
160 return stages
162 def upload_artifact(
163 self,
164 config: CompilerConfig,
165 structure: str,
166 destination: str,
167 compilation_result: CompilationResult | None,
168 upload_config: dict[str, Any] | None = None,
169 ) -> None:
170 """Upload project artifact to a destination."""
171 zip_needed = compilation_result.zip_needed if compilation_result else True
172 source_file = (
173 str(config.zip_file_path) if zip_needed else str(config.output_folder)
174 )
176 UploaderService.upload(
177 source_path=Path(source_file),
178 upload_type=cast(Literal["disk", "server"], structure),
179 destination=destination,
180 upload_config=upload_config,
181 )
183 @staticmethod
184 def assemble_release_dir(config: CompilerConfig) -> Path:
185 """Assemble le dossier release contenant le zip et l'installeur.
187 Layout (nettoyé à chaque run)::
189 release/
190 ├── <App>.zip (si le fichier existe)
191 └── <App>-<version>-setup.exe (si installer_enabled=True)
193 L'arbre TUF (metadata/ + targets/) reste dans tufup_repo_dir et est
194 poussé directement vers le backend d'update par upload().
196 Args:
197 config: Configuration (fournit output_folder et zip_file_path).
199 Returns:
200 Path: Le dossier ``release/`` assemblé.
201 """
202 release_dir = config.output_folder.parent / "release"
203 if release_dir.exists(): 203 ↛ 204line 203 didn't jump to line 204 because the condition on line 203 was never true
204 shutil.rmtree(release_dir)
205 release_dir.mkdir(parents=True)
207 zip_path = Path(config.zip_file_path)
208 if zip_path.is_file():
209 shutil.copy2(zip_path, release_dir / zip_path.name)
211 if config.installer_enabled: 211 ↛ 212line 211 didn't jump to line 212 because the condition on line 211 was never true
212 installer_dir = config.installer_output_dir or (
213 config.output_folder.parent / "installer"
214 )
215 installer_exe = (
216 installer_dir / f"{config.project_name}-{config.version}-setup.exe"
217 )
218 if installer_exe.is_file():
219 shutil.copy2(installer_exe, release_dir / installer_exe.name)
221 return release_dir
223 @staticmethod
224 def release_artifact(
225 config: CompilerConfig,
226 compilation_result: CompilationResult | None, # noqa: ARG004
227 ) -> Path:
228 """Build le repo TUF local depuis output_folder. Ne publie jamais."""
229 repo_dir = config.tuf_repo_dir or (config.output_folder / "repo")
230 keys_dir = config.tuf_keys_dir or (repo_dir / "keystore")
231 return ReleaseService.release_and_publish(
232 bundle_dir=config.output_folder,
233 app_name=config.project_name,
234 version=config.version,
235 repo_dir=repo_dir,
236 publish=False,
237 releaser_config={
238 "keys_dir": keys_dir,
239 "expiration_days": config.tuf_expiration_days,
240 },
241 )
243 @staticmethod
244 def build_installer(
245 config: CompilerConfig,
246 compilation_result: CompilationResult | None, # noqa: ARG004
247 ) -> Path | None:
248 """Build the Inno Setup installer when installer_enabled=True."""
249 if not config.installer_enabled:
250 return None
252 output_dir = config.installer_output_dir or (
253 config.output_folder.parent / "installer"
254 )
255 installer_config: dict[str, Any] = {
256 "icon": config.icon,
257 "company_name": config.company_name,
258 "per_user": config.installer_per_user,
259 }
260 if config.installer_iss_path is not None: 260 ↛ 261line 260 didn't jump to line 261 because the condition on line 260 was never true
261 installer_config["iss_path"] = config.installer_iss_path
263 return InstallerService.build_installer(
264 bundle_dir=config.output_folder,
265 app_name=config.project_name,
266 version=config.version,
267 output_dir=output_dir,
268 installer_config=installer_config,
269 )