Coverage for src / domain / validation / spec_result_builder.py: 33%
89 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-01-04 04:43 +0000
1"""Result building stage for mala validation.
3This module provides SpecResultBuilder which handles the post-command-execution
4stages of validation:
5- Coverage checks (threshold or no-decrease mode)
6- E2E execution (run-level only)
7- ValidationResult assembly
9The builder is called after commands have been executed and produces the final
10ValidationResult.
11"""
13from __future__ import annotations
15from dataclasses import dataclass
16from typing import TYPE_CHECKING
18from .coverage import (
19 CoverageResult,
20 CoverageStatus,
21 check_coverage_from_config,
22 parse_and_check_coverage,
23)
24from .e2e import E2EConfig as E2ERunnerConfig
25from .e2e import E2ERunner, E2EStatus
26from .result import ValidationResult
28from pathlib import Path
30if TYPE_CHECKING:
31 from collections.abc import Mapping
33 from src.core.protocols import CommandRunnerPort, EnvConfigPort
35 from .config import YamlCoverageConfig
36 from .e2e import E2EResult
37 from .result import ValidationStepResult
38 from .spec import (
39 CoverageConfig,
40 E2EConfig,
41 ValidationArtifacts,
42 ValidationContext,
43 ValidationSpec,
44 )
47@dataclass
48class ResultBuilderInput:
49 """Input for the result builder stage.
51 Attributes:
52 spec: The validation spec being executed.
53 context: The validation context.
54 steps: Steps from command execution.
55 artifacts: Validation artifacts to update.
56 cwd: Working directory where coverage.xml should be.
57 log_dir: Directory for logs.
58 env: Environment variables for E2E.
59 baseline_percent: Baseline coverage for "no decrease" mode.
60 env_config: Environment configuration for paths.
61 command_runner: Command runner for executing commands.
62 yaml_coverage_config: Coverage configuration from mala.yaml, or None to
63 use spec-based coverage checking (legacy mode).
64 """
66 spec: ValidationSpec
67 context: ValidationContext
68 steps: list[ValidationStepResult]
69 artifacts: ValidationArtifacts
70 cwd: Path
71 log_dir: Path
72 env: Mapping[str, str]
73 baseline_percent: float | None
74 env_config: EnvConfigPort
75 command_runner: CommandRunnerPort
76 yaml_coverage_config: YamlCoverageConfig | None = None
79class SpecResultBuilder:
80 """Builds ValidationResult from command execution output.
82 This builder handles the post-processing stages after commands have been
83 executed:
84 1. Coverage check (if enabled)
85 2. E2E execution (if enabled and scope is run-level)
86 3. Final ValidationResult assembly
88 The builder owns the coverage and E2E checking logic, keeping the runner
89 focused on workspace setup and command execution.
90 """
92 def build(self, input: ResultBuilderInput) -> ValidationResult:
93 """Build the final ValidationResult.
95 Args:
96 input: The builder input containing steps, artifacts, and config.
98 Returns:
99 Complete ValidationResult with coverage/E2E results.
100 """
101 # Step 1: Check coverage
102 cov = self._check_coverage_if_enabled(
103 spec=input.spec,
104 cwd=input.cwd,
105 log_dir=input.log_dir,
106 artifacts=input.artifacts,
107 baseline_percent=input.baseline_percent,
108 env=input.env,
109 yaml_coverage_config=input.yaml_coverage_config,
110 command_runner=input.command_runner,
111 )
112 if cov is not None and not cov.passed:
113 reason = cov.failure_reason or "Coverage check failed"
114 return self._build_failure_result(
115 steps=input.steps,
116 reason=reason,
117 artifacts=input.artifacts,
118 coverage_result=cov,
119 )
121 # Step 2: Run E2E
122 e2e = self._run_e2e_if_enabled(
123 spec=input.spec,
124 env=input.env,
125 cwd=input.cwd,
126 log_dir=input.log_dir,
127 artifacts=input.artifacts,
128 env_config=input.env_config,
129 command_runner=input.command_runner,
130 )
131 if e2e is not None and not e2e.passed and e2e.status != E2EStatus.SKIPPED:
132 reason = e2e.failure_reason or "E2E failed"
133 return self._build_failure_result(
134 steps=input.steps,
135 reason=reason,
136 artifacts=input.artifacts,
137 coverage_result=cov,
138 e2e_result=e2e,
139 )
141 # Step 3: Success
142 return ValidationResult(
143 passed=True,
144 steps=input.steps,
145 artifacts=input.artifacts,
146 coverage_result=cov,
147 e2e_result=e2e,
148 )
150 def _check_coverage_if_enabled(
151 self,
152 spec: ValidationSpec,
153 cwd: Path,
154 log_dir: Path,
155 artifacts: ValidationArtifacts,
156 baseline_percent: float | None,
157 env: Mapping[str, str],
158 command_runner: CommandRunnerPort,
159 yaml_coverage_config: YamlCoverageConfig | None = None,
160 ) -> CoverageResult | None:
161 """Run coverage check if enabled.
163 Args:
164 spec: Validation spec with coverage config.
165 cwd: Working directory where coverage.xml should be.
166 log_dir: Directory for logs.
167 artifacts: Artifacts to update with coverage report path.
168 baseline_percent: Baseline coverage for "no decrease" mode.
169 env: Environment variables for command execution.
170 command_runner: Command runner for executing coverage command.
171 yaml_coverage_config: Coverage configuration from mala.yaml, or None
172 to use spec-based coverage checking (legacy mode).
174 Returns:
175 CoverageResult if coverage is enabled, None otherwise.
176 """
177 if not spec.coverage.enabled:
178 return None
180 # If yaml_coverage_config is provided, use config-driven coverage checking
181 if yaml_coverage_config is not None:
182 coverage_command_result = self._run_coverage_command_if_configured(
183 yaml_coverage_config, cwd, env, command_runner
184 )
185 if coverage_command_result is not None:
186 return coverage_command_result
187 coverage_result = check_coverage_from_config(yaml_coverage_config, cwd)
188 if coverage_result is not None and coverage_result.report_path:
189 artifacts.coverage_report = coverage_result.report_path
190 return coverage_result
192 # Legacy mode: use spec.coverage
193 coverage_result = self._check_coverage(
194 spec.coverage, cwd, log_dir, baseline_percent
195 )
196 if coverage_result.report_path:
197 artifacts.coverage_report = coverage_result.report_path
199 return coverage_result
201 def _run_coverage_command_if_configured(
202 self,
203 coverage_config: YamlCoverageConfig,
204 cwd: Path,
205 env: Mapping[str, str],
206 command_runner: CommandRunnerPort,
207 ) -> CoverageResult | None:
208 """Run coverage command if configured.
210 Args:
211 coverage_config: Coverage configuration from mala.yaml.
212 cwd: Working directory for the command.
213 env: Environment variables for the command.
214 command_runner: Command runner for executing the command.
216 Returns:
217 CoverageResult on failure, None if command not configured or succeeded.
218 """
219 if coverage_config.command is None:
220 return None
222 timeout_seconds = coverage_config.timeout or 120
223 result = command_runner.run(
224 coverage_config.command,
225 env=env,
226 shell=True,
227 cwd=cwd,
228 timeout=timeout_seconds,
229 )
231 if result.ok:
232 return None
234 details: list[str] = []
235 if result.timed_out:
236 details.append("coverage command timed out")
237 else:
238 details.append(f"coverage command exited {result.returncode}")
240 tail = result.stderr_tail() or result.stdout_tail()
241 if tail:
242 details.append(tail)
244 report_path = Path(coverage_config.file)
245 if not report_path.is_absolute():
246 report_path = cwd / report_path
248 return CoverageResult(
249 percent=None,
250 passed=False,
251 status=CoverageStatus.ERROR,
252 report_path=report_path,
253 failure_reason="Coverage command failed: " + "; ".join(details),
254 )
256 def _check_coverage(
257 self,
258 config: CoverageConfig,
259 cwd: Path,
260 log_dir: Path,
261 baseline_percent: float | None = None,
262 ) -> CoverageResult:
263 """Check coverage against threshold.
265 Args:
266 config: Coverage configuration.
267 cwd: Working directory where coverage.xml should be.
268 log_dir: Directory for logs.
269 baseline_percent: Baseline coverage percentage for "no decrease" mode.
270 Used when config.min_percent is None.
272 Returns:
273 CoverageResult with pass/fail status.
274 """
275 # Look for coverage.xml in cwd
276 # Resolve relative paths against cwd to ensure correct file lookup
277 if config.report_path is not None:
278 report_path = config.report_path
279 if not report_path.is_absolute():
280 report_path = cwd / report_path
281 else:
282 report_path = cwd / "coverage.xml"
284 if not report_path.exists():
285 # Coverage report not found - this is only an error if coverage was expected
286 return CoverageResult(
287 percent=None,
288 passed=False,
289 status=CoverageStatus.ERROR,
290 report_path=report_path,
291 failure_reason=f"Coverage report not found: {report_path}",
292 )
294 # Determine threshold: use config.min_percent if set, else baseline_percent
295 threshold = (
296 config.min_percent if config.min_percent is not None else baseline_percent
297 )
299 return parse_and_check_coverage(report_path, threshold)
301 def _run_e2e_if_enabled(
302 self,
303 spec: ValidationSpec,
304 env: Mapping[str, str],
305 cwd: Path,
306 log_dir: Path,
307 artifacts: ValidationArtifacts,
308 env_config: EnvConfigPort,
309 command_runner: CommandRunnerPort,
310 ) -> E2EResult | None:
311 """Run E2E validation if enabled (only for run-level scope).
313 Args:
314 spec: Validation spec with E2E config.
315 env: Environment variables.
316 cwd: Working directory.
317 log_dir: Directory for logs.
318 artifacts: Artifacts to update with fixture path.
319 env_config: Environment configuration for paths.
320 command_runner: Command runner for executing commands.
322 Returns:
323 E2EResult if E2E is enabled and scope is run-level, None otherwise.
324 """
325 from .spec import ValidationScope
327 if not spec.e2e.enabled or spec.scope != ValidationScope.RUN_LEVEL:
328 return None
330 e2e_result = self._run_e2e(
331 spec.e2e, env, cwd, log_dir, env_config, command_runner
332 )
333 if e2e_result.fixture_path:
334 artifacts.e2e_fixture_path = e2e_result.fixture_path
336 return e2e_result
338 def _run_e2e(
339 self,
340 config: E2EConfig,
341 env: Mapping[str, str],
342 cwd: Path,
343 log_dir: Path,
344 env_config: EnvConfigPort,
345 command_runner: CommandRunnerPort,
346 ) -> E2EResult:
347 """Run E2E validation using the E2ERunner.
349 Args:
350 config: E2E configuration from spec.
351 env: Environment variables.
352 cwd: Working directory.
353 log_dir: Directory for logs.
354 env_config: Environment configuration for paths.
355 command_runner: Command runner for executing commands.
357 Returns:
358 E2EResult with pass/fail status.
359 """
360 runner_config = E2ERunnerConfig(
361 enabled=config.enabled,
362 skip_if_no_keys=True, # Don't fail if API key missing
363 keep_fixture=True, # Keep for debugging
364 timeout_seconds=1200.0, # 20 min for E2E mala run (default 10 min was too short)
365 )
366 runner = E2ERunner(env_config, command_runner, runner_config)
367 return runner.run(env=dict(env), cwd=cwd)
369 def _build_failure_result(
370 self,
371 steps: list[ValidationStepResult],
372 reason: str,
373 artifacts: ValidationArtifacts,
374 coverage_result: CoverageResult | None = None,
375 e2e_result: E2EResult | None = None,
376 ) -> ValidationResult:
377 """Build a failed ValidationResult.
379 Args:
380 steps: Command execution steps.
381 reason: Failure reason.
382 artifacts: Validation artifacts.
383 coverage_result: Optional coverage result.
384 e2e_result: Optional E2E result.
386 Returns:
387 ValidationResult with passed=False.
388 """
389 return ValidationResult(
390 passed=False,
391 steps=steps,
392 failure_reasons=[reason],
393 artifacts=artifacts,
394 coverage_result=coverage_result,
395 e2e_result=e2e_result,
396 )