diff --git a/apps/coordinator/src/forced_continuation.py b/apps/coordinator/src/forced_continuation.py new file mode 100644 index 0000000..5fdeef8 --- /dev/null +++ b/apps/coordinator/src/forced_continuation.py @@ -0,0 +1,144 @@ +"""Forced Continuation service for generating non-negotiable agent instructions.""" + +from src.quality_orchestrator import VerificationResult + + +class ForcedContinuationService: + """Generates forced continuation prompts for quality gate failures. + + This service creates non-negotiable, actionable prompts that instruct + agents to fix quality gate failures. The prompts are designed to: + - Be clear and directive (not suggestions) + - Include specific failure details + - Provide actionable remediation steps + - Block completion until all gates pass + """ + + def generate_prompt(self, verification: VerificationResult) -> str: + """Generate a forced continuation prompt for gate failures. + + Args: + verification: VerificationResult containing gate failure details + + Returns: + str: Non-negotiable prompt instructing agent to fix failures + + Raises: + ValueError: If verification.all_passed is True (no failures to fix) + """ + if verification.all_passed: + raise ValueError( + "Cannot generate continuation prompt when all gates pass. " + "This method should only be called when verification fails." + ) + + # Collect failed gates + failed_gates = { + name: result + for name, result in verification.gate_results.items() + if not result.passed + } + + # Build the prompt + prompt_parts = [ + "QUALITY GATES FAILED - COMPLETION BLOCKED", + "", + "The following quality gates have failed and MUST be fixed before completion:", + "", + ] + + # Add details for each failed gate + for gate_name, result in failed_gates.items(): + prompt_parts.append(f"❌ {gate_name.upper()} GATE FAILED") + prompt_parts.append(f" Message: {result.message}") + + # Add specific details if available + if result.details: + if "stderr" in result.details and result.details["stderr"]: + prompt_parts.append(" Details:") + # Include first few lines of stderr + stderr_lines = result.details["stderr"].split("\n")[:5] + for line in stderr_lines: + if line.strip(): + prompt_parts.append(f" {line}") + + # Add coverage-specific details + if "coverage_percent" in result.details: + coverage = result.details["coverage_percent"] + minimum = result.details.get("minimum_coverage", 85.0) + gap = minimum - coverage + prompt_parts.append(f" Current coverage: {coverage:.1f}%") + prompt_parts.append(f" Required coverage: {minimum:.1f}%") + prompt_parts.append(f" Coverage gap: {gap:.1f}%") + + prompt_parts.append("") + + # Add remediation instructions + prompt_parts.extend( + [ + "REQUIRED ACTIONS:", + "", + ] + ) + + # Add specific remediation steps based on which gates failed + if "build" in failed_gates: + prompt_parts.extend( + [ + "1. BUILD GATE - Fix all type errors:", + " - Run: mypy src/", + " - Fix all type errors reported", + " - Ensure all type annotations are correct", + "", + ] + ) + + if "lint" in failed_gates: + prompt_parts.extend( + [ + "2. LINT GATE - Fix all linting issues:", + " - Run: ruff check src/", + " - Fix all errors and warnings", + " - Ensure code follows style guidelines", + "", + ] + ) + + if "test" in failed_gates: + prompt_parts.extend( + [ + "3. TEST GATE - Fix all failing tests:", + " - Run: pytest -v", + " - Fix all test failures", + " - Ensure 100% test pass rate", + "", + ] + ) + + if "coverage" in failed_gates: + coverage_result = failed_gates["coverage"] + current = coverage_result.details.get("coverage_percent", 0.0) + minimum = coverage_result.details.get("minimum_coverage", 85.0) + + prompt_parts.extend( + [ + "4. COVERAGE GATE - Increase test coverage:", + " - Run: pytest --cov=src --cov-report=term-missing", + f" - Current: {current:.1f}% | Required: {minimum:.1f}%", + " - Add tests for uncovered code paths", + " - Focus on files with low coverage", + "", + ] + ) + + # Add final directive + prompt_parts.extend( + [ + "You MUST fix all failing gates before claiming completion.", + "After fixing issues, run all quality gates again to verify.", + "", + "DO NOT claim completion until all gates pass.", + ] + ) + + return "\n".join(prompt_parts) diff --git a/apps/coordinator/src/quality_orchestrator.py b/apps/coordinator/src/quality_orchestrator.py new file mode 100644 index 0000000..551929a --- /dev/null +++ b/apps/coordinator/src/quality_orchestrator.py @@ -0,0 +1,164 @@ +"""Quality Orchestrator service for coordinating quality gate execution.""" + +import asyncio +from typing import Any + +from pydantic import BaseModel, Field + +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 +from src.gates.test_gate import TestGate + + +class VerificationResult(BaseModel): + """Result of quality gate verification. + + Attributes: + all_passed: Whether all quality gates passed + gate_results: Dictionary mapping gate names to their results + """ + + all_passed: bool = Field(..., description="Whether all quality gates passed") + gate_results: dict[str, GateResult] = Field( + ..., description="Results from each quality gate" + ) + + +class QualityOrchestrator: + """Orchestrates execution of all quality gates in parallel. + + The Quality Orchestrator is responsible for: + - Running all quality gates (build, lint, test, coverage) in parallel + - Aggregating gate results + - Determining overall pass/fail status + """ + + def __init__( + self, + build_gate: BuildGate | None = None, + lint_gate: LintGate | None = None, + test_gate: TestGate | None = None, + coverage_gate: CoverageGate | None = None, + ) -> None: + """Initialize the Quality Orchestrator. + + Args: + build_gate: Optional BuildGate instance (for testing/DI) + lint_gate: Optional LintGate instance (for testing/DI) + test_gate: Optional TestGate instance (for testing/DI) + coverage_gate: Optional CoverageGate instance (for testing/DI) + """ + # Use provided gates or create new instances + # This allows for dependency injection in tests + self.build_gate = build_gate + self.lint_gate = lint_gate + self.test_gate = test_gate + self.coverage_gate = coverage_gate + + async def verify_completion(self) -> VerificationResult: + """Verify that all quality gates pass. + + Runs all quality gates in parallel and aggregates the results. + + Returns: + VerificationResult: Aggregated results from all gates + + Note: + This method runs all gates in parallel for efficiency. + Even if one gate fails, all gates will complete execution. + """ + # Instantiate gates if not provided (lazy initialization) + # This allows tests to inject mocks, while production uses real gates + build_gate = self.build_gate if self.build_gate is not None else BuildGate() + lint_gate = self.lint_gate if self.lint_gate is not None else LintGate() + test_gate = self.test_gate if self.test_gate is not None else TestGate() + coverage_gate = self.coverage_gate if self.coverage_gate is not None else CoverageGate() + + # Run all gates in parallel using asyncio.gather + results = await asyncio.gather( + self._run_gate_async("build", build_gate), + self._run_gate_async("lint", lint_gate), + self._run_gate_async("test", test_gate), + self._run_gate_async("coverage", coverage_gate), + return_exceptions=True, # Capture exceptions instead of raising + ) + + # Build gate results dictionary + gate_results: dict[str, GateResult] = {} + gate_names = ["build", "lint", "test", "coverage"] + + for gate_name, result in zip(gate_names, results, strict=True): + if isinstance(result, Exception): + # Convert exception to failed GateResult + gate_results[gate_name] = GateResult( + passed=False, + message=f"{gate_name.capitalize()} gate failed: Unexpected error: {result}", + details={"error": str(result), "exception_type": type(result).__name__}, + ) + elif isinstance(result, GateResult): + gate_results[gate_name] = result + else: + # Unexpected type - treat as error + gate_results[gate_name] = GateResult( + passed=False, + message=f"{gate_name.capitalize()} gate failed: Unexpected result type", + details={"error": f"Expected GateResult, got {type(result).__name__}"}, + ) + + # Determine if all gates passed + all_passed = all(result.passed for result in gate_results.values()) + + return VerificationResult(all_passed=all_passed, gate_results=gate_results) + + async def _run_gate_async(self, gate_name: str, gate: Any) -> GateResult: + """Run a gate check asynchronously. + + Args: + gate_name: Name of the gate for error reporting + gate: Gate instance to execute + + Returns: + GateResult: Result from the gate check + + Note: + This method handles both synchronous gates (production) and async mocks (testing). + Production gates are run in a thread pool to avoid blocking the event loop. + Test mocks can be async functions or lambdas returning coroutines. + """ + import inspect + from typing import cast + from unittest.mock import Mock + + # Check if gate.check is an async function + if inspect.iscoroutinefunction(gate.check): + return cast(GateResult, await gate.check()) + + # Check if gate.check is a Mock/MagicMock (testing scenario) + mock_types = ("Mock", "MagicMock", "AsyncMock") + if isinstance(gate.check, Mock) or type(gate.check).__name__ in mock_types: + # It's a mock - call it and handle the result + result_or_coro = gate.check() + if asyncio.iscoroutine(result_or_coro): + return cast(GateResult, await result_or_coro) + return cast(GateResult, result_or_coro) + + # Check if gate.check is a lambda or other callable (could be test or production) + # For lambdas in tests that return coroutines, we need to call and await + # But we need to avoid calling real production gates outside of to_thread + # The distinguishing factor: real gates are methods on BuildGate/LintGate/etc classes + + # Check if it's a bound method on a real gate class + if inspect.ismethod(gate.check): + # Check if the class is one of our real gate classes + gate_class_name = gate.__class__.__name__ + if gate_class_name in ("BuildGate", "LintGate", "TestGate", "CoverageGate"): + # It's a real gate - run in thread pool + return cast(GateResult, await asyncio.to_thread(gate.check)) + + # For any other callable (lambdas, functions), try calling and see what it returns + result_or_coro = gate.check() + if asyncio.iscoroutine(result_or_coro): + return cast(GateResult, await result_or_coro) + return cast(GateResult, result_or_coro) diff --git a/apps/coordinator/tests/test_forced_continuation.py b/apps/coordinator/tests/test_forced_continuation.py new file mode 100644 index 0000000..e87f899 --- /dev/null +++ b/apps/coordinator/tests/test_forced_continuation.py @@ -0,0 +1,343 @@ +"""Tests for ForcedContinuationService.""" + +import pytest + +from src.forced_continuation import ForcedContinuationService +from src.gates.quality_gate import GateResult +from src.quality_orchestrator import VerificationResult + + +class TestForcedContinuationService: + """Test suite for ForcedContinuationService.""" + + @pytest.fixture + def service(self) -> ForcedContinuationService: + """Create a ForcedContinuationService instance for testing.""" + return ForcedContinuationService() + + def test_generate_prompt_single_build_failure( + self, service: ForcedContinuationService + ) -> None: + """Test prompt generation for single build gate failure.""" + verification = VerificationResult( + all_passed=False, + gate_results={ + "build": GateResult( + passed=False, + message="Build gate failed: Type errors detected", + details={ + "return_code": 1, + "stderr": "src/main.py:10: error: Incompatible return value type", + }, + ), + "lint": GateResult(passed=True, message="Lint passed", details={}), + "test": GateResult(passed=True, message="Test passed", details={}), + "coverage": GateResult( + passed=True, message="Coverage passed", details={} + ), + }, + ) + + prompt = service.generate_prompt(verification) + + # Assert prompt structure + assert isinstance(prompt, str) + assert len(prompt) > 0 + assert "build" in prompt.lower() or "type" in prompt.lower() + assert "failed" in prompt.lower() or "error" in prompt.lower() + # Should be non-negotiable and directive + assert ( + "must" in prompt.lower() + or "required" in prompt.lower() + or "fix" in prompt.lower() + ) + + def test_generate_prompt_single_lint_failure( + self, service: ForcedContinuationService + ) -> None: + """Test prompt generation for single lint gate failure.""" + verification = VerificationResult( + all_passed=False, + gate_results={ + "build": GateResult(passed=True, message="Build passed", details={}), + "lint": GateResult( + passed=False, + message="Lint gate failed: Linting issues detected", + details={ + "return_code": 1, + "stderr": "src/main.py:10: E501 line too long\nsrc/models.py:5: F401 unused import", + }, + ), + "test": GateResult(passed=True, message="Test passed", details={}), + "coverage": GateResult( + passed=True, message="Coverage passed", details={} + ), + }, + ) + + prompt = service.generate_prompt(verification) + + # Assert prompt structure + assert isinstance(prompt, str) + assert len(prompt) > 0 + assert "lint" in prompt.lower() + assert "failed" in prompt.lower() or "error" in prompt.lower() + assert ( + "must" in prompt.lower() + or "required" in prompt.lower() + or "fix" in prompt.lower() + ) + + def test_generate_prompt_single_test_failure( + self, service: ForcedContinuationService + ) -> None: + """Test prompt generation for single test gate failure.""" + verification = VerificationResult( + all_passed=False, + gate_results={ + "build": GateResult(passed=True, message="Build passed", details={}), + "lint": GateResult(passed=True, message="Lint passed", details={}), + "test": GateResult( + passed=False, + message="Test gate failed: Test failures detected", + details={ + "return_code": 1, + "stderr": "FAILED tests/test_main.py::test_function - AssertionError", + }, + ), + "coverage": GateResult( + passed=True, message="Coverage passed", details={} + ), + }, + ) + + prompt = service.generate_prompt(verification) + + # Assert prompt structure + assert isinstance(prompt, str) + assert len(prompt) > 0 + assert "test" in prompt.lower() + assert "failed" in prompt.lower() or "error" in prompt.lower() + assert ( + "must" in prompt.lower() + or "required" in prompt.lower() + or "fix" in prompt.lower() + ) + + def test_generate_prompt_single_coverage_failure( + self, service: ForcedContinuationService + ) -> None: + """Test prompt generation for single coverage gate failure.""" + verification = VerificationResult( + all_passed=False, + gate_results={ + "build": GateResult(passed=True, message="Build passed", details={}), + "lint": GateResult(passed=True, message="Lint passed", details={}), + "test": GateResult(passed=True, message="Test passed", details={}), + "coverage": GateResult( + passed=False, + message="Coverage gate failed: 75.0% coverage below minimum 85%", + details={ + "coverage_percent": 75.0, + "minimum_coverage": 85.0, + }, + ), + }, + ) + + prompt = service.generate_prompt(verification) + + # Assert prompt structure + assert isinstance(prompt, str) + assert len(prompt) > 0 + assert "coverage" in prompt.lower() + assert "75" in prompt or "85" in prompt # Should include actual/minimum coverage + assert ( + "must" in prompt.lower() + or "required" in prompt.lower() + or "fix" in prompt.lower() + ) + + def test_generate_prompt_multiple_failures( + self, service: ForcedContinuationService + ) -> None: + """Test prompt generation for multiple gate failures.""" + verification = VerificationResult( + all_passed=False, + gate_results={ + "build": GateResult( + passed=False, + message="Build gate failed: Type errors detected", + details={ + "return_code": 1, + "stderr": "src/main.py:10: error: Incompatible return value type", + }, + ), + "lint": GateResult( + passed=False, + message="Lint gate failed: Linting issues detected", + details={ + "return_code": 1, + "stderr": "src/main.py:10: E501 line too long", + }, + ), + "test": GateResult(passed=True, message="Test passed", details={}), + "coverage": GateResult( + passed=False, + message="Coverage gate failed: 75.0% coverage below minimum 85%", + details={ + "coverage_percent": 75.0, + "minimum_coverage": 85.0, + }, + ), + }, + ) + + prompt = service.generate_prompt(verification) + + # Assert prompt structure + assert isinstance(prompt, str) + assert len(prompt) > 0 + # Should mention multiple failures + assert "build" in prompt.lower() or "type" in prompt.lower() + assert "lint" in prompt.lower() + assert "coverage" in prompt.lower() + # Should be non-negotiable + assert ( + "must" in prompt.lower() + or "required" in prompt.lower() + or "fix" in prompt.lower() + ) + + def test_generate_prompt_all_failures( + self, service: ForcedContinuationService + ) -> None: + """Test prompt generation when all gates fail.""" + verification = VerificationResult( + all_passed=False, + gate_results={ + "build": GateResult( + passed=False, + message="Build gate failed", + details={}, + ), + "lint": GateResult( + passed=False, + message="Lint gate failed", + details={}, + ), + "test": GateResult( + passed=False, + message="Test gate failed", + details={}, + ), + "coverage": GateResult( + passed=False, + message="Coverage gate failed", + details={}, + ), + }, + ) + + prompt = service.generate_prompt(verification) + + # Assert prompt structure + assert isinstance(prompt, str) + assert len(prompt) > 0 + # Should mention all gates + assert "build" in prompt.lower() or "type" in prompt.lower() + assert "lint" in prompt.lower() + assert "test" in prompt.lower() + assert "coverage" in prompt.lower() + # Should be strongly worded + assert ( + "must" in prompt.lower() + or "required" in prompt.lower() + or "fix" in prompt.lower() + ) + + def test_generate_prompt_includes_actionable_details( + self, service: ForcedContinuationService + ) -> None: + """Test that generated prompt includes actionable details from gate results.""" + verification = VerificationResult( + all_passed=False, + gate_results={ + "build": GateResult( + passed=False, + message="Build gate failed: Type errors detected", + details={ + "return_code": 1, + "stderr": "src/main.py:10: error: Incompatible return value type\n" + "src/models.py:5: error: Missing type annotation", + }, + ), + "lint": GateResult(passed=True, message="Lint passed", details={}), + "test": GateResult(passed=True, message="Test passed", details={}), + "coverage": GateResult( + passed=True, message="Coverage passed", details={} + ), + }, + ) + + prompt = service.generate_prompt(verification) + + # Assert prompt includes specific error details + assert isinstance(prompt, str) + assert len(prompt) > 0 + # Should include file references or specific errors when available + assert ( + "main.py" in prompt + or "models.py" in prompt + or "error" in prompt.lower() + ) + + def test_generate_prompt_clear_instructions( + self, service: ForcedContinuationService + ) -> None: + """Test that generated prompt provides clear instructions.""" + verification = VerificationResult( + all_passed=False, + gate_results={ + "build": GateResult(passed=True, message="Build passed", details={}), + "lint": GateResult(passed=True, message="Lint passed", details={}), + "test": GateResult( + passed=False, + message="Test gate failed: Test failures detected", + details={ + "return_code": 1, + }, + ), + "coverage": GateResult( + passed=True, message="Coverage passed", details={} + ), + }, + ) + + prompt = service.generate_prompt(verification) + + # Assert prompt has clear instructions + assert isinstance(prompt, str) + assert len(prompt) > 50 # Should be substantial, not just a one-liner + # Should tell agent what to do, not just what failed + assert "fix" in prompt.lower() or "resolve" in prompt.lower() + + def test_generate_prompt_raises_on_all_passed( + self, service: ForcedContinuationService + ) -> None: + """Test that generate_prompt raises error when all gates pass.""" + verification = VerificationResult( + all_passed=True, + gate_results={ + "build": GateResult(passed=True, message="Build passed", details={}), + "lint": GateResult(passed=True, message="Lint passed", details={}), + "test": GateResult(passed=True, message="Test passed", details={}), + "coverage": GateResult( + passed=True, message="Coverage passed", details={} + ), + }, + ) + + # Should raise ValueError or similar when trying to generate prompt for passing verification + with pytest.raises(ValueError, match="all.*pass"): + service.generate_prompt(verification) diff --git a/apps/coordinator/tests/test_quality_orchestrator.py b/apps/coordinator/tests/test_quality_orchestrator.py new file mode 100644 index 0000000..8cdbb69 --- /dev/null +++ b/apps/coordinator/tests/test_quality_orchestrator.py @@ -0,0 +1,328 @@ +"""Tests for QualityOrchestrator service.""" + +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from src.gates.quality_gate import GateResult +from src.quality_orchestrator import QualityOrchestrator, VerificationResult + + +class TestQualityOrchestrator: + """Test suite for QualityOrchestrator.""" + + @pytest.fixture + def orchestrator(self) -> QualityOrchestrator: + """Create a QualityOrchestrator instance for testing.""" + return QualityOrchestrator() + + @pytest.mark.asyncio + async def test_verify_completion_all_gates_pass( + self, orchestrator: QualityOrchestrator + ) -> None: + """Test that verify_completion passes when all gates pass.""" + # Mock all gates to return passing results + mock_build_result = GateResult( + passed=True, + message="Build gate passed: No type errors found", + details={"return_code": 0}, + ) + mock_lint_result = GateResult( + passed=True, + message="Lint gate passed: No linting issues found", + details={"return_code": 0}, + ) + mock_test_result = GateResult( + passed=True, + message="Test gate passed: All tests passed (100% pass rate)", + details={"return_code": 0}, + ) + mock_coverage_result = GateResult( + passed=True, + message="Coverage gate passed: 90.0% coverage (minimum: 85%)", + details={"coverage_percent": 90.0, "minimum_coverage": 85.0}, + ) + + with ( + patch("src.quality_orchestrator.BuildGate") as mock_build_gate, + patch("src.quality_orchestrator.LintGate") as mock_lint_gate, + patch("src.quality_orchestrator.TestGate") as mock_test_gate, + patch("src.quality_orchestrator.CoverageGate") as mock_coverage_gate, + ): + # Configure mocks + mock_build_gate.return_value.check.return_value = mock_build_result + mock_lint_gate.return_value.check.return_value = mock_lint_result + mock_test_gate.return_value.check.return_value = mock_test_result + mock_coverage_gate.return_value.check.return_value = mock_coverage_result + + # Verify completion + result = await orchestrator.verify_completion() + + # Assert result + assert isinstance(result, VerificationResult) + assert result.all_passed is True + assert len(result.gate_results) == 4 + assert "build" in result.gate_results + assert "lint" in result.gate_results + assert "test" in result.gate_results + assert "coverage" in result.gate_results + assert result.gate_results["build"].passed is True + assert result.gate_results["lint"].passed is True + assert result.gate_results["test"].passed is True + assert result.gate_results["coverage"].passed is True + + @pytest.mark.asyncio + async def test_verify_completion_one_gate_fails( + self, orchestrator: QualityOrchestrator + ) -> None: + """Test that verify_completion fails when one gate fails.""" + # Mock gates with one failure + mock_build_result = GateResult( + passed=True, + message="Build gate passed", + details={}, + ) + mock_lint_result = GateResult( + passed=False, + message="Lint gate failed: Linting issues detected", + details={ + "return_code": 1, + "stderr": "src/main.py:10: E501 line too long", + }, + ) + mock_test_result = GateResult( + passed=True, + message="Test gate passed", + details={}, + ) + mock_coverage_result = GateResult( + passed=True, + message="Coverage gate passed", + details={}, + ) + + with ( + patch("src.quality_orchestrator.BuildGate") as mock_build_gate, + patch("src.quality_orchestrator.LintGate") as mock_lint_gate, + patch("src.quality_orchestrator.TestGate") as mock_test_gate, + patch("src.quality_orchestrator.CoverageGate") as mock_coverage_gate, + ): + # Configure mocks + mock_build_gate.return_value.check.return_value = mock_build_result + mock_lint_gate.return_value.check.return_value = mock_lint_result + mock_test_gate.return_value.check.return_value = mock_test_result + mock_coverage_gate.return_value.check.return_value = mock_coverage_result + + # Verify completion + result = await orchestrator.verify_completion() + + # Assert result + assert isinstance(result, VerificationResult) + assert result.all_passed is False + assert result.gate_results["lint"].passed is False + assert result.gate_results["build"].passed is True + assert result.gate_results["test"].passed is True + assert result.gate_results["coverage"].passed is True + + @pytest.mark.asyncio + async def test_verify_completion_multiple_gates_fail( + self, orchestrator: QualityOrchestrator + ) -> None: + """Test that verify_completion fails when multiple gates fail.""" + # Mock gates with multiple failures + mock_build_result = GateResult( + passed=False, + message="Build gate failed: Type errors detected", + details={ + "return_code": 1, + "stderr": "src/main.py:10: error: Incompatible return value type", + }, + ) + mock_lint_result = GateResult( + passed=False, + message="Lint gate failed: Linting issues detected", + details={ + "return_code": 1, + "stderr": "src/main.py:10: E501 line too long", + }, + ) + mock_test_result = GateResult( + passed=True, + message="Test gate passed", + details={}, + ) + mock_coverage_result = GateResult( + passed=False, + message="Coverage gate failed: 75.0% coverage below minimum 85%", + details={"coverage_percent": 75.0, "minimum_coverage": 85.0}, + ) + + with ( + patch("src.quality_orchestrator.BuildGate") as mock_build_gate, + patch("src.quality_orchestrator.LintGate") as mock_lint_gate, + patch("src.quality_orchestrator.TestGate") as mock_test_gate, + patch("src.quality_orchestrator.CoverageGate") as mock_coverage_gate, + ): + # Configure mocks + mock_build_gate.return_value.check.return_value = mock_build_result + mock_lint_gate.return_value.check.return_value = mock_lint_result + mock_test_gate.return_value.check.return_value = mock_test_result + mock_coverage_gate.return_value.check.return_value = mock_coverage_result + + # Verify completion + result = await orchestrator.verify_completion() + + # Assert result + assert isinstance(result, VerificationResult) + assert result.all_passed is False + assert result.gate_results["build"].passed is False + assert result.gate_results["lint"].passed is False + assert result.gate_results["test"].passed is True + assert result.gate_results["coverage"].passed is False + + @pytest.mark.asyncio + async def test_verify_completion_runs_gates_in_parallel( + self, orchestrator: QualityOrchestrator + ) -> None: + """Test that verify_completion runs all gates in parallel.""" + # Create mock gates with delay to test parallelism + mock_build_result = GateResult(passed=True, message="Build passed", details={}) + mock_lint_result = GateResult(passed=True, message="Lint passed", details={}) + mock_test_result = GateResult(passed=True, message="Test passed", details={}) + mock_coverage_result = GateResult( + passed=True, message="Coverage passed", details={} + ) + + # Track call order + call_order = [] + + async def mock_gate_check(gate_name: str, result: GateResult) -> GateResult: + """Mock gate check with tracking.""" + call_order.append(f"{gate_name}_start") + await asyncio.sleep(0.01) # Simulate work + call_order.append(f"{gate_name}_end") + return result + + with ( + patch("src.quality_orchestrator.BuildGate") as mock_build_gate, + patch("src.quality_orchestrator.LintGate") as mock_lint_gate, + patch("src.quality_orchestrator.TestGate") as mock_test_gate, + patch("src.quality_orchestrator.CoverageGate") as mock_coverage_gate, + ): + # Configure mocks to use async tracking + mock_build_gate.return_value.check = lambda: mock_gate_check( + "build", mock_build_result + ) + mock_lint_gate.return_value.check = lambda: mock_gate_check( + "lint", mock_lint_result + ) + mock_test_gate.return_value.check = lambda: mock_gate_check( + "test", mock_test_result + ) + mock_coverage_gate.return_value.check = lambda: mock_gate_check( + "coverage", mock_coverage_result + ) + + # Verify completion + result = await orchestrator.verify_completion() + + # Assert all gates completed + assert result.all_passed is True + assert len(result.gate_results) == 4 + + # Assert gates were started before any ended (parallel execution) + # In parallel execution, all "_start" events should appear before all "_end" events + start_events = [e for e in call_order if e.endswith("_start")] + end_events = [e for e in call_order if e.endswith("_end")] + + # All gates should have started + assert len(start_events) == 4 + # All gates should have ended + assert len(end_events) == 4 + + @pytest.mark.asyncio + async def test_verify_completion_handles_gate_exception( + self, orchestrator: QualityOrchestrator + ) -> None: + """Test that verify_completion handles exceptions from gates gracefully.""" + # Mock gates with one raising an exception + mock_build_result = GateResult(passed=True, message="Build passed", details={}) + + with ( + patch("src.quality_orchestrator.BuildGate") as mock_build_gate, + patch("src.quality_orchestrator.LintGate") as mock_lint_gate, + patch("src.quality_orchestrator.TestGate") as mock_test_gate, + patch("src.quality_orchestrator.CoverageGate") as mock_coverage_gate, + ): + # Configure mocks - one raises exception + mock_build_gate.return_value.check.return_value = mock_build_result + mock_lint_gate.return_value.check.side_effect = RuntimeError( + "Lint gate crashed" + ) + mock_test_gate.return_value.check.return_value = GateResult( + passed=True, message="Test passed", details={} + ) + mock_coverage_gate.return_value.check.return_value = GateResult( + passed=True, message="Coverage passed", details={} + ) + + # Verify completion + result = await orchestrator.verify_completion() + + # Assert result - exception should be converted to failure + assert isinstance(result, VerificationResult) + assert result.all_passed is False + assert result.gate_results["lint"].passed is False + assert "error" in result.gate_results["lint"].message.lower() + assert result.gate_results["build"].passed is True + + @pytest.mark.asyncio + async def test_verify_completion_all_gates_fail( + self, orchestrator: QualityOrchestrator + ) -> None: + """Test that verify_completion fails when all gates fail.""" + # Mock all gates to return failing results + mock_build_result = GateResult( + passed=False, + message="Build gate failed", + details={}, + ) + mock_lint_result = GateResult( + passed=False, + message="Lint gate failed", + details={}, + ) + mock_test_result = GateResult( + passed=False, + message="Test gate failed", + details={}, + ) + mock_coverage_result = GateResult( + passed=False, + message="Coverage gate failed", + details={}, + ) + + with ( + patch("src.quality_orchestrator.BuildGate") as mock_build_gate, + patch("src.quality_orchestrator.LintGate") as mock_lint_gate, + patch("src.quality_orchestrator.TestGate") as mock_test_gate, + patch("src.quality_orchestrator.CoverageGate") as mock_coverage_gate, + ): + # Configure mocks + mock_build_gate.return_value.check.return_value = mock_build_result + mock_lint_gate.return_value.check.return_value = mock_lint_result + mock_test_gate.return_value.check.return_value = mock_test_result + mock_coverage_gate.return_value.check.return_value = mock_coverage_result + + # Verify completion + result = await orchestrator.verify_completion() + + # Assert result + assert isinstance(result, VerificationResult) + assert result.all_passed is False + assert result.gate_results["build"].passed is False + assert result.gate_results["lint"].passed is False + assert result.gate_results["test"].passed is False + assert result.gate_results["coverage"].passed is False