feat(#147): Implement core quality gates (TDD - GREEN phase)
Implement four quality gates enforcing non-negotiable quality standards: 1. BuildGate: Runs mypy type checking - Detects compilation/type errors - Uses strict mode from pyproject.toml - Returns GateResult with pass/fail status 2. LintGate: Runs ruff linting - Treats warnings as failures (non-negotiable) - Checks code style and quality - Enforces rules from pyproject.toml 3. TestGate: Runs pytest tests - Requires 100% test pass rate (non-negotiable) - Runs without coverage (separate gate) - Detects test failures and missing tests 4. CoverageGate: Measures test coverage - Enforces 85% minimum coverage (non-negotiable) - Extracts coverage from JSON and output - Handles edge cases gracefully All gates implement QualityGate protocol with check() method. All gates return GateResult with passed/message/details. All implementations achieve 100% test coverage. Files created: - src/gates/quality_gate.py: Protocol and result model - src/gates/build_gate.py: Type checking enforcement - src/gates/lint_gate.py: Linting enforcement - src/gates/test_gate.py: Test execution enforcement - src/gates/coverage_gate.py: Coverage enforcement - src/gates/__init__.py: Module exports Related to #147 Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
16
apps/coordinator/src/gates/__init__.py
Normal file
16
apps/coordinator/src/gates/__init__.py
Normal file
@@ -0,0 +1,16 @@
|
||||
"""Quality gates for code quality enforcement."""
|
||||
|
||||
from src.gates.build_gate import BuildGate
|
||||
from src.gates.coverage_gate import CoverageGate
|
||||
from src.gates.lint_gate import LintGate
|
||||
from src.gates.quality_gate import GateResult, QualityGate
|
||||
from src.gates.test_gate import TestGate
|
||||
|
||||
__all__ = [
|
||||
"QualityGate",
|
||||
"GateResult",
|
||||
"BuildGate",
|
||||
"LintGate",
|
||||
"TestGate",
|
||||
"CoverageGate",
|
||||
]
|
||||
69
apps/coordinator/src/gates/build_gate.py
Normal file
69
apps/coordinator/src/gates/build_gate.py
Normal file
@@ -0,0 +1,69 @@
|
||||
"""BuildGate - Enforces type checking via mypy."""
|
||||
|
||||
import subprocess
|
||||
|
||||
from src.gates.quality_gate import GateResult
|
||||
|
||||
|
||||
class BuildGate:
|
||||
"""Quality gate that runs mypy type checking.
|
||||
|
||||
Executes mypy on the src/ directory and fails if any type errors are found.
|
||||
Uses strict mode configuration from pyproject.toml.
|
||||
"""
|
||||
|
||||
def check(self) -> GateResult:
|
||||
"""Run mypy type checker on source code.
|
||||
|
||||
Returns:
|
||||
GateResult: Result indicating if type checking passed
|
||||
"""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["mypy", "src/"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False, # Don't raise on non-zero exit
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
return GateResult(
|
||||
passed=True,
|
||||
message="Build gate passed: No type errors found",
|
||||
details={
|
||||
"return_code": result.returncode,
|
||||
"stdout": result.stdout,
|
||||
"stderr": result.stderr,
|
||||
},
|
||||
)
|
||||
else:
|
||||
return GateResult(
|
||||
passed=False,
|
||||
message="Build gate failed: Type errors detected",
|
||||
details={
|
||||
"return_code": result.returncode,
|
||||
"stdout": result.stdout,
|
||||
"stderr": result.stderr,
|
||||
},
|
||||
)
|
||||
|
||||
except FileNotFoundError as e:
|
||||
return GateResult(
|
||||
passed=False,
|
||||
message="Build gate failed: mypy not found or not installed",
|
||||
details={"error": str(e)},
|
||||
)
|
||||
|
||||
except subprocess.CalledProcessError as e:
|
||||
return GateResult(
|
||||
passed=False,
|
||||
message="Build gate failed: Error running mypy",
|
||||
details={"error": str(e), "return_code": e.returncode},
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return GateResult(
|
||||
passed=False,
|
||||
message=f"Build gate failed: Unexpected error: {e}",
|
||||
details={"error": str(e)},
|
||||
)
|
||||
149
apps/coordinator/src/gates/coverage_gate.py
Normal file
149
apps/coordinator/src/gates/coverage_gate.py
Normal file
@@ -0,0 +1,149 @@
|
||||
"""CoverageGate - Enforces 85% minimum test coverage via pytest-cov."""
|
||||
|
||||
import json
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
from src.gates.quality_gate import GateResult
|
||||
|
||||
|
||||
class CoverageGate:
|
||||
"""Quality gate that runs pytest with coverage measurement.
|
||||
|
||||
Executes pytest with coverage and enforces 85% minimum coverage (non-negotiable).
|
||||
"""
|
||||
|
||||
MINIMUM_COVERAGE = 85.0
|
||||
|
||||
def check(self) -> GateResult:
|
||||
"""Run pytest with coverage measurement.
|
||||
|
||||
Returns:
|
||||
GateResult: Result indicating if coverage meets 85% minimum
|
||||
"""
|
||||
try:
|
||||
# Run pytest with coverage
|
||||
result = subprocess.run(
|
||||
[
|
||||
"python",
|
||||
"-m",
|
||||
"pytest",
|
||||
"--cov=src",
|
||||
"--cov-report=json",
|
||||
"--cov-report=term-missing",
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False, # Don't raise on non-zero exit
|
||||
)
|
||||
|
||||
# Try to read coverage data from coverage.json
|
||||
coverage_percent = self._extract_coverage_from_json()
|
||||
if coverage_percent is None:
|
||||
# Fallback to parsing stdout
|
||||
coverage_percent = self._extract_coverage_from_output(result.stdout)
|
||||
|
||||
if coverage_percent is None:
|
||||
return GateResult(
|
||||
passed=False,
|
||||
message="Coverage gate failed: No coverage data found",
|
||||
details={
|
||||
"return_code": result.returncode,
|
||||
"stdout": result.stdout,
|
||||
"stderr": result.stderr,
|
||||
"error": "Could not extract coverage percentage",
|
||||
},
|
||||
)
|
||||
|
||||
# Check if coverage meets minimum threshold
|
||||
if coverage_percent >= self.MINIMUM_COVERAGE:
|
||||
return GateResult(
|
||||
passed=True,
|
||||
message=(
|
||||
f"Coverage gate passed: {coverage_percent:.1f}% coverage "
|
||||
f"(minimum: {self.MINIMUM_COVERAGE}%)"
|
||||
),
|
||||
details={
|
||||
"return_code": result.returncode,
|
||||
"coverage_percent": coverage_percent,
|
||||
"minimum_coverage": self.MINIMUM_COVERAGE,
|
||||
"stdout": result.stdout,
|
||||
"stderr": result.stderr,
|
||||
},
|
||||
)
|
||||
else:
|
||||
return GateResult(
|
||||
passed=False,
|
||||
message=(
|
||||
f"Coverage gate failed: {coverage_percent:.1f}% coverage "
|
||||
f"below minimum {self.MINIMUM_COVERAGE}%"
|
||||
),
|
||||
details={
|
||||
"return_code": result.returncode,
|
||||
"coverage_percent": coverage_percent,
|
||||
"minimum_coverage": self.MINIMUM_COVERAGE,
|
||||
"stdout": result.stdout,
|
||||
"stderr": result.stderr,
|
||||
},
|
||||
)
|
||||
|
||||
except FileNotFoundError as e:
|
||||
return GateResult(
|
||||
passed=False,
|
||||
message="Coverage gate failed: pytest not found or not installed",
|
||||
details={"error": str(e)},
|
||||
)
|
||||
|
||||
except subprocess.CalledProcessError as e:
|
||||
return GateResult(
|
||||
passed=False,
|
||||
message="Coverage gate failed: Error running pytest",
|
||||
details={"error": str(e), "return_code": e.returncode},
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return GateResult(
|
||||
passed=False,
|
||||
message=f"Coverage gate failed: Unexpected error: {e}",
|
||||
details={"error": str(e)},
|
||||
)
|
||||
|
||||
def _extract_coverage_from_json(self) -> float | None:
|
||||
"""Extract coverage percentage from coverage.json file.
|
||||
|
||||
Returns:
|
||||
float | None: Coverage percentage or None if file not found
|
||||
"""
|
||||
try:
|
||||
coverage_file = Path("coverage.json")
|
||||
if coverage_file.exists():
|
||||
with open(coverage_file) as f:
|
||||
data = json.load(f)
|
||||
percent = data.get("totals", {}).get("percent_covered")
|
||||
if percent is not None and isinstance(percent, (int, float)):
|
||||
return float(percent)
|
||||
except (FileNotFoundError, json.JSONDecodeError, KeyError):
|
||||
pass
|
||||
return None
|
||||
|
||||
def _extract_coverage_from_output(self, output: str) -> float | None:
|
||||
"""Extract coverage percentage from pytest output.
|
||||
|
||||
Args:
|
||||
output: stdout from pytest run
|
||||
|
||||
Returns:
|
||||
float | None: Coverage percentage or None if not found
|
||||
"""
|
||||
# Look for "TOTAL" line with coverage percentage
|
||||
# Example: "TOTAL 150 15 90%"
|
||||
for line in output.split("\n"):
|
||||
if "TOTAL" in line and "%" in line:
|
||||
parts = line.split()
|
||||
for part in parts:
|
||||
if "%" in part:
|
||||
try:
|
||||
return float(part.rstrip("%"))
|
||||
except ValueError:
|
||||
continue
|
||||
return None
|
||||
69
apps/coordinator/src/gates/lint_gate.py
Normal file
69
apps/coordinator/src/gates/lint_gate.py
Normal file
@@ -0,0 +1,69 @@
|
||||
"""LintGate - Enforces code style and quality via ruff."""
|
||||
|
||||
import subprocess
|
||||
|
||||
from src.gates.quality_gate import GateResult
|
||||
|
||||
|
||||
class LintGate:
|
||||
"""Quality gate that runs ruff linting.
|
||||
|
||||
Executes ruff check on the src/ directory and fails if any linting errors
|
||||
or warnings are found. Treats all warnings as failures (non-negotiable).
|
||||
"""
|
||||
|
||||
def check(self) -> GateResult:
|
||||
"""Run ruff linter on source code.
|
||||
|
||||
Returns:
|
||||
GateResult: Result indicating if linting passed
|
||||
"""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["ruff", "check", "src/"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False, # Don't raise on non-zero exit
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
return GateResult(
|
||||
passed=True,
|
||||
message="Lint gate passed: No linting issues found",
|
||||
details={
|
||||
"return_code": result.returncode,
|
||||
"stdout": result.stdout,
|
||||
"stderr": result.stderr,
|
||||
},
|
||||
)
|
||||
else:
|
||||
return GateResult(
|
||||
passed=False,
|
||||
message="Lint gate failed: Linting issues detected",
|
||||
details={
|
||||
"return_code": result.returncode,
|
||||
"stdout": result.stdout,
|
||||
"stderr": result.stderr,
|
||||
},
|
||||
)
|
||||
|
||||
except FileNotFoundError as e:
|
||||
return GateResult(
|
||||
passed=False,
|
||||
message="Lint gate failed: ruff not found or not installed",
|
||||
details={"error": str(e)},
|
||||
)
|
||||
|
||||
except subprocess.CalledProcessError as e:
|
||||
return GateResult(
|
||||
passed=False,
|
||||
message="Lint gate failed: Error running ruff",
|
||||
details={"error": str(e), "return_code": e.returncode},
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return GateResult(
|
||||
passed=False,
|
||||
message=f"Lint gate failed: Unexpected error: {e}",
|
||||
details={"error": str(e)},
|
||||
)
|
||||
36
apps/coordinator/src/gates/quality_gate.py
Normal file
36
apps/coordinator/src/gates/quality_gate.py
Normal file
@@ -0,0 +1,36 @@
|
||||
"""Quality gate interface and result model."""
|
||||
|
||||
from typing import Any, Protocol
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class GateResult(BaseModel):
|
||||
"""Result of a quality gate check.
|
||||
|
||||
Attributes:
|
||||
passed: Whether the gate check passed
|
||||
message: Human-readable message describing the result
|
||||
details: Optional additional details about the result (e.g., errors, warnings)
|
||||
"""
|
||||
|
||||
passed: bool = Field(..., description="Whether the gate check passed")
|
||||
message: str = Field(..., description="Human-readable result message")
|
||||
details: dict[str, Any] = Field(
|
||||
default_factory=dict, description="Additional details about the result"
|
||||
)
|
||||
|
||||
|
||||
class QualityGate(Protocol):
|
||||
"""Protocol for quality gate implementations.
|
||||
|
||||
All quality gates must implement this protocol to ensure consistent interface.
|
||||
"""
|
||||
|
||||
def check(self) -> GateResult:
|
||||
"""Execute the quality gate check.
|
||||
|
||||
Returns:
|
||||
GateResult: Result of the gate check with pass/fail status and details
|
||||
"""
|
||||
...
|
||||
69
apps/coordinator/src/gates/test_gate.py
Normal file
69
apps/coordinator/src/gates/test_gate.py
Normal file
@@ -0,0 +1,69 @@
|
||||
"""TestGate - Enforces 100% test pass rate via pytest."""
|
||||
|
||||
import subprocess
|
||||
|
||||
from src.gates.quality_gate import GateResult
|
||||
|
||||
|
||||
class TestGate:
|
||||
"""Quality gate that runs pytest tests.
|
||||
|
||||
Executes pytest and requires 100% pass rate (non-negotiable).
|
||||
Runs tests without coverage - coverage is handled by CoverageGate separately.
|
||||
"""
|
||||
|
||||
def check(self) -> GateResult:
|
||||
"""Run pytest test suite.
|
||||
|
||||
Returns:
|
||||
GateResult: Result indicating if all tests passed
|
||||
"""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["python", "-m", "pytest", "--no-cov", "-v"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False, # Don't raise on non-zero exit
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
return GateResult(
|
||||
passed=True,
|
||||
message="Test gate passed: All tests passed (100% pass rate)",
|
||||
details={
|
||||
"return_code": result.returncode,
|
||||
"stdout": result.stdout,
|
||||
"stderr": result.stderr,
|
||||
},
|
||||
)
|
||||
else:
|
||||
return GateResult(
|
||||
passed=False,
|
||||
message="Test gate failed: Test failures detected (requires 100% pass rate)",
|
||||
details={
|
||||
"return_code": result.returncode,
|
||||
"stdout": result.stdout,
|
||||
"stderr": result.stderr,
|
||||
},
|
||||
)
|
||||
|
||||
except FileNotFoundError as e:
|
||||
return GateResult(
|
||||
passed=False,
|
||||
message="Test gate failed: pytest not found or not installed",
|
||||
details={"error": str(e)},
|
||||
)
|
||||
|
||||
except subprocess.CalledProcessError as e:
|
||||
return GateResult(
|
||||
passed=False,
|
||||
message="Test gate failed: Error running pytest",
|
||||
details={"error": str(e), "return_code": e.returncode},
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return GateResult(
|
||||
passed=False,
|
||||
message=f"Test gate failed: Unexpected error: {e}",
|
||||
details={"error": str(e)},
|
||||
)
|
||||
Reference in New Issue
Block a user