"""Step definitions for Fix-then-Revalidate coverage boost scenarios. Covers validation edge cases, property accessors, error paths, callback exception handling, multi-failure scenarios, and boundary tests to ensure >=97% coverage on fix_then_revalidate.py. """ from __future__ import annotations from typing import Any from behave import given, then, when from behave.runner import Context from pydantic import ValidationError as PydanticValidationError from cleveragents.application.services.fix_then_revalidate import ( FixAttemptRecord, FixThenRevalidateOrchestrator, FixThenRevalidateResult, ) from cleveragents.application.services.validation_pipeline import ( ValidationPipeline, ValidationResult, ) from cleveragents.core.exceptions import ValidationError from cleveragents.domain.models.core.tool import ValidationMode from cleveragents.infrastructure.events.models import DomainEvent from cleveragents.infrastructure.events.types import EventType # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _mock_executor(validation_name: str, arguments: dict[str, Any]) -> dict[str, Any]: return {"passed": True, "message": f"{validation_name} ok"} def _make_pipeline() -> ValidationPipeline: return ValidationPipeline(commands=[], executor=_mock_executor) def _always_fix(result: ValidationResult) -> str: return f"Fixed {result.validation_name}" def _always_pass(result: ValidationResult) -> ValidationResult: return ValidationResult( validation_name=result.validation_name, resource_id=result.resource_id, resource_name=result.resource_name, mode=result.mode, passed=True, message="passed", data=result.data, duration_ms=1.0, ) def _make_failed_result( name: str, mode: ValidationMode = ValidationMode.REQUIRED, ) -> ValidationResult: 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, ) def _raising_fix(result: ValidationResult) -> str: raise RuntimeError(f"Fix callback crash for {result.validation_name}") class _RaisingRevalidateCallback: """Revalidate callback that always raises.""" def __call__(self, result: ValidationResult) -> ValidationResult: raise RuntimeError(f"Revalidate callback crash for {result.validation_name}") 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, ) class _CountingRevalidateCallback: """Revalidate callback that passes after N calls.""" def __init__(self, pass_after: int) -> None: self._pass_after = pass_after self._count = 0 def __call__(self, result: ValidationResult) -> ValidationResult: self._count += 1 passed = self._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._count})", data=result.data, duration_ms=5.0, ) class _PerValidationRevalidateCallback: """Revalidate callback with per-validation pass-after config.""" def __init__(self, config: dict[str, int | None]) -> None: self._config = config self._counts: dict[str, int] = {} def __call__(self, result: ValidationResult) -> ValidationResult: vname = result.validation_name self._counts[vname] = self._counts.get(vname, 0) + 1 pass_after = self._config.get(vname) passed = pass_after is not None and self._counts[vname] >= 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._counts[vname]})", data=result.data, duration_ms=5.0, ) # --------------------------------------------------------------------------- # Given steps # --------------------------------------------------------------------------- @given("a fix-revalidate coverage orchestrator with max_retries {n:d}") def step_given_coverage_orchestrator(context: Context, n: int) -> None: context.frc_orchestrator = FixThenRevalidateOrchestrator( validation_pipeline=_make_pipeline(), max_retries=n, auto_strategy_revision=1.0, ) context.frc_error = None context.frc_failed_results = [] context.frc_fix_callback = _always_fix context.frc_revalidate_callback = _CountingRevalidateCallback(pass_after=1) @given("a fix-revalidate coverage orchestrator with auto_strategy_revision enabled") def step_given_coverage_orchestrator_asr(context: Context) -> None: context.frc_orchestrator = FixThenRevalidateOrchestrator( validation_pipeline=_make_pipeline(), max_retries=3, auto_strategy_revision=0.0, ) context.frc_error = None context.frc_failed_results = [] context.frc_fix_callback = _always_fix context.frc_revalidate_callback = _CountingRevalidateCallback(pass_after=1) @given('a required coverage validation "{name}" that fails') def step_given_required_coverage_validation(context: Context, name: str) -> None: context.frc_failed_results.append(_make_failed_result(name)) @given("a fix-revalidate coverage fix callback that raises an exception") def step_given_raising_fix_callback(context: Context) -> None: context.frc_fix_callback = _raising_fix @given("a fix-revalidate coverage fix callback that always succeeds") def step_given_always_fix_callback(context: Context) -> None: context.frc_fix_callback = _always_fix @given("a fix-revalidate coverage revalidate callback that never passes") def step_given_never_pass_revalidate(context: Context) -> None: context.frc_revalidate_callback = _NeverPassRevalidateCallback() @given("a fix-revalidate coverage revalidate callback that raises an exception") def step_given_raising_revalidate(context: Context) -> None: context.frc_revalidate_callback = _RaisingRevalidateCallback() @given( "a fix-revalidate coverage revalidate callback where " "check-a passes after 1 and check-b never passes" ) def step_given_per_validation_revalidate(context: Context) -> None: context.frc_revalidate_callback = _PerValidationRevalidateCallback( config={"check-a": 1, "check-b": None} ) @given("a fix-revalidate coverage revalidate callback that passes after {n:d}") def step_given_counting_revalidate(context: Context, n: int) -> None: context.frc_revalidate_callback = _CountingRevalidateCallback(pass_after=n) # --------------------------------------------------------------------------- # When steps (constructor validation) # --------------------------------------------------------------------------- @when("the fix-revalidate orchestrator is created with None pipeline") def step_when_create_none_pipeline(context: Context) -> None: context.frc_error = None try: bad_pipeline: Any = None FixThenRevalidateOrchestrator( validation_pipeline=bad_pipeline, max_retries=3, ) except ValidationError as exc: context.frc_error = str(exc) @when("the fix-revalidate orchestrator is created with non-integer max_retries") def step_when_create_non_int_max_retries(context: Context) -> None: context.frc_error = None try: bad_max_retries: Any = "three" FixThenRevalidateOrchestrator( validation_pipeline=_make_pipeline(), max_retries=bad_max_retries, ) except ValidationError as exc: context.frc_error = str(exc) @when("the fix-revalidate orchestrator is created with max_retries {n:d}") def step_when_create_bad_max_retries(context: Context, n: int) -> None: context.frc_error = None try: FixThenRevalidateOrchestrator( validation_pipeline=_make_pipeline(), max_retries=n, ) except ValidationError as exc: context.frc_error = str(exc) # --------------------------------------------------------------------------- # When steps (method validation) # --------------------------------------------------------------------------- @when("the fix-revalidate get_retry_count is called with empty plan_id") def step_when_get_retry_count_empty_plan(context: Context) -> None: context.frc_error = None try: context.frc_orchestrator.get_retry_count("", "check-a") except ValidationError as exc: context.frc_error = str(exc) @when("the fix-revalidate get_retry_count is called with empty validation_name") def step_when_get_retry_count_empty_vname(context: Context) -> None: context.frc_error = None try: context.frc_orchestrator.get_retry_count("P1", "") except ValidationError as exc: context.frc_error = str(exc) @when("the fix-revalidate reset_retry_counts is called with empty plan_id") def step_when_reset_empty_plan(context: Context) -> None: context.frc_error = None try: context.frc_orchestrator.reset_retry_counts("") except ValidationError as exc: context.frc_error = str(exc) @when("the fix-revalidate run_fix_loop is called with empty plan_id") def step_when_run_loop_empty_plan(context: Context) -> None: context.frc_error = None try: context.frc_orchestrator.run_fix_loop( plan_id="", failed_results=[], fix_callback=_always_fix, revalidate_callback=_always_pass, ) except ValidationError as exc: context.frc_error = str(exc) @when("the fix-revalidate run_fix_loop is called with None failed_results") def step_when_run_loop_none_results(context: Context) -> None: context.frc_error = None try: bad_results: Any = None context.frc_orchestrator.run_fix_loop( plan_id="P1", failed_results=bad_results, fix_callback=_always_fix, revalidate_callback=_always_pass, ) except ValidationError as exc: context.frc_error = str(exc) @when("the fix-revalidate run_fix_loop is called with None fix_callback") def step_when_run_loop_none_fix(context: Context) -> None: context.frc_error = None try: bad_callback: Any = None context.frc_orchestrator.run_fix_loop( plan_id="P1", failed_results=[], fix_callback=bad_callback, revalidate_callback=_always_pass, ) except ValidationError as exc: context.frc_error = str(exc) @when("the fix-revalidate run_fix_loop is called with None revalidate_callback") def step_when_run_loop_none_revalidate(context: Context) -> None: context.frc_error = None try: bad_callback: Any = None context.frc_orchestrator.run_fix_loop( plan_id="P1", failed_results=[], fix_callback=_always_fix, revalidate_callback=bad_callback, ) except ValidationError as exc: context.frc_error = str(exc) @when('the fix-revalidate coverage loop runs for plan "{plan_id}"') def step_when_run_coverage_loop(context: Context, plan_id: str) -> None: result: FixThenRevalidateResult = context.frc_orchestrator.run_fix_loop( plan_id=plan_id, failed_results=context.frc_failed_results, fix_callback=context.frc_fix_callback, revalidate_callback=context.frc_revalidate_callback, ) context.frc_result = result @when('the fix-revalidate coverage retry counts are reset for plan "{plan_id}"') def step_when_reset_coverage_counts(context: Context, plan_id: str) -> None: context.frc_orchestrator.reset_retry_counts(plan_id) @when("a fix-revalidate coverage revalidate callback that passes after {n:d}") def step_when_set_counting_revalidate(context: Context, n: int) -> None: context.frc_revalidate_callback = _CountingRevalidateCallback(pass_after=n) @when( 'the fix-revalidate coverage loop runs again for plan "{plan_id}" ' 'with validation "{vname}"' ) def step_when_run_coverage_loop_again( context: Context, plan_id: str, vname: str ) -> None: failed_results = [_make_failed_result(vname)] result: FixThenRevalidateResult = context.frc_orchestrator.run_fix_loop( plan_id=plan_id, failed_results=failed_results, fix_callback=context.frc_fix_callback, revalidate_callback=context.frc_revalidate_callback, ) context.frc_result = result # --------------------------------------------------------------------------- # Then steps # --------------------------------------------------------------------------- @then('a fix-revalidate validation error should be raised with "{text}"') def step_then_validation_error(context: Context, text: str) -> None: assert context.frc_error is not None, ( "Expected a ValidationError but none was raised" ) assert text in context.frc_error, ( f"Expected error containing '{text}', got '{context.frc_error}'" ) @then("the fix-revalidate max_retries property should return {n:d}") def step_then_max_retries_property(context: Context, n: int) -> None: assert context.frc_orchestrator.max_retries == n, ( f"Expected max_retries={n}, got {context.frc_orchestrator.max_retries}" ) @then("the fix-revalidate auto_strategy_revision property should be 0.0") def step_then_asr_property(context: Context) -> None: assert context.frc_orchestrator.auto_strategy_revision == 0.0, ( "Expected auto_strategy_revision=0.0" ) @then( 'the fix-revalidate get_retry_count for plan "{plan_id}" ' 'validation "{vname}" should be {n:d}' ) def step_then_retry_count(context: Context, plan_id: str, vname: str, n: int) -> None: count = context.frc_orchestrator.get_retry_count(plan_id, vname) assert count == n, f"Expected retry count {n}, got {count}" @then("the fix-revalidate coverage result should have final_passed {expected}") def step_then_coverage_final_passed(context: Context, expected: str) -> None: result: FixThenRevalidateResult = context.frc_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 coverage result should be terminal failure") def step_then_coverage_terminal(context: Context) -> None: result: FixThenRevalidateResult = context.frc_result assert result.terminal_failure is True, "Expected terminal_failure=True" @then("the fix-revalidate coverage result should not be terminal failure") def step_then_coverage_not_terminal(context: Context) -> None: result: FixThenRevalidateResult = context.frc_result assert result.terminal_failure is False, "Expected terminal_failure=False" @then("the fix-revalidate coverage result should need user escalation") def step_then_coverage_needs_user_escalation(context: Context) -> None: result: FixThenRevalidateResult = context.frc_result assert result.needs_user_escalation is True, "Expected needs_user_escalation=True" @then("the fix-revalidate coverage validation fix history should have {n:d} records") def step_then_coverage_fix_history_count(context: Context, n: int) -> None: result: FixThenRevalidateResult = context.frc_result assert len(result.validation_fix_history) == n, ( f"Expected {n} fix history records, got {len(result.validation_fix_history)}" ) @then( "each fix-revalidate coverage validation fix history record " "should have success False" ) def step_then_all_records_failed(context: Context) -> None: result: FixThenRevalidateResult = context.frc_result for i, rec in enumerate(result.validation_fix_history): assert rec.success is False, ( f"Record {i + 1} has success={rec.success}, expected False" ) @then("the fix-revalidate coverage result should have validation_attempts {n:d}") def step_then_coverage_total_attempts(context: Context, n: int) -> None: result: FixThenRevalidateResult = context.frc_result assert result.validation_attempts == n, ( f"Expected validation_attempts={n}, got {result.validation_attempts}" ) @then( "fix-revalidate coverage validation fix history record {idx:d} " 'should have validation_name "{name}"' ) def step_then_record_vname(context: Context, idx: int, name: str) -> None: result: FixThenRevalidateResult = context.frc_result record = result.validation_fix_history[idx - 1] assert record.validation_name == name, ( f"Record {idx} validation_name={record.validation_name}, expected {name}" ) @then( "fix-revalidate coverage validation fix history record {idx:d} " "should have attempt_number {n:d}" ) def step_then_record_attempt_number(context: Context, idx: int, n: int) -> None: result: FixThenRevalidateResult = context.frc_result record = result.validation_fix_history[idx - 1] assert record.attempt_number == n, ( f"Record {idx} attempt_number={record.attempt_number}, expected {n}" ) @then( "fix-revalidate coverage validation fix history record {idx:d} " "should have success {expected}" ) def step_then_record_success(context: Context, idx: int, expected: str) -> None: result: FixThenRevalidateResult = context.frc_result record = result.validation_fix_history[idx - 1] expected_bool = expected.strip().lower() == "true" assert record.success is expected_bool, ( f"Record {idx} success={record.success}, expected {expected_bool}" ) @then("the fix-revalidate event_bus property should be None") def step_then_event_bus_none(context: Context) -> None: assert context.frc_orchestrator.event_bus is None, "Expected event_bus=None" # --------------------------------------------------------------------------- # Mock EventBus for M4 review fix # --------------------------------------------------------------------------- class _MockEventBus: """Mock EventBus that records emitted events.""" def __init__(self) -> None: self.events: list[DomainEvent] = [] def emit(self, event: DomainEvent) -> None: self.events.append(event) def subscribe( self, event_type: EventType, handler: Any, ) -> None: pass # Not needed for this test class _FailingEventBus: """Mock EventBus whose emit always raises.""" def emit(self, event: DomainEvent) -> None: raise RuntimeError("EventBus emit failure") def subscribe( self, event_type: EventType, handler: Any, ) -> None: pass # --------------------------------------------------------------------------- # Given steps for new review-fix scenarios # --------------------------------------------------------------------------- @given( "a fix-revalidate coverage orchestrator with max_retries {n:d} and a mock event bus" ) def step_given_coverage_orchestrator_with_mock_bus(context: Context, n: int) -> None: context.frc_mock_event_bus = _MockEventBus() context.frc_orchestrator = FixThenRevalidateOrchestrator( validation_pipeline=_make_pipeline(), max_retries=n, auto_strategy_revision=1.0, event_bus=context.frc_mock_event_bus, ) context.frc_error = None context.frc_failed_results = [] context.frc_fix_callback = _always_fix context.frc_revalidate_callback = _CountingRevalidateCallback(pass_after=1) @given( "a fix-revalidate coverage orchestrator with max_retries {n:d} " "and a failing event bus" ) def step_given_coverage_orchestrator_with_failing_bus(context: Context, n: int) -> None: context.frc_mock_event_bus = _FailingEventBus() context.frc_orchestrator = FixThenRevalidateOrchestrator( validation_pipeline=_make_pipeline(), auto_strategy_revision=1.0, max_retries=n, event_bus=context.frc_mock_event_bus, ) context.frc_error = None context.frc_failed_results = [] context.frc_fix_callback = _always_fix context.frc_revalidate_callback = _CountingRevalidateCallback(pass_after=1) # --------------------------------------------------------------------------- # When steps for new review-fix scenarios # --------------------------------------------------------------------------- @when("the fix-revalidate orchestrator is created with boolean max_retries True") def step_when_create_bool_true_max_retries(context: Context) -> None: context.frc_error = None try: bad_retries: Any = True FixThenRevalidateOrchestrator( validation_pipeline=_make_pipeline(), max_retries=bad_retries, ) except ValidationError as exc: context.frc_error = str(exc) @when("the fix-revalidate orchestrator is created with boolean max_retries False") def step_when_create_bool_false_max_retries(context: Context) -> None: context.frc_error = None try: bad_retries: Any = False FixThenRevalidateOrchestrator( validation_pipeline=_make_pipeline(), max_retries=bad_retries, ) except ValidationError as exc: context.frc_error = str(exc) @when("a fix-revalidate FixAttemptRecord is created with attempt_number 0") def step_when_create_record_bad_attempt(context: Context) -> None: context.frc_pydantic_error = None try: FixAttemptRecord( attempt_number=0, validation_name="check-a", fix_description="test", success=False, ) except PydanticValidationError as exc: context.frc_pydantic_error = str(exc) @when("a fix-revalidate FixAttemptRecord is created with whitespace validation_name") def step_when_create_record_whitespace_vname(context: Context) -> None: context.frc_pydantic_error = None try: FixAttemptRecord( attempt_number=1, validation_name=" ", fix_description="test", success=False, ) except PydanticValidationError as exc: context.frc_pydantic_error = str(exc) # --------------------------------------------------------------------------- # Then steps for new review-fix scenarios # --------------------------------------------------------------------------- @then("the mock event bus should have received {n:d} events") def step_then_mock_bus_event_count(context: Context, n: int) -> None: bus: _MockEventBus = context.frc_mock_event_bus assert len(bus.events) == n, ( f"Expected {n} events, got {len(bus.events)}: " f"{[e.event_type for e in bus.events]}" ) @then('mock event bus event {idx:d} should have event_type "{etype}"') def step_then_mock_bus_event_type(context: Context, idx: int, etype: str) -> None: bus: _MockEventBus = context.frc_mock_event_bus event = bus.events[idx - 1] assert event.event_type.value == etype, ( f"Event {idx} type={event.event_type.value}, expected {etype}" ) @then("a fix-revalidate pydantic validation error should be raised") def step_then_pydantic_error(context: Context) -> None: assert context.frc_pydantic_error is not None, ( "Expected a Pydantic ValidationError but none was raised" ) # --------------------------------------------------------------------------- # Steps for B3, A1, R1, B2 review fixes # --------------------------------------------------------------------------- def _long_description_fix(result: ValidationResult) -> str: """Fix callback that returns a description exceeding 2000 chars.""" return "x" * 3000 class _MismatchedRevalidateCallback: """Revalidate callback that returns a result with the wrong name.""" def __call__(self, result: ValidationResult) -> ValidationResult: return ValidationResult( validation_name="wrong-validation-name", resource_id=result.resource_id, resource_name=result.resource_name, mode=result.mode, passed=True, message="passed but with wrong name", data=result.data, duration_ms=1.0, ) @given("a fix-revalidate coverage fix callback that returns a 3000 char description") def step_given_long_fix_callback(context: Context) -> None: context.frc_fix_callback = _long_description_fix @given( "a fix-revalidate coverage revalidate callback " "that returns a different validation_name" ) def step_given_mismatched_revalidate(context: Context) -> None: context.frc_revalidate_callback = _MismatchedRevalidateCallback() @when( "a fix-revalidate FixThenRevalidateResult is created with " "escalated True and terminal_failure True" ) def step_when_create_result_both_flags(context: Context) -> None: context.frc_pydantic_error = None try: FixThenRevalidateResult( validation_attempts=3, final_passed=False, escalated=True, terminal_failure=True, ) except PydanticValidationError as exc: context.frc_pydantic_error = str(exc) @when( "a fix-revalidate FixAttemptRecord is created " "with validation_name exceeding 255 chars" ) def step_when_create_record_long_vname(context: Context) -> None: context.frc_pydantic_error = None try: FixAttemptRecord( attempt_number=1, validation_name="v" * 256, fix_description="test", success=False, ) except PydanticValidationError as exc: context.frc_pydantic_error = str(exc) @then( "fix-revalidate coverage validation fix history record {idx:d} " "fix_description should be at most {n:d} chars" ) def step_then_record_fix_description_length(context: Context, idx: int, n: int) -> None: result: FixThenRevalidateResult = context.frc_result record = result.validation_fix_history[idx - 1] assert len(record.fix_description) <= n, ( f"Record {idx} fix_description has {len(record.fix_description)} chars, " f"expected at most {n}" ) # --------------------------------------------------------------------------- # New steps for review-fix round 2/3 coverage # --------------------------------------------------------------------------- @when("the fix-revalidate orchestrator is created with boolean auto_strategy_revision") def step_when_create_bool_asr(context: Context) -> None: context.frc_error = None try: bad_asr: Any = True FixThenRevalidateOrchestrator( validation_pipeline=_make_pipeline(), max_retries=3, auto_strategy_revision=bad_asr, ) except ValidationError as exc: context.frc_error = str(exc) @when("the fix-revalidate orchestrator is created with auto_strategy_revision {val}") def step_when_create_bad_asr_range(context: Context, val: str) -> None: context.frc_error = None try: FixThenRevalidateOrchestrator( validation_pipeline=_make_pipeline(), max_retries=3, auto_strategy_revision=float(val), ) except ValidationError as exc: context.frc_error = str(exc) @when("the fix-revalidate orchestrator is created with boolean auto_validation_fix") def step_when_create_bool_avf(context: Context) -> None: context.frc_error = None try: bad_avf: Any = True FixThenRevalidateOrchestrator( validation_pipeline=_make_pipeline(), max_retries=3, auto_validation_fix=bad_avf, ) except ValidationError as exc: context.frc_error = str(exc) @when("the fix-revalidate orchestrator is created with auto_validation_fix {val}") def step_when_create_bad_avf_range(context: Context, val: str) -> None: context.frc_error = None try: FixThenRevalidateOrchestrator( validation_pipeline=_make_pipeline(), max_retries=3, auto_validation_fix=float(val), ) except ValidationError as exc: context.frc_error = str(exc) @when("the fix-revalidate orchestrator is created with invalid event_bus type") def step_when_create_bad_event_bus(context: Context) -> None: context.frc_error = None try: bad_bus: Any = 42 FixThenRevalidateOrchestrator( validation_pipeline=_make_pipeline(), max_retries=3, event_bus=bad_bus, ) except ValidationError as exc: context.frc_error = str(exc) @when("the fix-revalidate run_fix_loop is called with non-callable fix_callback") def step_when_run_loop_non_callable_fix(context: Context) -> None: context.frc_error = None try: bad_fix: Any = "not_callable" context.frc_orchestrator.run_fix_loop( plan_id="P1", failed_results=[], fix_callback=bad_fix, revalidate_callback=_always_pass, ) except ValidationError as exc: context.frc_error = str(exc) @when("the fix-revalidate run_fix_loop is called with non-callable revalidate_callback") def step_when_run_loop_non_callable_revalidate(context: Context) -> None: context.frc_error = None try: bad_revalidate: Any = "not_callable" context.frc_orchestrator.run_fix_loop( plan_id="P1", failed_results=[], fix_callback=_always_fix, revalidate_callback=bad_revalidate, ) except ValidationError as exc: context.frc_error = str(exc) def _none_fix(result: ValidationResult) -> None: """Fix callback that returns None to signal unfixable.""" return None @given("a fix-revalidate coverage fix callback that returns None") def step_given_none_fix_callback(context: Context) -> None: context.frc_fix_callback = _none_fix class _PassedWithErrorRevalidateCallback: """Revalidate callback that returns passed=True with an error set.""" 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=True, message="passed but has error", data=result.data, duration_ms=1.0, error="runtime error in validation", ) @given("a fix-revalidate coverage revalidate callback that returns passed with error") def step_given_passed_with_error_revalidate(context: Context) -> None: context.frc_revalidate_callback = _PassedWithErrorRevalidateCallback() @when( "a fix-revalidate FixThenRevalidateResult is created " "with final_passed True and escalated True" ) def step_when_create_result_final_passed_escalated(context: Context) -> None: context.frc_pydantic_error = None try: FixThenRevalidateResult( validation_attempts=1, final_passed=True, escalated=True, ) except PydanticValidationError as exc: context.frc_pydantic_error = str(exc) @then("the fix-revalidate auto_validation_fix property should be {val}") def step_then_avf_property(context: Context, val: str) -> None: expected = float(val) actual = context.frc_orchestrator.auto_validation_fix assert actual == expected, f"Expected auto_validation_fix={expected}, got {actual}" @given( "a fix-revalidate coverage orchestrator with max_retries {n:d} " "and mock event bus and escalation" ) def step_given_coverage_orchestrator_mock_bus_escalation( context: Context, n: int ) -> None: context.frc_mock_event_bus = _MockEventBus() context.frc_orchestrator = FixThenRevalidateOrchestrator( validation_pipeline=_make_pipeline(), max_retries=n, auto_strategy_revision=0.0, event_bus=context.frc_mock_event_bus, ) context.frc_error = None context.frc_failed_results = [] context.frc_fix_callback = _always_fix context.frc_revalidate_callback = _CountingRevalidateCallback(pass_after=1) @given('an informational coverage validation "{name}" that fails') def step_given_informational_coverage_validation(context: Context, name: str) -> None: context.frc_failed_results.append( _make_failed_result(name, ValidationMode.INFORMATIONAL) )