Coverage for src / pipeline / gate_metadata.py: 27%
37 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-01-04 04:43 +0000
1"""Gate metadata extraction for MalaOrchestrator.
3This module contains the GateMetadata dataclass and helper functions for
4extracting quality gate results for run finalization.
5"""
7from __future__ import annotations
9from dataclasses import dataclass
10from typing import TYPE_CHECKING, cast
12from src.infra.io.log_output.run_metadata import (
13 QualityGateResult,
14 ValidationResult as MetaValidationResult,
15)
17if TYPE_CHECKING:
18 from pathlib import Path
20 from src.core.protocols import (
21 GateChecker,
22 GateResultProtocol,
23 ValidationSpecProtocol,
24 )
25 from src.domain.quality_gate import GateResult
26 from src.domain.validation.spec import ValidationSpec
29@dataclass
30class GateMetadata:
31 """Metadata extracted from quality gate results for finalization.
33 This dataclass holds the processed results from a quality gate check,
34 separating the extraction logic from the finalization flow.
35 """
37 quality_gate_result: QualityGateResult | None = None
38 validation_result: MetaValidationResult | None = None
41def build_gate_metadata(
42 gate_result: GateResult | GateResultProtocol | None,
43 passed: bool,
44) -> GateMetadata:
45 """Build GateMetadata from a stored gate result.
47 Extracts evidence from the stored gate result without re-running validation.
48 This is the primary path used when gate results are available from
49 _run_quality_gate_sync.
51 Args:
52 gate_result: The stored gate result (may be None if no gate ran).
53 passed: Whether the overall run passed (affects quality_gate_result.passed).
55 Returns:
56 GateMetadata with extracted quality gate and validation results.
57 """
58 if gate_result is None:
59 return GateMetadata()
61 evidence = gate_result.validation_evidence
62 commit_hash = gate_result.commit_hash
64 # Build evidence dict from stored evidence
65 evidence_dict: dict[str, bool] = {}
66 if evidence is not None:
67 evidence_dict = evidence.to_evidence_dict()
68 evidence_dict["commit_found"] = commit_hash is not None
70 quality_gate_result = QualityGateResult(
71 passed=passed if passed else gate_result.passed,
72 evidence=evidence_dict,
73 failure_reasons=[] if passed else list(gate_result.failure_reasons),
74 )
76 validation_result: MetaValidationResult | None = None
77 if evidence is not None:
78 # Use kind.value for human-readable command names
79 commands_run = [
80 kind.value for kind, ran in evidence.commands_ran.items() if ran
81 ]
82 # failed_commands is already filtered by QUALITY_GATE_IGNORED_KINDS
83 # at the source in parse_validation_evidence_with_spec
84 validation_result = MetaValidationResult(
85 passed=passed if passed else gate_result.passed,
86 commands_run=commands_run,
87 commands_failed=list(evidence.failed_commands),
88 )
90 return GateMetadata(
91 quality_gate_result=quality_gate_result,
92 validation_result=validation_result,
93 )
96def build_gate_metadata_from_logs(
97 log_path: Path,
98 result_summary: str,
99 result_success: bool,
100 quality_gate: GateChecker,
101 per_issue_spec: ValidationSpec | ValidationSpecProtocol | None,
102) -> GateMetadata:
103 """Build GateMetadata by parsing logs directly (fallback path).
105 This is a fallback path used when no stored gate result is available.
106 It parses the log file directly to extract validation evidence.
108 Args:
109 log_path: Path to the session log file.
110 result_summary: Summary from the issue result (for extracting failure reasons).
111 result_success: Whether the run succeeded (determines passed status).
112 quality_gate: The GateChecker instance for parsing.
113 per_issue_spec: ValidationSpec for parsing evidence (if None, returns empty).
115 Returns:
116 GateMetadata with extracted results, or empty if spec is None.
117 """
118 if per_issue_spec is None:
119 return GateMetadata()
121 evidence = quality_gate.parse_validation_evidence_with_spec(
122 log_path, cast("ValidationSpecProtocol", per_issue_spec)
123 )
125 # Extract failure reasons from result summary
126 failure_reasons: list[str] = []
127 if "Quality gate failed:" in result_summary:
128 reasons_part = result_summary.replace("Quality gate failed: ", "")
129 failure_reasons = [r.strip() for r in reasons_part.split(";")]
131 # Build spec-driven evidence dict
132 evidence_dict = evidence.to_evidence_dict()
134 # Check commit exists (we don't have stored result, so check now)
135 # Note: We don't call check_commit_exists here because it's expensive
136 # and this is a fallback path - just mark as unknown
137 evidence_dict["commit_found"] = False
139 quality_gate_result = QualityGateResult(
140 passed=result_success,
141 evidence=evidence_dict,
142 failure_reasons=failure_reasons,
143 )
145 # Build validation result from evidence (matches build_gate_metadata behavior)
146 commands_run = [kind.value for kind, ran in evidence.commands_ran.items() if ran]
147 # failed_commands is already filtered by QUALITY_GATE_IGNORED_KINDS
148 # at the source in parse_validation_evidence_with_spec
149 validation_result = MetaValidationResult(
150 passed=result_success,
151 commands_run=commands_run,
152 commands_failed=list(evidence.failed_commands),
153 )
155 return GateMetadata(
156 quality_gate_result=quality_gate_result,
157 validation_result=validation_result,
158 )