Dev User
a3ba3c3eaf
Rename step texts to avoid case-sensitive conflicts between different step modules: - edge_case_plan_steps.py: 'a Pydantic validation error' -> 'an Edge Case Pydantic validation error' - plan_executor_coverage_boost_steps.py: 'the rollback result should be False' -> 'the executor rollback result should be False' - plan_explain_steps.py: 'the json output should be valid json' -> 'the plan explain json output should be valid' - plan_model_steps.py: 'I create a plan in strategize phase' -> 'I create a PlanModel in strategize phase' - project_repository_steps.py: 'the remove result should be False' -> 'the project repo remove result should be False' - service_retry_wiring_steps.py: 'I create a ServiceRetryWiring from those Settings' -> 'I create a ServiceRetryWiring from those retry Settings' - session_model_steps.py: 'I get the session CLI dict' -> 'I get the session model CLI dict' These pre-existing bugs prevented all behave tests from loading.
535 lines
17 KiB
Python
535 lines
17 KiB
Python
"""Step definitions for ServiceRetryWiring execute and async tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
from typing import Any
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.config.settings import Settings
|
|
from cleveragents.core.retry_patterns import CircuitBreakerOpen
|
|
|
|
|
|
@given("I have default Settings")
|
|
def step_default_settings(context: Any) -> None:
|
|
context.settings = Settings(
|
|
database_url="sqlite:///test_retry.db",
|
|
mock_providers=True,
|
|
env="test",
|
|
)
|
|
|
|
|
|
@when("I create a ServiceRetryWiring from Settings")
|
|
def step_create_wiring(context: Any) -> None:
|
|
from cleveragents.application.services.service_retry_wiring import (
|
|
ServiceRetryWiring,
|
|
)
|
|
|
|
context.wiring = ServiceRetryWiring(context.settings)
|
|
|
|
|
|
@then("the wiring should have circuit breakers for known services")
|
|
def step_check_wiring_cbs(context: Any) -> None:
|
|
cb = context.wiring.get_circuit_breaker("plan_service")
|
|
assert cb is not None
|
|
|
|
|
|
@then("the wiring registry should contain known service policies")
|
|
def step_check_wiring_registry(context: Any) -> None:
|
|
services = context.wiring.registry.registered_services()
|
|
assert "plan_service" in services
|
|
assert "session_service" in services
|
|
|
|
|
|
@given("I have a ServiceRetryWiring instance")
|
|
def step_create_wiring_instance(context: Any) -> None:
|
|
from cleveragents.application.services.service_retry_wiring import (
|
|
ServiceRetryWiring,
|
|
)
|
|
|
|
settings = Settings(
|
|
database_url="sqlite:///test_retry.db",
|
|
mock_providers=True,
|
|
env="test",
|
|
)
|
|
context.wiring = ServiceRetryWiring(settings)
|
|
|
|
|
|
@given("I have a function that fails once then succeeds")
|
|
def step_create_fail_once_func(context: Any) -> None:
|
|
context.call_count = 0
|
|
|
|
def fail_once() -> str:
|
|
context.call_count += 1
|
|
if context.call_count <= 1:
|
|
raise RuntimeError("Transient failure")
|
|
return "success"
|
|
|
|
context.test_func = fail_once
|
|
|
|
|
|
@when("I execute the function through the wiring as an idempotent operation")
|
|
def step_execute_idempotent(context: Any) -> None:
|
|
try:
|
|
context.result = context.wiring.execute(
|
|
service_name="plan_service",
|
|
operation_name="test_op",
|
|
func=context.test_func,
|
|
idempotent=True,
|
|
)
|
|
context.exec_error = None
|
|
except Exception as exc:
|
|
context.exec_error = exc
|
|
context.result = None
|
|
|
|
|
|
@then("the function should succeed after retry")
|
|
def step_check_succeeded(context: Any) -> None:
|
|
assert context.exec_error is None
|
|
assert context.result == "success"
|
|
|
|
|
|
@then("the function should have been called {n:d} times")
|
|
def step_check_call_count(context: Any, n: int) -> None:
|
|
assert context.call_count == n
|
|
|
|
|
|
@given("I have a function that fails on first call")
|
|
def step_create_fail_first(context: Any) -> None:
|
|
context.call_count = 0
|
|
|
|
def fail_first() -> str:
|
|
context.call_count += 1
|
|
raise RuntimeError("Permanent failure")
|
|
|
|
context.test_func = fail_first
|
|
|
|
|
|
@when("I execute the function through the wiring as a non-idempotent operation")
|
|
def step_execute_non_idempotent(context: Any) -> None:
|
|
try:
|
|
context.result = context.wiring.execute(
|
|
service_name="plan_service",
|
|
operation_name="test_write_op",
|
|
func=context.test_func,
|
|
idempotent=False,
|
|
)
|
|
context.exec_error = None
|
|
except Exception as exc:
|
|
context.exec_error = exc
|
|
context.result = None
|
|
|
|
|
|
@then("the function should fail immediately")
|
|
def step_check_failed(context: Any) -> None:
|
|
assert context.exec_error is not None
|
|
|
|
|
|
@then("the function should have been called exactly {n:d} time")
|
|
def step_check_exact_calls(context: Any, n: int) -> None:
|
|
assert context.call_count == n
|
|
|
|
|
|
@when('I execute a function through the wiring for "{name}"')
|
|
def step_execute_with_cb(context: Any, name: str) -> None:
|
|
try:
|
|
context.result = context.wiring.execute(
|
|
service_name=name,
|
|
operation_name="test_op",
|
|
func=lambda: "ok",
|
|
idempotent=True,
|
|
)
|
|
context.exec_error = None
|
|
except CircuitBreakerOpen as exc:
|
|
context.exec_error = exc
|
|
except Exception as exc:
|
|
context.exec_error = exc
|
|
|
|
|
|
@when('I get a decorator from wrap_service_method for "{name}"')
|
|
def step_get_decorator(context: Any, name: str) -> None:
|
|
context.decorator = context.wiring.wrap_service_method(
|
|
name, "test_op", idempotent=True
|
|
)
|
|
|
|
|
|
@then("the decorator should be callable")
|
|
def step_check_decorator_callable(context: Any) -> None:
|
|
assert callable(context.decorator)
|
|
|
|
|
|
@then("the settings should have {field_name} field")
|
|
def step_check_settings_field(context: Any, field_name: str) -> None:
|
|
assert hasattr(context.settings, field_name)
|
|
|
|
|
|
@given(
|
|
"I have Settings with retry_service_overrides setting plan_service max_attempts to {value:d}"
|
|
)
|
|
def step_settings_with_overrides(context: Any, value: int) -> None:
|
|
import json
|
|
import os
|
|
|
|
overrides = json.dumps({"plan_service": {"retry": {"max_attempts": value}}})
|
|
# Settings uses validation_alias so the field must be set via env var
|
|
os.environ["CLEVERAGENTS_RETRY_SERVICE_OVERRIDES"] = overrides
|
|
try:
|
|
context.settings = Settings(
|
|
database_url="sqlite:///test_retry.db",
|
|
mock_providers=True,
|
|
env="test",
|
|
)
|
|
finally:
|
|
os.environ.pop("CLEVERAGENTS_RETRY_SERVICE_OVERRIDES", None)
|
|
|
|
|
|
@when("I create a ServiceRetryWiring from those retry Settings")
|
|
def step_create_wiring_from_settings(context: Any) -> None:
|
|
from cleveragents.application.services.service_retry_wiring import (
|
|
ServiceRetryWiring,
|
|
)
|
|
|
|
context.wiring = ServiceRetryWiring(context.settings)
|
|
|
|
|
|
@when("I execute the function through the wiring with logging captured")
|
|
def step_execute_with_logging(context: Any) -> None:
|
|
import structlog
|
|
|
|
import cleveragents.core.retry_service_patterns as _rsp_mod
|
|
|
|
# U8: Capture actual structlog output so the Then step can verify
|
|
# that structured retry log messages were emitted.
|
|
captured_entries: list[dict[str, Any]] = []
|
|
|
|
def _capture_processor(_logger: Any, _method: str, event_dict: Any) -> Any:
|
|
captured_entries.append(event_dict.copy())
|
|
raise structlog.DropEvent
|
|
|
|
old_config = structlog.get_config()
|
|
structlog.configure(
|
|
processors=[_capture_processor],
|
|
wrapper_class=structlog.stdlib.BoundLogger,
|
|
context_class=dict,
|
|
logger_factory=structlog.PrintLoggerFactory(),
|
|
cache_logger_on_first_use=False,
|
|
)
|
|
|
|
# Replace the module-level logger in retry_service_patterns with a
|
|
# fresh lazy proxy. When configure_structlog() (called during
|
|
# container init in before_all) sets cache_logger_on_first_use=True,
|
|
# earlier scenarios cause the proxy to cache a resolved BoundLogger
|
|
# that ignores subsequent structlog.configure() calls. Creating a
|
|
# new proxy ensures our capture processor is used.
|
|
saved_rsp_logger = _rsp_mod.logger
|
|
_rsp_mod.logger = structlog.get_logger(_rsp_mod.__name__)
|
|
|
|
try:
|
|
context.result = context.wiring.execute(
|
|
service_name="plan_service",
|
|
operation_name="test_logged_op",
|
|
func=context.test_func,
|
|
idempotent=True,
|
|
)
|
|
context.exec_error = None
|
|
except Exception as exc:
|
|
context.exec_error = exc
|
|
finally:
|
|
_rsp_mod.logger = saved_rsp_logger
|
|
structlog.configure(**old_config)
|
|
|
|
context.captured_log_entries = captured_entries
|
|
|
|
|
|
@then("structured retry log messages should be present")
|
|
def step_check_log_messages(context: Any) -> None:
|
|
# The retry infrastructure emits logs on retry attempts.
|
|
# Verify the operation succeeded (meaning retries happened).
|
|
assert context.exec_error is None
|
|
assert context.result == "success"
|
|
assert context.call_count >= 2
|
|
# U8: Verify that at least one retry-related log entry was captured.
|
|
retry_entries = [
|
|
e
|
|
for e in context.captured_log_entries
|
|
if "retry" in str(e.get("event", "")).lower()
|
|
or e.get("operation") == "test_logged_op"
|
|
]
|
|
assert len(retry_entries) > 0, (
|
|
f"No retry log entries found in {context.captured_log_entries}"
|
|
)
|
|
|
|
|
|
@given("I have a function that always fails with RuntimeError")
|
|
def step_create_always_failing_func(context: Any) -> None:
|
|
context.always_fail_count = 0
|
|
|
|
def always_fail() -> str:
|
|
context.always_fail_count += 1
|
|
raise RuntimeError("Persistent failure")
|
|
|
|
context.always_fail_func = always_fail
|
|
|
|
|
|
@when("I execute the always-failing function through the wiring as idempotent")
|
|
def step_execute_always_failing(context: Any) -> None:
|
|
try:
|
|
context.result = context.wiring.execute(
|
|
service_name="plan_service",
|
|
operation_name="test_exhaustion",
|
|
func=context.always_fail_func,
|
|
idempotent=True,
|
|
)
|
|
context.exhaustion_error = None
|
|
except RuntimeError as exc:
|
|
context.exhaustion_error = exc
|
|
except Exception as exc:
|
|
context.exhaustion_error = exc
|
|
|
|
|
|
@then("the RuntimeError should propagate after exhausting retries")
|
|
def step_check_exhaustion_error(context: Any) -> None:
|
|
assert isinstance(context.exhaustion_error, RuntimeError)
|
|
assert "Persistent failure" in str(context.exhaustion_error)
|
|
|
|
|
|
@then("the always-failing function should have been called max_attempts times")
|
|
def step_check_always_fail_call_count(context: Any) -> None:
|
|
policy = context.wiring.get_policy("plan_service")
|
|
assert context.always_fail_count == policy.retry.max_attempts
|
|
|
|
|
|
@when("I decorate and call a function via wrap_service_method")
|
|
def step_wrap_and_call(context: Any) -> None:
|
|
decorator = context.wiring.wrap_service_method(
|
|
"session_service", "test_wrap_exec", idempotent=True
|
|
)
|
|
|
|
@decorator
|
|
def sample_func() -> str:
|
|
return "wrapped_result"
|
|
|
|
context.wrap_result = sample_func()
|
|
|
|
|
|
@then("the wrapped function should return its expected value")
|
|
def step_check_wrap_result(context: Any) -> None:
|
|
assert context.wrap_result == "wrapped_result"
|
|
|
|
|
|
@given('I have Settings with retry_service_overrides set to "{raw_json}"')
|
|
def step_settings_with_invalid_overrides(context: Any, raw_json: str) -> None:
|
|
import os
|
|
|
|
os.environ["CLEVERAGENTS_RETRY_SERVICE_OVERRIDES"] = raw_json
|
|
try:
|
|
context.settings = Settings(
|
|
database_url="sqlite:///test_retry.db",
|
|
mock_providers=True,
|
|
env="test",
|
|
)
|
|
finally:
|
|
os.environ.pop("CLEVERAGENTS_RETRY_SERVICE_OVERRIDES", None)
|
|
|
|
|
|
@when("I create a ServiceRetryWiring from those gracefully-invalid Settings")
|
|
def step_create_wiring_invalid_overrides(context: Any) -> None:
|
|
from cleveragents.application.services.service_retry_wiring import (
|
|
ServiceRetryWiring,
|
|
)
|
|
|
|
context.graceful_error = None
|
|
try:
|
|
context.wiring = ServiceRetryWiring(context.settings)
|
|
except Exception as exc:
|
|
context.graceful_error = exc
|
|
|
|
|
|
@then("the wiring should initialise without errors")
|
|
def step_check_graceful_init(context: Any) -> None:
|
|
assert context.graceful_error is None
|
|
assert context.wiring is not None
|
|
|
|
|
|
@given("I have an async function that fails once then succeeds")
|
|
def step_create_async_fail_once(context: Any) -> None:
|
|
context.async_call_count = 0
|
|
|
|
async def async_fail_once() -> str:
|
|
context.async_call_count += 1
|
|
if context.async_call_count <= 1:
|
|
raise RuntimeError("Transient async failure")
|
|
return "async_success"
|
|
|
|
context.async_test_func = async_fail_once
|
|
|
|
|
|
@when("I execute the async function through async_execute as idempotent")
|
|
def step_execute_async_idempotent(context: Any) -> None:
|
|
loop = asyncio.new_event_loop()
|
|
try:
|
|
context.async_result = loop.run_until_complete(
|
|
context.wiring.async_execute(
|
|
service_name="plan_service",
|
|
operation_name="test_async_op",
|
|
func=context.async_test_func,
|
|
idempotent=True,
|
|
)
|
|
)
|
|
context.async_exec_error = None
|
|
except Exception as exc:
|
|
context.async_exec_error = exc
|
|
context.async_result = None
|
|
finally:
|
|
loop.close()
|
|
|
|
|
|
@then("the async function should succeed after retry")
|
|
def step_check_async_succeeded(context: Any) -> None:
|
|
assert context.async_exec_error is None
|
|
assert context.async_result == "async_success"
|
|
|
|
|
|
@then("the async function should have been called {n:d} times")
|
|
def step_check_async_call_count(context: Any, n: int) -> None:
|
|
assert context.async_call_count == n
|
|
|
|
|
|
@given("I have an async function that always fails")
|
|
def step_create_async_always_fails(context: Any) -> None:
|
|
context.async_call_count = 0
|
|
|
|
async def async_always_fail() -> str:
|
|
context.async_call_count += 1
|
|
raise RuntimeError("Permanent async failure")
|
|
|
|
context.async_test_func = async_always_fail
|
|
|
|
|
|
@when("I execute the async function through async_execute as non-idempotent")
|
|
def step_execute_async_non_idempotent(context: Any) -> None:
|
|
loop = asyncio.new_event_loop()
|
|
try:
|
|
context.async_result = loop.run_until_complete(
|
|
context.wiring.async_execute(
|
|
service_name="plan_service",
|
|
operation_name="test_async_write",
|
|
func=context.async_test_func,
|
|
idempotent=False,
|
|
)
|
|
)
|
|
context.async_exec_error = None
|
|
except RuntimeError as exc:
|
|
context.async_exec_error = exc
|
|
context.async_result = None
|
|
finally:
|
|
loop.close()
|
|
|
|
|
|
@then("the async function should fail immediately with RuntimeError")
|
|
def step_check_async_failed_immediately(context: Any) -> None:
|
|
assert isinstance(context.async_exec_error, RuntimeError)
|
|
|
|
|
|
@then("the async function call count should be {n:d}")
|
|
def step_check_async_fail_count(context: Any, n: int) -> None:
|
|
assert context.async_call_count == n
|
|
|
|
|
|
@given("I have a ServiceRetryWiring instance for a service without circuit breaker")
|
|
def step_create_wiring_no_cb(context: Any) -> None:
|
|
from cleveragents.application.services.service_retry_wiring import (
|
|
ServiceRetryWiring,
|
|
)
|
|
|
|
settings = Settings(
|
|
database_url="sqlite:///test_retry.db",
|
|
mock_providers=True,
|
|
env="test",
|
|
)
|
|
context.wiring = ServiceRetryWiring(settings)
|
|
# Manually remove the CB for a service to test the None path
|
|
context.wiring._circuit_breakers.pop("plan_service", None)
|
|
|
|
|
|
@when("I execute a function through the wiring for the unprotected service")
|
|
def step_execute_no_cb(context: Any) -> None:
|
|
try:
|
|
context.result = context.wiring.execute(
|
|
service_name="plan_service",
|
|
operation_name="test_no_cb",
|
|
func=lambda: "no_cb_result",
|
|
idempotent=True,
|
|
)
|
|
context.exec_error = None
|
|
except Exception as exc:
|
|
context.exec_error = exc
|
|
|
|
|
|
@then("the function should execute successfully without circuit breaker")
|
|
def step_check_no_cb_success(context: Any) -> None:
|
|
assert context.exec_error is None
|
|
assert context.result == "no_cb_result"
|
|
|
|
|
|
@given('I have Settings with non-default retry_backoff_strategy "{strategy}"')
|
|
def step_settings_backoff_strategy(context: Any, strategy: str) -> None:
|
|
import os
|
|
|
|
os.environ["CLEVERAGENTS_RETRY_BACKOFF_STRATEGY"] = strategy
|
|
try:
|
|
context.settings = Settings(
|
|
database_url="sqlite:///test_retry.db",
|
|
mock_providers=True,
|
|
env="test",
|
|
)
|
|
finally:
|
|
os.environ.pop("CLEVERAGENTS_RETRY_BACKOFF_STRATEGY", None)
|
|
|
|
|
|
@when("I create a ServiceRetryWiring from those backoff-strategy Settings")
|
|
def step_create_wiring_backoff(context: Any) -> None:
|
|
from cleveragents.application.services.service_retry_wiring import (
|
|
ServiceRetryWiring,
|
|
)
|
|
|
|
context.wiring = ServiceRetryWiring(context.settings)
|
|
|
|
|
|
@when("I execute a nested retry operation through the wiring")
|
|
def step_execute_nested_retry(context: Any) -> None:
|
|
context.inner_call_count = 0
|
|
context.outer_call_count = 0
|
|
|
|
def inner_func() -> str:
|
|
context.inner_call_count += 1
|
|
raise RuntimeError("inner failure")
|
|
|
|
def outer_func() -> str:
|
|
context.outer_call_count += 1
|
|
# This nested execute should NOT retry inner_func
|
|
with contextlib.suppress(RuntimeError):
|
|
context.wiring.execute(
|
|
service_name="session_service",
|
|
operation_name="nested_inner",
|
|
func=inner_func,
|
|
idempotent=True,
|
|
)
|
|
return "outer_ok"
|
|
|
|
context.nested_result = context.wiring.execute(
|
|
service_name="plan_service",
|
|
operation_name="nested_outer",
|
|
func=outer_func,
|
|
idempotent=True,
|
|
)
|
|
|
|
|
|
@then("the inner function should have been called only once per outer attempt")
|
|
def step_check_inner_call_count(context: Any) -> None:
|
|
# Inner should have been called exactly once per outer call
|
|
# (no retries due to nesting guard)
|
|
assert context.inner_call_count == context.outer_call_count
|