Coverage for src / ezcompiler / services / pipeline_service.py: 66.67%
33 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-27 06:49 +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
19from collections.abc import Callable
20from pathlib import Path
21from typing import Any
23from ..shared.compilation_result import CompilationResult
25# Local imports
26from ..shared.compiler_config import CompilerConfig
27from .compiler_service import CompilerService
28from .uploader_service import UploaderService
30# ///////////////////////////////////////////////////////////////
31# CLASSES
32# ///////////////////////////////////////////////////////////////
35class PipelineService:
36 """Service that coordinates compile, zip and upload stages."""
38 def __init__(
39 self,
40 compiler_service_factory: (
41 Callable[[CompilerConfig], CompilerService] | None
42 ) = None,
43 ) -> None:
44 """Initialise the pipeline service.
46 Args:
47 compiler_service_factory: Optional factory to create a CompilerService
48 from a CompilerConfig. Defaults to ``CompilerService`` constructor.
49 Inject a custom factory in tests to avoid triggering real compilation.
50 """
51 self._compiler_service_factory: Callable[[CompilerConfig], CompilerService] = (
52 compiler_service_factory or CompilerService
53 )
55 def compile_project(
56 self,
57 config: CompilerConfig,
58 console: bool = True,
59 compiler: str | None = None,
60 ) -> tuple[CompilerService, CompilationResult]:
61 """Compile a project and return service + result."""
62 compiler_service = self._compiler_service_factory(config)
63 compilation_result = compiler_service.compile(
64 console=console,
65 compiler=compiler, # type: ignore[arg-type]
66 )
67 return compiler_service, compilation_result
69 def zip_artifact(
70 self,
71 config: CompilerConfig,
72 compiler_service: CompilerService,
73 compilation_result: CompilationResult | None,
74 progress_callback: Callable[[str, int], None] | None = None,
75 ) -> bool:
76 """Create ZIP artifact when required and return True when created."""
77 zip_needed = (
78 compilation_result.zip_needed if compilation_result else config.zip_needed
79 )
80 if not zip_needed:
81 return False
83 compiler_service._zip_artifact(
84 output_path=str(config.zip_file_path),
85 progress_callback=progress_callback,
86 )
87 return True
89 @staticmethod
90 def build_stages(
91 config: CompilerConfig,
92 should_zip: bool = False,
93 should_upload: bool = False,
94 ) -> list[dict[str, Any]]:
95 """
96 Build the stage list for dynamic_layered_progress.
98 Args:
99 config: Compiler configuration (used for display labels)
100 should_zip: Whether a ZIP stage should be included
101 should_upload: Whether an upload stage should be included
103 Returns:
104 list[dict]: Stage configuration list ready for dynamic_layered_progress
105 """
106 stages: list[dict[str, Any]] = [
107 {
108 "name": "main",
109 "type": "main",
110 "description": f"Building {config.project_name} v{config.version}",
111 },
112 {
113 "name": "version",
114 "type": "spinner",
115 "description": "Generating version file",
116 },
117 {
118 "name": "compile",
119 "type": "spinner",
120 "description": f"Compiling with {config.compiler}",
121 },
122 ]
123 if should_zip:
124 stages.append(
125 {
126 "name": "zip",
127 "type": "progress",
128 "description": "Creating ZIP archive",
129 "total": 100,
130 }
131 )
132 if should_upload:
133 stages.append(
134 {
135 "name": "upload",
136 "type": "spinner",
137 "description": "Uploading artifacts",
138 }
139 )
140 return stages
142 def upload_artifact(
143 self,
144 config: CompilerConfig,
145 structure: str,
146 destination: str,
147 compilation_result: CompilationResult | None,
148 upload_config: dict[str, Any] | None = None,
149 ) -> None:
150 """Upload project artifact to a destination."""
151 zip_needed = (
152 compilation_result.zip_needed if compilation_result else config.zip_needed
153 )
154 source_file = (
155 str(config.zip_file_path) if zip_needed else str(config.output_folder)
156 )
158 UploaderService.upload(
159 source_path=Path(source_file),
160 upload_type=structure, # type: ignore[arg-type]
161 destination=destination,
162 upload_config=upload_config,
163 )