Coverage for src / ezqt_app / services / bootstrap / sequence.py: 70.40%
97 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-26 07:07 +0000
1# ///////////////////////////////////////////////////////////////
2# SERVICES.BOOTSTRAP.SEQUENCE - Initialization sequence orchestrator
3# Project: ezqt_app
4# ///////////////////////////////////////////////////////////////
6"""Explicit step-by-step initialization sequence with status tracking."""
8from __future__ import annotations
10# ///////////////////////////////////////////////////////////////
11# IMPORTS
12# ///////////////////////////////////////////////////////////////
13# Standard library imports
14from collections.abc import Callable
16# Local imports
17from ...domain.errors import EzQtError
18from ...domain.results import InitResult, InitStepResult
19from ...domain.results.result_error import ResultError
20from ...utils.printer import get_printer
21from .contracts.options import InitOptions
22from .contracts.steps import InitStep, StepStatus
25# ///////////////////////////////////////////////////////////////
26# CLASSES
27# ///////////////////////////////////////////////////////////////
28class InitializationSequence:
29 """Orchestrates the ordered initialization steps for EzQt_App.
31 Each step is registered with a name, description, callable and a
32 *required* flag. If a required step fails the sequence stops
33 immediately; non-required steps are allowed to fail silently.
34 """
36 def __init__(self, options: InitOptions | None = None) -> None:
37 self.options = (options or InitOptions()).resolve()
38 self.steps: list[InitStep] = []
39 self.current_step: InitStep | None = None
40 self.printer = get_printer(self.options.verbose)
42 # Register the resolved bin path early so all runtime services
43 # (themes, fonts, config, translations) use the correct directory.
44 if self.options.bin_path is not None: 44 ↛ 49line 44 didn't jump to line 49 because the condition on line 44 was always true
45 from ...utils.runtime_paths import set_bin_path
47 set_bin_path(self.options.bin_path)
49 self._setup_steps()
51 # ------------------------------------------------------------------
52 # Error code
53 # ------------------------------------------------------------------
55 def _error_code_for_step(self, step_name: str) -> str:
56 """Return a normalized error code for a failed step."""
57 slug = step_name.strip().lower().replace(" ", "_")
58 return f"bootstrap.step.{slug}.failed"
60 # ------------------------------------------------------------------
61 # Default steps
62 # ------------------------------------------------------------------
64 def _setup_steps(self) -> None:
65 """Register the default EzQt_App boot steps."""
66 from ..application.app_service import AppService
67 from ..application.file_service import FileService
68 from .startup_config import StartupConfig
70 options = self.options
72 self.add_step(
73 name="Configure Startup",
74 description="Configure UTF-8 encoding, locale, and environment variables",
75 function=lambda: StartupConfig().configure(options.project_root),
76 required=True,
77 )
78 self.add_step(
79 name="Create Directories",
80 description="Create necessary directories for assets and config",
81 function=lambda: FileService(
82 base_path=options.project_root,
83 bin_path=options.bin_path,
84 overwrite_policy=options.overwrite_policy.value,
85 ).make_assets_binaries(),
86 required=True,
87 )
88 self.add_step(
89 name="Copy Configurations",
90 description="Copy package configuration files to project bin/config directory",
91 function=lambda: AppService.copy_package_configs_to_project(),
92 required=False,
93 )
94 self.add_step(
95 name="Generate Files",
96 description="Generate required configuration and resource files",
97 function=lambda: AppService.make_required_files(
98 mk_theme=options.mk_theme,
99 mk_config=options.mk_config,
100 mk_translations=options.mk_translations,
101 base_path=options.project_root,
102 bin_path=options.bin_path,
103 overwrite_policy=options.overwrite_policy.value,
104 ),
105 required=True,
106 )
107 self.add_step(
108 name="Check Requirements",
109 description="Verify that all required assets and dependencies are available",
110 function=lambda: (
111 AppService.check_assets_requirements(
112 base_path=options.project_root,
113 bin_path=options.bin_path,
114 overwrite_policy=options.overwrite_policy.value,
115 )
116 if options.build_resources
117 else None
118 ),
119 required=True,
120 )
121 self.add_step(
122 name="Load Resources",
123 description="Register compiled Qt resources (bin/resources_rc.py)",
124 function=lambda: (
125 __import__(
126 "ezqt_app.shared.resources",
127 fromlist=["load_runtime_rc"],
128 ).load_runtime_rc()
129 if options.build_resources
130 else None
131 ),
132 required=False,
133 )
135 # ------------------------------------------------------------------
136 # Step management
137 # ------------------------------------------------------------------
139 def add_step(
140 self,
141 name: str,
142 description: str,
143 function: Callable[[], None],
144 required: bool = True,
145 ) -> None:
146 """Append a new step to the sequence."""
147 self.steps.append(
148 InitStep(
149 name=name, description=description, function=function, required=required
150 )
151 )
153 # ------------------------------------------------------------------
154 # Execution
155 # ------------------------------------------------------------------
157 def execute(self, verbose: bool = True) -> InitResult:
158 """Run all registered steps in order.
160 Returns
161 -------
162 InitResult
163 Typed aggregate result with step-level execution details.
164 """
165 import time
167 if verbose: 167 ↛ 168line 167 didn't jump to line 168 because the condition on line 167 was never true
168 self.printer.custom_print(
169 "~ [Initializer] Starting EzQt_App Initialization Sequence",
170 color="MAGENTA",
171 )
172 self.printer.raw_print("...")
174 start_time = time.time()
175 successful_steps = 0
176 failed_steps = 0
177 skipped_steps = 0
178 first_error: Exception | None = None
179 first_failed_step: str | None = None
181 for step in self.steps:
182 self.current_step = step
183 step_start = time.time()
184 step.status = StepStatus.RUNNING
186 try:
187 step.function()
188 step.status = StepStatus.SUCCESS
189 step.duration = time.time() - step_start
190 successful_steps += 1
191 except Exception as e:
192 step.status = StepStatus.FAILED
193 step.error_message = str(e)
194 step.duration = time.time() - step_start
195 failed_steps += 1
196 if first_error is None: 196 ↛ 200line 196 didn't jump to line 200 because the condition on line 196 was always true
197 first_error = e
198 first_failed_step = step.name
200 if verbose: 200 ↛ 201line 200 didn't jump to line 201 because the condition on line 200 was never true
201 self.printer.error(
202 f"[Initializer] Step failed ({step.duration:.2f}s): {e}"
203 )
205 if step.required:
206 if verbose: 206 ↛ 207line 206 didn't jump to line 207 because the condition on line 206 was never true
207 self.printer.error(
208 f"[Initializer] Initialization failed at required step: {step.name}"
209 )
210 break
212 total_time = time.time() - start_time
213 summary = InitResult(
214 success=(failed_steps == 0),
215 total_steps=len(self.steps),
216 successful=successful_steps,
217 failed=failed_steps,
218 skipped=skipped_steps,
219 total_time=total_time,
220 steps=[
221 InitStepResult(
222 name=step.name,
223 description=step.description,
224 required=step.required,
225 status=step.status.value,
226 error_message=step.error_message,
227 duration=step.duration,
228 )
229 for step in self.steps
230 ],
231 )
233 if first_error is not None and first_failed_step is not None:
234 if isinstance(first_error, EzQtError): 234 ↛ 235line 234 didn't jump to line 235 because the condition on line 234 was never true
235 summary.error = ResultError(
236 code=first_error.code,
237 message=first_error.message,
238 context=first_error.context,
239 )
240 else:
241 summary.error = ResultError(
242 code=self._error_code_for_step(first_failed_step),
243 message=str(first_error),
244 context={
245 "step": first_failed_step,
246 "error_type": type(first_error).__name__,
247 },
248 )
250 if verbose: 250 ↛ 251line 250 didn't jump to line 251 because the condition on line 250 was never true
251 self._print_summary(summary)
253 return summary
255 def _print_summary(self, summary: InitResult) -> None:
256 self.printer.raw_print("...")
257 if summary.success:
258 self.printer.custom_print(
259 "~ [Initializer] Initialization completed successfully!",
260 color="MAGENTA",
261 )
262 else:
263 self.printer.custom_print(
264 "~ [Initializer] Initialization failed!", color="MAGENTA"
265 )
267 # ------------------------------------------------------------------
268 # Introspection helpers
269 # ------------------------------------------------------------------
271 def get_step_status(self, step_name: str) -> StepStatus | None:
272 for step in self.steps:
273 if step.name == step_name: 273 ↛ 272line 273 didn't jump to line 272 because the condition on line 273 was always true
274 return step.status
275 return None
277 def get_failed_steps(self) -> list[InitStep]:
278 return [s for s in self.steps if s.status == StepStatus.FAILED]
280 def get_successful_steps(self) -> list[InitStep]:
281 return [s for s in self.steps if s.status == StepStatus.SUCCESS]
283 def reset(self) -> None:
284 for step in self.steps:
285 step.status = StepStatus.PENDING
286 step.error_message = None
287 step.duration = None
288 self.current_step = None