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