forked from HAL9000/cleveragents-core
freemo
68f9871f33
structlog's cache_logger_on_first_use=True causes module-level loggers to permanently cache their processor chain on first use. Tests using capture_logs() reconfigure processors, but cached loggers never pick up the new configuration — resulting in empty capture lists. Fixed by adding custom capture context managers that: 1. Temporarily disable logger caching 2. Replace the module-level logger with a fresh uncached instance 3. Restore original logger and config on exit Fixes 11 LSP server stub scenarios and 2 retry policy wiring scenarios.
329 lines
11 KiB
Python
329 lines
11 KiB
Python
"""Step definitions for ServiceRetryWiring async and advanced tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from typing import Any
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.core.retry_patterns import CircuitBreakerOpen
|
|
|
|
|
|
@then('all service policies should have backoff_strategy "{strategy}"')
|
|
def step_check_all_backoff(context: Any, strategy: str) -> None:
|
|
for name in context.wiring.registry.registered_services():
|
|
policy = context.wiring.registry.get(name)
|
|
actual = (
|
|
policy.retry.backoff_strategy.value
|
|
if hasattr(policy.retry.backoff_strategy, "value")
|
|
else str(policy.retry.backoff_strategy)
|
|
)
|
|
assert actual == strategy, (
|
|
f"Service {name} has backoff_strategy={actual}, expected {strategy}"
|
|
)
|
|
|
|
|
|
@when('I async_execute a function for "{name}" with open circuit')
|
|
def step_async_execute_open_circuit(context: Any, name: str) -> None:
|
|
loop = asyncio.new_event_loop()
|
|
try:
|
|
loop.run_until_complete(
|
|
context.wiring.async_execute(
|
|
service_name=name,
|
|
operation_name="test_async_cb_open",
|
|
func=_async_ok_func,
|
|
idempotent=True,
|
|
)
|
|
)
|
|
context.async_cb_error = None
|
|
except CircuitBreakerOpen as exc:
|
|
context.async_cb_error = exc
|
|
except Exception as exc:
|
|
context.async_cb_error = exc
|
|
finally:
|
|
loop.close()
|
|
|
|
|
|
async def _async_ok_func() -> str:
|
|
return "ok"
|
|
|
|
|
|
@then("a CircuitBreakerOpen exception should be raised for async_execute")
|
|
def step_check_async_cb_open(context: Any) -> None:
|
|
assert isinstance(context.async_cb_error, CircuitBreakerOpen)
|
|
|
|
|
|
@when("I execute the function through the wiring with structlog captured")
|
|
def step_execute_with_structlog_captured(context: Any) -> None:
|
|
import structlog
|
|
|
|
import cleveragents.core.retry_service_patterns as _rsp_mod
|
|
|
|
captured_entries: list[dict[str, Any]] = []
|
|
|
|
def capture_processor(
|
|
_logger: Any, _method: str, event_dict: dict[str, Any]
|
|
) -> dict[str, 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_structlog_op",
|
|
func=context.test_func,
|
|
idempotent=True,
|
|
)
|
|
context.exec_error = None
|
|
except Exception as exc:
|
|
context.exec_error = exc
|
|
finally:
|
|
# Restore original module logger and structlog configuration
|
|
_rsp_mod.logger = saved_rsp_logger
|
|
structlog.configure(**old_config)
|
|
|
|
context.captured_log_entries = captured_entries
|
|
|
|
|
|
@when("I execute a succeeding function for an unregistered service name")
|
|
def step_execute_for_unregistered_service(context: Any) -> None:
|
|
context.unregistered_name = "unregistered_lazy_test_svc"
|
|
# Ensure the name is NOT in the pre-built cache
|
|
assert context.unregistered_name not in context.wiring._wait_strategies
|
|
|
|
context.lazy_result = context.wiring.execute(
|
|
service_name=context.unregistered_name,
|
|
operation_name="lazy_build_op",
|
|
func=lambda: "lazy_ok",
|
|
idempotent=True,
|
|
)
|
|
|
|
|
|
@then("the wait strategy cache should contain the unregistered service")
|
|
def step_check_wait_strategy_cache(context: Any) -> None:
|
|
assert context.lazy_result == "lazy_ok"
|
|
assert context.unregistered_name in context.wiring._wait_strategies
|
|
|
|
|
|
@when("I async_execute a nested retry operation through the wiring")
|
|
def step_async_execute_nested(context: Any) -> None:
|
|
context.async_inner_call_count = 0
|
|
|
|
async def async_inner() -> str:
|
|
context.async_inner_call_count += 1
|
|
raise RuntimeError("async inner failure")
|
|
|
|
async def async_outer() -> str:
|
|
import contextlib as _ctxlib
|
|
|
|
with _ctxlib.suppress(RuntimeError):
|
|
await context.wiring.async_execute(
|
|
service_name="session_service",
|
|
operation_name="async_nested_inner",
|
|
func=async_inner,
|
|
idempotent=True,
|
|
)
|
|
return "async_outer_ok"
|
|
|
|
loop = asyncio.new_event_loop()
|
|
try:
|
|
context.async_nested_result = loop.run_until_complete(
|
|
context.wiring.async_execute(
|
|
service_name="plan_service",
|
|
operation_name="async_nested_outer",
|
|
func=async_outer,
|
|
idempotent=True,
|
|
)
|
|
)
|
|
finally:
|
|
loop.close()
|
|
|
|
|
|
@then("the async inner function should have been called only once")
|
|
def step_check_async_inner_once(context: Any) -> None:
|
|
assert context.async_nested_result == "async_outer_ok"
|
|
# Inner runs exactly once — nesting guard skips retries at depth >= 1
|
|
assert context.async_inner_call_count == 1
|
|
|
|
|
|
@when("I call wrap_service_method twice with the same arguments")
|
|
def step_call_wrap_twice(context: Any) -> None:
|
|
context.wrap_first = context.wiring.wrap_service_method(
|
|
"plan_service", "cache_test_op", idempotent=True
|
|
)
|
|
context.wrap_second = context.wiring.wrap_service_method(
|
|
"plan_service", "cache_test_op", idempotent=True
|
|
)
|
|
|
|
|
|
@then("both wrap_service_method calls should return the same object")
|
|
def step_check_wrap_same_object(context: Any) -> None:
|
|
assert context.wrap_first is context.wrap_second
|
|
|
|
|
|
@when("I async_execute the always-failing function as idempotent")
|
|
def step_async_execute_always_failing(context: Any) -> None:
|
|
loop = asyncio.new_event_loop()
|
|
try:
|
|
loop.run_until_complete(
|
|
context.wiring.async_execute(
|
|
service_name="plan_service",
|
|
operation_name="test_async_exhaustion",
|
|
func=context.async_test_func,
|
|
idempotent=True,
|
|
)
|
|
)
|
|
context.async_exhaustion_error = None
|
|
except RuntimeError as exc:
|
|
context.async_exhaustion_error = exc
|
|
except Exception as exc:
|
|
context.async_exhaustion_error = exc
|
|
finally:
|
|
loop.close()
|
|
|
|
|
|
@then("the async RuntimeError should propagate after exhausting retries")
|
|
def step_check_async_exhaustion_error(context: Any) -> None:
|
|
assert isinstance(context.async_exhaustion_error, RuntimeError)
|
|
assert "Permanent async failure" in str(context.async_exhaustion_error)
|
|
|
|
|
|
@then("the async always-failing function should have been called max_attempts times")
|
|
def step_check_async_always_fail_count(context: Any) -> None:
|
|
policy = context.wiring.get_policy("plan_service")
|
|
assert context.async_call_count == policy.retry.max_attempts
|
|
|
|
|
|
@then("the captured log should contain a retry warning message")
|
|
def step_check_captured_retry_log(context: Any) -> None:
|
|
assert context.exec_error is None
|
|
assert context.result == "success"
|
|
# Find a retry log entry
|
|
retry_entries = [
|
|
e
|
|
for e in context.captured_log_entries
|
|
if "retry" in str(e.get("event", "")).lower()
|
|
or e.get("operation") == "test_structlog_op"
|
|
]
|
|
assert len(retry_entries) > 0, (
|
|
f"No retry log entries found in {context.captured_log_entries}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _build_cached_wait with string backoff strategy
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("I have a ServiceRetryWiring instance with string backoff policy")
|
|
def step_create_wiring_string_backoff(context: Any) -> None:
|
|
from cleveragents.application.services.service_retry_wiring import (
|
|
ServiceRetryWiring,
|
|
)
|
|
from cleveragents.config.settings import Settings
|
|
|
|
settings = Settings(
|
|
database_url="sqlite:///test_retry.db",
|
|
mock_providers=True,
|
|
env="test",
|
|
)
|
|
context.wiring = ServiceRetryWiring(settings)
|
|
# Manually set a string backoff_strategy on a policy to test the fallback
|
|
policy = context.wiring.get_policy("plan_service")
|
|
policy.retry.backoff_strategy = "fixed" # type: ignore[assignment]
|
|
context.string_backoff_svc = "plan_service"
|
|
|
|
|
|
@when("I request the wait strategy for the string-backoff service")
|
|
def step_request_string_backoff_wait(context: Any) -> None:
|
|
# Force rebuild by clearing cache
|
|
context.wiring._wait_strategies.pop(context.string_backoff_svc, None)
|
|
context.string_backoff_wait = context.wiring._get_wait_strategy(
|
|
context.string_backoff_svc
|
|
)
|
|
|
|
|
|
@then("the wait strategy should be built successfully")
|
|
def step_check_string_backoff_built(context: Any) -> None:
|
|
assert context.string_backoff_wait is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Async execute without circuit breaker
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I async_execute the function for the unprotected service")
|
|
def step_async_execute_unprotected(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_no_cb",
|
|
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()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Sync nesting guard without CB
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I execute a nested sync operation for the unprotected service")
|
|
def step_execute_nested_no_cb(context: Any) -> None:
|
|
context.nested_no_cb_inner_count = 0
|
|
|
|
def inner_func() -> str:
|
|
context.nested_no_cb_inner_count += 1
|
|
return "inner_no_cb_ok"
|
|
|
|
def outer_func() -> str:
|
|
# This nested call should hit the nesting guard, CB=None path
|
|
return context.wiring.execute(
|
|
service_name="plan_service",
|
|
operation_name="nested_no_cb_inner",
|
|
func=inner_func,
|
|
idempotent=True,
|
|
)
|
|
|
|
context.nested_no_cb_result = context.wiring.execute(
|
|
service_name="plan_service",
|
|
operation_name="nested_no_cb_outer",
|
|
func=outer_func,
|
|
idempotent=True,
|
|
)
|
|
|
|
|
|
@then("the nested inner function should execute directly without CB")
|
|
def step_check_nested_no_cb(context: Any) -> None:
|
|
assert context.nested_no_cb_result == "inner_no_cb_ok"
|
|
assert context.nested_no_cb_inner_count == 1
|