forked from HAL9000/cleveragents-core
Luis Mendes
3cf3f1f69e
Implemented FixThenRevalidateOrchestrator with full diagnosis → self-fix → re-validation → retry limit → strategy revision → user escalation → terminal failure flow. Added retry counting per validation per plan, configurable retry limit (default 3), auto_strategy_revision flag support, and validation_fix_history recording in plan execution metadata. Review fixes applied (round 1): - Added try/except around fix_callback and revalidate_callback (spec: validation errors treated as required failure regardless of mode) - Wired optional EventBus for VALIDATION_FIX_ATTEMPTED/SUCCEEDED/EXHAUSTED events - Fixed total_attempts derivation from fix_history length instead of cumulative retry counts - Updated module docstring to reflect implemented vs caller-responsible steps - Added exhausted-retry logging on re-invocation - Replaced bare assert in Robot helper with explicit _check() function - Added 10 new Behave scenarios covering exception paths, multi-failure, reset, field validation, boundary values, event_bus property, and exhausted re-invocation Review fixes applied (round 2 — PR #711): - B1+B7: Fixed TOCTOU race in _fix_single_validation; atomic claim-per-iteration under RLock; eliminated defaultdict auto-vivification with .get() reads - B2: Added validation_name mismatch check on revalidate_callback return - B3: Added fix_description truncation to 2000 chars before FixAttemptRecord - T1: Switched threading.Lock to threading.RLock for defensive reentrancy - A1: Added model_validator preventing escalated+terminal_failure both True - R1: Added max_length=255 to FixAttemptRecord.validation_name - S7: Fixed misleading docstring about automation_profile integration - D4: Fixed timestamp field description to specify UTC - D1: Renamed misleading benchmark time_fix_after_two_retries - Added 5 new Behave scenarios for truncation, model constraints, name mismatch ISSUES CLOSED: #583
963 lines
33 KiB
Python
963 lines
33 KiB
Python
"""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)
|
|
)
|