"""Step definitions for Fix-then-Revalidate orchestration scenarios. All step names are prefixed with 'fix-revalidate' context to avoid AmbiguousStep collisions with existing step definitions. """ from __future__ import annotations from typing import Any from behave import given, then, when from behave.runner import Context from cleveragents.application.services.fix_then_revalidate import ( FixThenRevalidateOrchestrator, FixThenRevalidateResult, ) from cleveragents.application.services.validation_pipeline import ( ValidationPipeline, ValidationResult, ) from cleveragents.domain.models.core.tool import ValidationMode # --------------------------------------------------------------------------- # Mock helpers # --------------------------------------------------------------------------- def _mock_executor(validation_name: str, arguments: dict[str, Any]) -> dict[str, Any]: """Always-passing mock executor for pipeline construction.""" return {"passed": True, "message": f"{validation_name} ok"} def _make_failed_result( name: str, mode: ValidationMode = ValidationMode.REQUIRED, ) -> ValidationResult: """Create a failed ValidationResult for testing.""" return ValidationResult( validation_name=name, resource_id="RES0001", resource_name="test-resource", mode=mode, passed=False, message=f"{name} failed: expected condition not met", data={"hint": "check configuration"}, duration_ms=10.0, ) class _CountingRevalidateCallback: """Revalidate callback that passes after a configurable number of fixes.""" def __init__(self, pass_after: int) -> None: self._pass_after = pass_after self._call_count = 0 def __call__(self, result: ValidationResult) -> ValidationResult: self._call_count += 1 passed = self._call_count >= self._pass_after return ValidationResult( validation_name=result.validation_name, resource_id=result.resource_id, resource_name=result.resource_name, mode=result.mode, passed=passed, message=f"{'passed' if passed else 'still failing'} " f"(attempt {self._call_count})", data=result.data, duration_ms=5.0, ) class _NeverPassRevalidateCallback: """Revalidate callback that never passes.""" def __call__(self, result: ValidationResult) -> ValidationResult: return ValidationResult( validation_name=result.validation_name, resource_id=result.resource_id, resource_name=result.resource_name, mode=result.mode, passed=False, message="still failing", data=result.data, duration_ms=5.0, ) def _always_fix(result: ValidationResult) -> str: """Fix callback that always returns a fix description.""" return f"Applied fix for {result.validation_name}: adjusted configuration" # --------------------------------------------------------------------------- # Given steps # --------------------------------------------------------------------------- @given("a fix-revalidate orchestrator with max_retries {n:d}") def step_given_orchestrator(context: Context, n: int) -> None: pipeline = ValidationPipeline( commands=[], executor=_mock_executor, ) context.fr_pipeline = pipeline context.fr_orchestrator = FixThenRevalidateOrchestrator( validation_pipeline=pipeline, max_retries=n, auto_strategy_revision=1.0, ) context.fr_failed_results = [] # list[ValidationResult] context.fr_fix_callback = _always_fix context.fr_revalidate_callback = _CountingRevalidateCallback(pass_after=1) @given("a fix-revalidate orchestrator with auto_strategy_revision enabled") def step_given_orchestrator_auto_revision(context: Context) -> None: pipeline = ValidationPipeline( commands=[], executor=_mock_executor, ) context.fr_pipeline = pipeline context.fr_orchestrator = FixThenRevalidateOrchestrator( validation_pipeline=pipeline, max_retries=3, auto_strategy_revision=0.0, ) context.fr_failed_results = [] context.fr_fix_callback = _always_fix context.fr_revalidate_callback = _CountingRevalidateCallback(pass_after=1) @given('a required validation "{name}" that fails initially') def step_given_required_validation_fails(context: Context, name: str) -> None: result = _make_failed_result(name, ValidationMode.REQUIRED) context.fr_failed_results.append(result) @given('an informational validation "{name}" that fails') def step_given_informational_validation_fails(context: Context, name: str) -> None: result = _make_failed_result(name, ValidationMode.INFORMATIONAL) context.fr_failed_results.append(result) @given("a fix callback that always succeeds") def step_given_fix_callback(context: Context) -> None: context.fr_fix_callback = _always_fix @given("a revalidate callback that passes after {n:d} fix") def step_given_revalidate_passes_after_n(context: Context, n: int) -> None: context.fr_revalidate_callback = _CountingRevalidateCallback(pass_after=n) @given("a revalidate callback that passes after {n:d} fixes") def step_given_revalidate_passes_after_n_plural(context: Context, n: int) -> None: context.fr_revalidate_callback = _CountingRevalidateCallback(pass_after=n) @given("a revalidate callback that never passes") def step_given_revalidate_never_passes(context: Context) -> None: context.fr_revalidate_callback = _NeverPassRevalidateCallback() # --------------------------------------------------------------------------- # When steps # --------------------------------------------------------------------------- @when('the fix-revalidate loop runs for plan "{plan_id}"') def step_when_run_fix_loop(context: Context, plan_id: str) -> None: orchestrator: FixThenRevalidateOrchestrator = context.fr_orchestrator result: FixThenRevalidateResult = orchestrator.run_fix_loop( plan_id=plan_id, failed_results=context.fr_failed_results, fix_callback=context.fr_fix_callback, revalidate_callback=context.fr_revalidate_callback, ) context.fr_result = result # --------------------------------------------------------------------------- # Then steps # --------------------------------------------------------------------------- @then("the fix-revalidate result should have validation_attempts {n:d}") def step_then_total_attempts(context: Context, n: int) -> None: result: FixThenRevalidateResult = context.fr_result assert result.validation_attempts == n, ( f"Expected validation_attempts={n}, got {result.validation_attempts}" ) @then("the fix-revalidate result should have final_passed {expected}") def step_then_final_passed(context: Context, expected: str) -> None: result: FixThenRevalidateResult = context.fr_result expected_bool = expected.strip().lower() == "true" assert result.final_passed is expected_bool, ( f"Expected final_passed={expected_bool}, got {result.final_passed}" ) @then("the fix-revalidate result should not be escalated") def step_then_not_escalated(context: Context) -> None: result: FixThenRevalidateResult = context.fr_result assert result.escalated is False, "Expected escalated=False" @then("the fix-revalidate result should be escalated") def step_then_escalated(context: Context) -> None: result: FixThenRevalidateResult = context.fr_result assert result.escalated is True, "Expected escalated=True" @then("the fix-revalidate result should not be terminal failure") def step_then_not_terminal(context: Context) -> None: result: FixThenRevalidateResult = context.fr_result assert result.terminal_failure is False, "Expected terminal_failure=False" @then("the fix-revalidate result should be terminal failure") def step_then_terminal(context: Context) -> None: result: FixThenRevalidateResult = context.fr_result assert result.terminal_failure is True, "Expected terminal_failure=True" @then("the validation fix history should have {n:d} records") def step_then_fix_history_count(context: Context, n: int) -> None: result: FixThenRevalidateResult = context.fr_result assert len(result.validation_fix_history) == n, ( f"Expected {n} fix history records, got {len(result.validation_fix_history)}" ) @then("the fix-revalidate result should need user escalation") def step_then_needs_user_escalation(context: Context) -> None: result: FixThenRevalidateResult = context.fr_result assert result.needs_user_escalation is True, "Expected needs_user_escalation=True" @then("the fix-revalidate result should not need user escalation") def step_then_not_needs_user_escalation(context: Context) -> None: result: FixThenRevalidateResult = context.fr_result assert result.needs_user_escalation is False, "Expected needs_user_escalation=False"