diff --git a/apps/coordinator/src/gates/__init__.py b/apps/coordinator/src/gates/__init__.py new file mode 100644 index 0000000..3484d8f --- /dev/null +++ b/apps/coordinator/src/gates/__init__.py @@ -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", +] diff --git a/apps/coordinator/src/gates/build_gate.py b/apps/coordinator/src/gates/build_gate.py new file mode 100644 index 0000000..4cbb650 --- /dev/null +++ b/apps/coordinator/src/gates/build_gate.py @@ -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)}, + ) diff --git a/apps/coordinator/src/gates/coverage_gate.py b/apps/coordinator/src/gates/coverage_gate.py new file mode 100644 index 0000000..d658ad2 --- /dev/null +++ b/apps/coordinator/src/gates/coverage_gate.py @@ -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 diff --git a/apps/coordinator/src/gates/lint_gate.py b/apps/coordinator/src/gates/lint_gate.py new file mode 100644 index 0000000..7d3524d --- /dev/null +++ b/apps/coordinator/src/gates/lint_gate.py @@ -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)}, + ) diff --git a/apps/coordinator/src/gates/quality_gate.py b/apps/coordinator/src/gates/quality_gate.py new file mode 100644 index 0000000..cfd652b --- /dev/null +++ b/apps/coordinator/src/gates/quality_gate.py @@ -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 + """ + ... diff --git a/apps/coordinator/src/gates/test_gate.py b/apps/coordinator/src/gates/test_gate.py new file mode 100644 index 0000000..bc29cd5 --- /dev/null +++ b/apps/coordinator/src/gates/test_gate.py @@ -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)}, + )