forked from HAL9000/cleveragents-core
freemo
31472b5413
Add Behave feature/step pairs that exercise uncovered branches across handlers, LSP, CLI, and service layers to reach the coverage gate. ISSUES CLOSED: #1232
280 lines
10 KiB
Python
280 lines
10 KiB
Python
"""Step definitions for Fix-then-Revalidate coverage round 3 scenarios.
|
|
|
|
Covers uncovered lines: 354, 576, 650, 765-769, 771, 808 in
|
|
fix_then_revalidate.py.
|
|
|
|
All step names use the 'ftrcov3' prefix to avoid Behave AmbiguousStep errors
|
|
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
|
|
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]:
|
|
"""Always-passing mock executor for pipeline construction."""
|
|
return {"passed": True, "message": f"{validation_name} ok"}
|
|
|
|
|
|
def _make_pipeline() -> ValidationPipeline:
|
|
return ValidationPipeline(commands=[], executor=_mock_executor)
|
|
|
|
|
|
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 _always_fix(result: ValidationResult) -> str:
|
|
"""Fix callback that always returns a fix description."""
|
|
return f"Fixed {result.validation_name}"
|
|
|
|
|
|
def _integer_fix(result: ValidationResult) -> Any:
|
|
"""Fix callback that returns a non-string (integer) value."""
|
|
return 42
|
|
|
|
|
|
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 _FailingEventBus:
|
|
"""Mock EventBus whose emit always raises, used to trigger the circuit breaker."""
|
|
|
|
def emit(self, event: DomainEvent) -> None:
|
|
raise RuntimeError("EventBus emit failure")
|
|
|
|
def subscribe(
|
|
self,
|
|
event_type: EventType,
|
|
handler: Any,
|
|
) -> None:
|
|
pass
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a ftrcov3 orchestrator with max_retries {n:d}")
|
|
def step_given_ftrcov3_orchestrator(context: Context, n: int) -> None:
|
|
context.ftrcov3_orchestrator = FixThenRevalidateOrchestrator(
|
|
validation_pipeline=_make_pipeline(),
|
|
max_retries=n,
|
|
auto_strategy_revision=1.0,
|
|
)
|
|
context.ftrcov3_error = None
|
|
context.ftrcov3_failed_results = []
|
|
context.ftrcov3_fix_callback = _always_fix
|
|
context.ftrcov3_revalidate_callback = _CountingRevalidateCallback(pass_after=1)
|
|
|
|
|
|
@given(
|
|
'the ftrcov3 internal retry count for plan "{plan_id}" '
|
|
'validation "{vname}" resource "{rid}" is set to {n:d}'
|
|
)
|
|
def step_given_ftrcov3_seed_retry_count(
|
|
context: Context, plan_id: str, vname: str, rid: str, n: int
|
|
) -> None:
|
|
"""Directly set internal retry counts to simulate pre-exhausted budget."""
|
|
orch: FixThenRevalidateOrchestrator = context.ftrcov3_orchestrator
|
|
# Access the internal defaultdict to seed the retry count.
|
|
# This simulates a scenario where retries were consumed by a
|
|
# concurrent thread or a previous partial invocation that didn't reset.
|
|
orch._retry_counts[plan_id][(vname, rid)] = n
|
|
|
|
|
|
@given('a ftrcov3 required validation "{name}" that fails')
|
|
def step_given_ftrcov3_required_validation(context: Context, name: str) -> None:
|
|
context.ftrcov3_failed_results.append(_make_failed_result(name))
|
|
|
|
|
|
@given("a ftrcov3 fix callback that always succeeds")
|
|
def step_given_ftrcov3_always_fix(context: Context) -> None:
|
|
context.ftrcov3_fix_callback = _always_fix
|
|
|
|
|
|
@given("a ftrcov3 fix callback that returns an integer")
|
|
def step_given_ftrcov3_integer_fix(context: Context) -> None:
|
|
context.ftrcov3_fix_callback = _integer_fix
|
|
|
|
|
|
@given("a ftrcov3 revalidate callback that never passes")
|
|
def step_given_ftrcov3_never_pass(context: Context) -> None:
|
|
context.ftrcov3_revalidate_callback = _NeverPassRevalidateCallback()
|
|
|
|
|
|
@given("a ftrcov3 revalidate callback that passes after {n:d} fix")
|
|
def step_given_ftrcov3_passes_after(context: Context, n: int) -> None:
|
|
context.ftrcov3_revalidate_callback = _CountingRevalidateCallback(pass_after=n)
|
|
|
|
|
|
@given("a ftrcov3 orchestrator with max_retries {n:d} and a failing event bus")
|
|
def step_given_ftrcov3_orchestrator_failing_bus(context: Context, n: int) -> None:
|
|
context.ftrcov3_failing_event_bus = _FailingEventBus()
|
|
context.ftrcov3_orchestrator = FixThenRevalidateOrchestrator(
|
|
validation_pipeline=_make_pipeline(),
|
|
max_retries=n,
|
|
auto_strategy_revision=1.0,
|
|
event_bus=context.ftrcov3_failing_event_bus,
|
|
)
|
|
context.ftrcov3_error = None
|
|
context.ftrcov3_failed_results = []
|
|
context.ftrcov3_fix_callback = _always_fix
|
|
context.ftrcov3_revalidate_callback = _CountingRevalidateCallback(pass_after=1)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('the ftrcov3 fix loop runs for plan "{plan_id}"')
|
|
def step_when_ftrcov3_run_fix_loop(context: Context, plan_id: str) -> None:
|
|
orch: FixThenRevalidateOrchestrator = context.ftrcov3_orchestrator
|
|
try:
|
|
result: FixThenRevalidateResult = orch.run_fix_loop(
|
|
plan_id=plan_id,
|
|
failed_results=context.ftrcov3_failed_results,
|
|
fix_callback=context.ftrcov3_fix_callback,
|
|
revalidate_callback=context.ftrcov3_revalidate_callback,
|
|
)
|
|
context.ftrcov3_result = result
|
|
except Exception as exc:
|
|
context.ftrcov3_error = exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the ftrcov3 result should have final_passed {expected}")
|
|
def step_then_ftrcov3_final_passed(context: Context, expected: str) -> None:
|
|
assert context.ftrcov3_error is None, f"Unexpected error: {context.ftrcov3_error}"
|
|
result: FixThenRevalidateResult = context.ftrcov3_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 ftrcov3 result should need user escalation")
|
|
def step_then_ftrcov3_needs_user_escalation(context: Context) -> None:
|
|
assert context.ftrcov3_error is None, f"Unexpected error: {context.ftrcov3_error}"
|
|
result: FixThenRevalidateResult = context.ftrcov3_result
|
|
assert result.needs_user_escalation is True, (
|
|
f"Expected needs_user_escalation=True, got {result.needs_user_escalation}"
|
|
)
|
|
|
|
|
|
@then("the ftrcov3 result should have validation_attempts {n:d}")
|
|
def step_then_ftrcov3_validation_attempts(context: Context, n: int) -> None:
|
|
assert context.ftrcov3_error is None, f"Unexpected error: {context.ftrcov3_error}"
|
|
result: FixThenRevalidateResult = context.ftrcov3_result
|
|
assert result.validation_attempts == n, (
|
|
f"Expected validation_attempts={n}, got {result.validation_attempts}"
|
|
)
|
|
|
|
|
|
@then(
|
|
'the ftrcov3 get_retry_count for plan "{plan_id}" '
|
|
'validation "{vname}" resource "{rid}" should be {n:d}'
|
|
)
|
|
def step_then_ftrcov3_get_retry_count(
|
|
context: Context, plan_id: str, vname: str, rid: str, n: int
|
|
) -> None:
|
|
orch: FixThenRevalidateOrchestrator = context.ftrcov3_orchestrator
|
|
actual = orch.get_retry_count(plan_id, vname, rid)
|
|
assert actual == n, f"Expected get_retry_count={n}, got {actual}"
|
|
|
|
|
|
@then('the ftrcov3 fix history record {idx:d} fix_description should contain "{text}"')
|
|
def step_then_ftrcov3_fix_description_contains(
|
|
context: Context, idx: int, text: str
|
|
) -> None:
|
|
assert context.ftrcov3_error is None, f"Unexpected error: {context.ftrcov3_error}"
|
|
result: FixThenRevalidateResult = context.ftrcov3_result
|
|
record = result.validation_fix_history[idx - 1]
|
|
assert text in record.fix_description, (
|
|
f"Record {idx} fix_description='{record.fix_description}', "
|
|
f"expected it to contain '{text}'"
|
|
)
|
|
|
|
|
|
@then("the ftrcov3 event bus consecutive failures should be at least {n:d}")
|
|
def step_then_ftrcov3_circuit_breaker(context: Context, n: int) -> None:
|
|
assert context.ftrcov3_error is None, f"Unexpected error: {context.ftrcov3_error}"
|
|
orch: FixThenRevalidateOrchestrator = context.ftrcov3_orchestrator
|
|
# Access internal state to verify the circuit breaker tripped
|
|
actual = orch._event_bus_consecutive_failures
|
|
assert actual >= n, f"Expected consecutive failures >= {n}, got {actual}"
|