freemo
4ad51561fc
Cap time.sleep and asyncio.sleep globally at 10ms in before_all to eliminate retry/backoff waits (tenacity wait_fixed, retry_auto_debug exponential backoff). Save originals as time._original_sleep and asyncio._original_sleep for tests that need real wall-clock delays (CircuitBreaker recovery, debounce timers, validation timeouts). Enhance template-DB patch: add "db." prefix to _SCENARIO_DB_PREFIXES and check db_path.stat().st_size > 0 so 0-byte auto-created SQLite files receive the template copy instead of falling through to real migrations. Replace subprocess.run CLI invocations with typer.testing.CliRunner in module_coverage, main_coverage_complete, and coverage_extras step files, eliminating ~6s Python cold-start overhead per call. Switch plan_persistence and action_persistence to in-memory SQLite by default; only cross-restart scenarios use file-based via an explicit Given step. Replace MigrationRunner.run_migrations with Base.metadata.create_all in plan_service_steps for freshly created in-memory engines. Removed leftover debug comment in environment.py. Total tier runtime: 565s -> 21s (96% reduction), 20 features all under 5s behave-internal time. ISSUES CLOSED: #480
1206 lines
42 KiB
Python
1206 lines
42 KiB
Python
"""Step definitions for retry patterns feature tests."""
|
|
|
|
import asyncio
|
|
import builtins
|
|
import contextlib
|
|
import logging
|
|
import time
|
|
from types import SimpleNamespace
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.core import retry_patterns as retry_patterns_module
|
|
from cleveragents.core.exceptions import (
|
|
NetworkError,
|
|
RateLimitError,
|
|
)
|
|
from cleveragents.core.retry_patterns import (
|
|
RETRY_CATEGORIES,
|
|
CircuitBreaker,
|
|
CircuitBreakerOpen,
|
|
RetryContext,
|
|
async_retry_with_exponential_backoff,
|
|
configure_retry_logging,
|
|
get_retry_decorator,
|
|
log_after_retry,
|
|
log_before_retry,
|
|
retry_auto_debug,
|
|
retry_file_operation,
|
|
retry_network_operation,
|
|
retry_on_result,
|
|
retry_provider_operation,
|
|
retry_with_exponential_backoff,
|
|
retry_with_jitter,
|
|
retry_with_timeout,
|
|
should_retry_result,
|
|
)
|
|
|
|
|
|
@given("I have the retry patterns module imported")
|
|
def step_retry_module_imported(context):
|
|
"""Verify retry patterns module is available."""
|
|
context.retry_module_available = True
|
|
assert retry_with_exponential_backoff is not None
|
|
assert CircuitBreaker is not None
|
|
assert RetryContext is not None
|
|
|
|
|
|
@given("I have a function that fails {num:d} times then succeeds")
|
|
def step_create_failing_function(context, num):
|
|
"""Create a function that fails specified number of times."""
|
|
context.fail_count = num
|
|
context.call_count = 0
|
|
|
|
def failing_function():
|
|
context.call_count += 1
|
|
if context.call_count <= context.fail_count:
|
|
raise Exception(f"Failure {context.call_count}")
|
|
return "success"
|
|
|
|
context.test_function = failing_function
|
|
|
|
|
|
@given("I have an async function that fails {num:d} times then succeeds")
|
|
def step_create_async_failing_function(context, num):
|
|
"""Create an async function that fails specified number of times."""
|
|
context.fail_count = num
|
|
context.call_count = 0
|
|
|
|
async def async_failing_function():
|
|
context.call_count += 1
|
|
if context.call_count <= context.fail_count:
|
|
raise Exception(f"Async failure {context.call_count}")
|
|
return "async success"
|
|
|
|
context.test_function = async_failing_function
|
|
|
|
|
|
@when("I apply exponential backoff retry with max {max_attempts:d} attempts")
|
|
def step_apply_exponential_retry(context, max_attempts):
|
|
"""Apply exponential backoff retry to function."""
|
|
decorated_func = retry_with_exponential_backoff(
|
|
max_attempts=max_attempts,
|
|
min_wait=0.01, # Short waits for testing
|
|
max_wait=0.1,
|
|
)(context.test_function)
|
|
|
|
try:
|
|
context.result = decorated_func()
|
|
context.retry_succeeded = True
|
|
except Exception as e:
|
|
context.retry_error = e
|
|
context.retry_succeeded = False
|
|
|
|
|
|
@when("I apply async exponential backoff retry with max {max_attempts:d} attempts")
|
|
def step_apply_async_exponential_retry(context, max_attempts):
|
|
"""Apply async exponential backoff retry to function."""
|
|
decorated_func = async_retry_with_exponential_backoff(
|
|
max_attempts=max_attempts, min_wait=0.01, max_wait=0.1
|
|
)(context.test_function)
|
|
|
|
async def run_async():
|
|
try:
|
|
result = await decorated_func()
|
|
return result, True
|
|
except Exception as e:
|
|
return e, False
|
|
|
|
loop = asyncio.new_event_loop()
|
|
result, succeeded = loop.run_until_complete(run_async())
|
|
loop.close()
|
|
|
|
if succeeded:
|
|
context.result = result
|
|
context.retry_succeeded = True
|
|
else:
|
|
context.retry_error = result
|
|
context.retry_succeeded = False
|
|
|
|
|
|
@then("the function should eventually succeed")
|
|
def step_verify_function_succeeded(context):
|
|
"""Verify function succeeded after retries."""
|
|
assert context.retry_succeeded, (
|
|
f"Function failed: {getattr(context, 'retry_error', 'Unknown error')}"
|
|
)
|
|
assert context.result == "success"
|
|
|
|
|
|
@then("the async function should eventually succeed")
|
|
def step_verify_async_function_succeeded(context):
|
|
"""Verify async function succeeded after retries."""
|
|
assert context.retry_succeeded, (
|
|
f"Async function failed: {getattr(context, 'retry_error', 'Unknown error')}"
|
|
)
|
|
assert context.result == "async success"
|
|
|
|
|
|
@then("the function should be called {expected_calls:d} times")
|
|
def step_verify_call_count(context, expected_calls):
|
|
"""Verify function was called expected number of times."""
|
|
assert context.call_count == expected_calls, (
|
|
f"Expected {expected_calls} calls, got {context.call_count}"
|
|
)
|
|
|
|
|
|
# Network retry pattern tests
|
|
@given("I have a function that raises NetworkError twice")
|
|
def step_create_network_error_function(context):
|
|
"""Create function that raises NetworkError."""
|
|
context.call_count = 0
|
|
|
|
def network_function():
|
|
context.call_count += 1
|
|
if context.call_count <= 2:
|
|
raise NetworkError(f"Network failure {context.call_count}")
|
|
return "network success"
|
|
|
|
context.test_function = network_function
|
|
|
|
|
|
@when("I apply network retry pattern")
|
|
def step_apply_network_retry(context):
|
|
"""Apply network-specific retry pattern."""
|
|
decorated_func = retry_network_operation(context.test_function)
|
|
|
|
try:
|
|
context.result = decorated_func()
|
|
context.retry_succeeded = True
|
|
except Exception as e:
|
|
context.retry_error = e
|
|
context.retry_succeeded = False
|
|
|
|
|
|
@then("the function should retry on NetworkError")
|
|
def step_verify_network_retry(context):
|
|
"""Verify function retried on NetworkError."""
|
|
assert context.retry_succeeded, "Network retry should have succeeded"
|
|
assert context.result == "network success"
|
|
|
|
|
|
@then("the function should be called {max_calls:d} times maximum")
|
|
def step_verify_max_calls(context, max_calls):
|
|
"""Verify function wasn't called more than max times."""
|
|
assert context.call_count <= max_calls, (
|
|
f"Function called {context.call_count} times, max is {max_calls}"
|
|
)
|
|
|
|
|
|
# Provider retry pattern tests
|
|
@given("I have a function that raises RateLimitError")
|
|
def step_create_rate_limit_function(context):
|
|
"""Create function that raises RateLimitError."""
|
|
context.call_count = 0
|
|
|
|
def provider_function():
|
|
context.call_count += 1
|
|
if context.call_count <= 2:
|
|
raise RateLimitError("Rate limit exceeded")
|
|
return "provider success"
|
|
|
|
context.test_function = provider_function
|
|
|
|
|
|
@when("I apply provider retry pattern")
|
|
def step_apply_provider_retry(context):
|
|
"""Apply provider-specific retry pattern."""
|
|
decorated_func = retry_provider_operation(context.test_function)
|
|
|
|
try:
|
|
context.result = decorated_func()
|
|
context.retry_succeeded = True
|
|
except Exception as e:
|
|
context.retry_error = e
|
|
context.retry_succeeded = False
|
|
|
|
|
|
@then("the function should retry with exponential jitter")
|
|
def step_verify_exponential_jitter(context):
|
|
"""Verify function used exponential backoff with jitter."""
|
|
assert context.retry_succeeded, "Provider retry should have succeeded"
|
|
assert context.result == "provider success"
|
|
# The jitter is implemented internally by tenacity
|
|
|
|
|
|
@then("the function should respect rate limit backoff")
|
|
def step_verify_rate_limit_backoff(context):
|
|
"""Verify rate limit backoff was respected."""
|
|
# Rate limit backoff is handled by the wait strategy
|
|
assert context.call_count >= 2, "Should have retried at least once"
|
|
|
|
|
|
# File operation retry pattern tests
|
|
@given("I have a file operation that fails with OSError once")
|
|
def step_create_file_operation(context):
|
|
"""Create a file operation that fails once before succeeding."""
|
|
context.file_call_count = 0
|
|
|
|
def file_operation():
|
|
context.file_call_count += 1
|
|
if context.file_call_count == 1:
|
|
raise OSError("transient file failure")
|
|
return "file success"
|
|
|
|
context.file_operation = file_operation
|
|
|
|
|
|
@when("I apply file retry pattern")
|
|
def step_apply_file_retry(context):
|
|
"""Apply the file retry decorator to the operation."""
|
|
decorated_func = retry_file_operation(context.file_operation)
|
|
|
|
try:
|
|
context.file_result = decorated_func()
|
|
context.file_retry_succeeded = True
|
|
except Exception as exc: # pragma: no cover - defensive branch
|
|
context.file_retry_error = exc
|
|
context.file_retry_succeeded = False
|
|
|
|
|
|
@then("the file operation should eventually succeed")
|
|
def step_verify_file_retry_success(context):
|
|
"""Verify the file operation eventually succeeds."""
|
|
assert context.file_retry_succeeded, (
|
|
f"File retry failed: {getattr(context, 'file_retry_error', 'Unknown error')}"
|
|
)
|
|
assert context.file_result == "file success"
|
|
|
|
|
|
@then("the file operation should be invoked {expected:d} times")
|
|
def step_verify_file_retry_invocations(context, expected):
|
|
"""Verify the file operation was invoked the expected number of times."""
|
|
assert context.file_call_count == expected, (
|
|
f"Expected {expected} calls, got {context.file_call_count}"
|
|
)
|
|
|
|
|
|
# Circuit breaker tests
|
|
@given("I have a circuit breaker with threshold {threshold:d}")
|
|
def step_create_circuit_breaker(context, threshold):
|
|
"""Create a circuit breaker with specified threshold."""
|
|
context.circuit_breaker = CircuitBreaker(
|
|
failure_threshold=threshold,
|
|
recovery_timeout=0.5, # Short timeout for testing
|
|
expected_exception=Exception,
|
|
)
|
|
context.failure_count = 0
|
|
|
|
|
|
@when("a function fails {failures:d} times")
|
|
def step_function_fails_multiple(context, failures):
|
|
"""Simulate function failures."""
|
|
|
|
def failing_function():
|
|
raise Exception("Test failure")
|
|
|
|
for _ in range(failures):
|
|
try:
|
|
context.circuit_breaker.call(failing_function)
|
|
except Exception:
|
|
context.failure_count += 1
|
|
|
|
|
|
@then("the circuit breaker should open")
|
|
def step_verify_circuit_open(context):
|
|
"""Verify circuit breaker is open."""
|
|
assert context.circuit_breaker.state == "open"
|
|
|
|
|
|
@then("subsequent calls should fail immediately")
|
|
def step_verify_immediate_failure(context):
|
|
"""Verify calls fail immediately when circuit is open."""
|
|
|
|
def test_function():
|
|
return "should not execute"
|
|
|
|
try:
|
|
context.circuit_breaker.call(test_function)
|
|
# If we get here, the circuit breaker didn't raise an exception
|
|
raise AssertionError(
|
|
"Circuit breaker should have raised CircuitBreakerOpen exception"
|
|
)
|
|
except CircuitBreakerOpen:
|
|
# This is the expected behavior - circuit is open
|
|
pass
|
|
except Exception as e:
|
|
# Unexpected exception
|
|
raise AssertionError(
|
|
f"Expected CircuitBreakerOpen but got: {type(e).__name__}: {e}"
|
|
) from e
|
|
|
|
|
|
@given("I have an open circuit breaker")
|
|
def step_create_open_circuit_breaker(context):
|
|
"""Create a circuit breaker in open state."""
|
|
context.circuit_breaker = CircuitBreaker(
|
|
failure_threshold=1, recovery_timeout=0.1, expected_exception=Exception
|
|
)
|
|
# Force circuit to open
|
|
with contextlib.suppress(builtins.BaseException):
|
|
context.circuit_breaker.call(lambda: 1 / 0)
|
|
assert context.circuit_breaker.state == "open"
|
|
|
|
|
|
@when("the recovery timeout expires")
|
|
def step_wait_recovery_timeout(context):
|
|
"""Wait for recovery timeout.
|
|
|
|
Uses ``time._original_sleep`` (the un-patched sleep) because the
|
|
CircuitBreaker checks real wall-clock time to decide whether the
|
|
recovery window has elapsed. The global sleep cap installed by
|
|
``environment.py`` would reduce this to 10ms, preventing the 100ms
|
|
recovery timeout from expiring.
|
|
"""
|
|
_real_sleep = getattr(time, "_original_sleep", time.sleep)
|
|
_real_sleep(0.2) # Wait longer than recovery timeout (0.1s)
|
|
|
|
|
|
@then("the circuit breaker should enter half-open state")
|
|
def step_verify_half_open_state(context):
|
|
"""Verify circuit breaker is in half-open state."""
|
|
|
|
# Attempting a call should transition to half-open
|
|
def success_function():
|
|
return "success"
|
|
|
|
try:
|
|
context.circuit_breaker.call(success_function)
|
|
context.call_succeeded = True
|
|
except CircuitBreakerOpen:
|
|
context.call_succeeded = False
|
|
|
|
|
|
@then("successful calls should close the circuit")
|
|
def step_verify_circuit_closed(context):
|
|
"""Verify successful calls close the circuit."""
|
|
assert context.call_succeeded, "Call should have succeeded"
|
|
|
|
# After one successful call, circuit should be in half-open state
|
|
# Need a second successful call to close it
|
|
def success_function():
|
|
return "success"
|
|
|
|
result = context.circuit_breaker.call(success_function)
|
|
assert result == "success", "Second call should succeed"
|
|
assert context.circuit_breaker.state == "closed"
|
|
|
|
|
|
@given("the circuit breaker has recorded failures")
|
|
def step_circuit_breaker_recorded_failures(context):
|
|
"""Simulate prior failures to exercise closed-state success path."""
|
|
context.circuit_breaker.failure_count = 3
|
|
context.circuit_breaker.state = "closed"
|
|
context.circuit_breaker.last_failure_time = time.time()
|
|
|
|
|
|
@when("I execute a successful call while the circuit is closed")
|
|
def step_success_call_closed_state(context):
|
|
"""Execute a successful call without half-open transition."""
|
|
context.closed_state_result = context.circuit_breaker.call(lambda: "closed success")
|
|
|
|
|
|
@then("the circuit breaker should reset failure tracking after success")
|
|
def step_verify_closed_state_success(context):
|
|
"""Ensure closed-state success resets counters."""
|
|
assert context.closed_state_result == "closed success"
|
|
assert context.circuit_breaker.failure_count == 0
|
|
assert context.circuit_breaker.state == "closed"
|
|
|
|
|
|
# Retry context tests
|
|
@given('I have a retry context for "{operation}"')
|
|
def step_create_retry_context(context, operation):
|
|
"""Create a retry context for an operation."""
|
|
context.retry_context = RetryContext(operation_name=operation, max_attempts=3)
|
|
|
|
|
|
@when("I execute a function that fails twice")
|
|
def step_execute_with_retry_context(context):
|
|
"""Execute function with retry context."""
|
|
context.execution_count = 0
|
|
context.recorded_errors = []
|
|
|
|
def failing_function():
|
|
context.execution_count += 1
|
|
if context.execution_count <= 2:
|
|
error = Exception(f"Failure {context.execution_count}")
|
|
context.recorded_errors.append(error)
|
|
raise error
|
|
return "context success"
|
|
|
|
try:
|
|
context.result = context.retry_context.execute(failing_function)
|
|
context.execution_succeeded = True
|
|
except Exception as e:
|
|
context.execution_error = e
|
|
context.execution_succeeded = False
|
|
|
|
|
|
@then("the retry context should track {attempts:d} attempts")
|
|
def step_verify_retry_context_attempts(context, attempts):
|
|
"""Verify retry context tracked attempts."""
|
|
assert context.retry_context.attempt_count == attempts
|
|
|
|
|
|
@then("the errors should be recorded")
|
|
def step_verify_errors_recorded(context):
|
|
"""Verify errors were recorded in retry context."""
|
|
# Note: The current implementation doesn't fully track errors in execute()
|
|
# but the context manager pattern supports it
|
|
assert len(context.recorded_errors) >= 2
|
|
|
|
|
|
@when("I execute a failing block under the retry context manager")
|
|
def step_retry_context_manager_failure(context):
|
|
"""Execute a failing block under the retry context manager."""
|
|
errors_before = len(context.retry_context.errors)
|
|
try:
|
|
with context.retry_context:
|
|
raise RuntimeError("context manager failure")
|
|
except RuntimeError:
|
|
context.retry_context_manager_error = True
|
|
else: # pragma: no cover - defensive branch
|
|
context.retry_context_manager_error = False
|
|
context.retry_context_errors_added_sync = (
|
|
len(context.retry_context.errors) - errors_before
|
|
)
|
|
|
|
|
|
@then("the retry context manager should capture the exception")
|
|
def step_verify_retry_context_manager_capture(context):
|
|
"""Ensure the retry context manager captured the exception."""
|
|
assert context.retry_context_manager_error, "Expected runtime error to propagate"
|
|
assert context.retry_context_errors_added_sync >= 1
|
|
assert isinstance(context.retry_context.errors[-1], Exception)
|
|
|
|
|
|
@when("I execute a failing block under the async retry context manager")
|
|
def step_async_retry_context_manager_failure(context):
|
|
"""Execute a failing block under the async retry context manager."""
|
|
errors_before = len(context.retry_context.errors)
|
|
|
|
async def run_async_block():
|
|
try:
|
|
async with context.retry_context:
|
|
raise RuntimeError("async context failure")
|
|
except RuntimeError:
|
|
return True
|
|
return False
|
|
|
|
loop = asyncio.new_event_loop()
|
|
try:
|
|
context.async_retry_context_manager_error = loop.run_until_complete(
|
|
run_async_block()
|
|
)
|
|
finally:
|
|
loop.close()
|
|
|
|
context.retry_context_errors_added_async = (
|
|
len(context.retry_context.errors) - errors_before
|
|
)
|
|
|
|
|
|
@then("the async retry context manager should capture the exception")
|
|
def step_verify_async_retry_context_manager_capture(context):
|
|
"""Ensure the async retry context manager captured the exception."""
|
|
assert context.async_retry_context_manager_error, (
|
|
"Expected async runtime error to propagate"
|
|
)
|
|
assert context.retry_context_errors_added_async >= 1
|
|
assert isinstance(context.retry_context.errors[-1], Exception)
|
|
|
|
|
|
@when("I execute a successful block under the retry context manager")
|
|
def step_retry_context_manager_success(context):
|
|
"""Execute a successful block to cover no-error path."""
|
|
errors_before = len(context.retry_context.errors)
|
|
with context.retry_context:
|
|
context.retry_context_success_value = "sync context success"
|
|
context.retry_context_errors_delta_sync = (
|
|
len(context.retry_context.errors) - errors_before
|
|
)
|
|
|
|
|
|
@then("the retry context manager should not record errors")
|
|
def step_verify_retry_context_manager_success(context):
|
|
"""Confirm no errors were recorded for successful context block."""
|
|
assert context.retry_context_success_value == "sync context success"
|
|
assert context.retry_context_errors_delta_sync == 0
|
|
|
|
|
|
@when("I execute a successful block under the async retry context manager")
|
|
def step_async_retry_context_manager_success(context):
|
|
"""Execute a successful async block to cover no-error path."""
|
|
errors_before = len(context.retry_context.errors)
|
|
|
|
async def run_async_block():
|
|
async with context.retry_context:
|
|
return "async context success"
|
|
|
|
loop = asyncio.new_event_loop()
|
|
try:
|
|
context.async_retry_context_manager_result = loop.run_until_complete(
|
|
run_async_block()
|
|
)
|
|
finally:
|
|
loop.close()
|
|
|
|
context.retry_context_errors_delta_async = (
|
|
len(context.retry_context.errors) - errors_before
|
|
)
|
|
|
|
|
|
@then("the async retry context manager should not record errors")
|
|
def step_verify_async_retry_context_manager_success(context):
|
|
"""Confirm async context manager leaves errors untouched when successful."""
|
|
assert context.async_retry_context_manager_result == "async context success"
|
|
assert context.retry_context_errors_delta_async == 0
|
|
|
|
|
|
@when("I execute the async function with the retry context")
|
|
def step_execute_async_with_retry_context(context):
|
|
"""Execute async function using RetryContext.async_execute."""
|
|
loop = asyncio.new_event_loop()
|
|
try:
|
|
context.async_execute_result = loop.run_until_complete(
|
|
context.retry_context.async_execute(context.test_function)
|
|
)
|
|
finally:
|
|
loop.close()
|
|
context.async_execute_attempts = context.retry_context.attempt_count
|
|
|
|
|
|
@then("the async retry context execute should succeed after {attempts:d} attempts")
|
|
def step_verify_async_execute_attempts(context, attempts):
|
|
"""Verify async execute succeeded after expected attempts."""
|
|
assert context.async_execute_result == "async success"
|
|
assert context.async_execute_attempts == attempts
|
|
|
|
|
|
@when("I apply retry with timeout of {max_attempts:d} attempts and {timeout} seconds")
|
|
def step_apply_retry_with_timeout(context, max_attempts, timeout):
|
|
"""Apply retry_with_timeout while stubbing out sleep and monotonic clock."""
|
|
timeout_seconds = float(timeout)
|
|
original_sleep = time.sleep
|
|
original_monotonic = time.monotonic
|
|
|
|
_fake_base = original_monotonic()
|
|
_call_counter = [0]
|
|
|
|
def _fake_monotonic():
|
|
_call_counter[0] += 1
|
|
return _fake_base + _call_counter[0] * 1e-6
|
|
|
|
time.sleep = lambda _seconds: None
|
|
time.monotonic = _fake_monotonic
|
|
|
|
def restore_sleep():
|
|
time.sleep = original_sleep
|
|
time.monotonic = original_monotonic
|
|
|
|
# Register cleanup first so it runs even if the step body fails.
|
|
context.add_cleanup(restore_sleep)
|
|
|
|
try:
|
|
decorated_func = retry_with_timeout(
|
|
max_attempts=max_attempts, timeout_seconds=timeout_seconds
|
|
)(context.test_function)
|
|
|
|
try:
|
|
context.result = decorated_func()
|
|
context.retry_succeeded = True
|
|
except Exception as exc: # pragma: no cover - defensive safety
|
|
context.retry_error = exc
|
|
context.retry_succeeded = False
|
|
finally:
|
|
# Eagerly restore time.sleep so subsequent scenarios in the same
|
|
# feature cannot observe the patched no-op — even if behave's
|
|
# add_cleanup fires slightly after scenario-boundary hooks.
|
|
restore_sleep()
|
|
|
|
|
|
# Retry with jitter tests
|
|
@given("I have multiple concurrent operations")
|
|
def step_setup_concurrent_operations(context):
|
|
"""Setup multiple operations for concurrent testing."""
|
|
context.operation_times = []
|
|
context.operation_count = 5
|
|
|
|
|
|
@when("I apply retry with jitter")
|
|
def step_apply_retry_with_jitter(context):
|
|
"""Apply retry with jitter to multiple operations."""
|
|
|
|
@retry_with_jitter(max_attempts=3, base_delay=0.01, max_delay=0.1)
|
|
def operation_with_jitter(op_id):
|
|
# First attempt always fails
|
|
if len([t for t in context.operation_times if t[0] == op_id]) == 0:
|
|
context.operation_times.append((op_id, time.monotonic_ns()))
|
|
raise Exception(f"Operation {op_id} failed")
|
|
return f"Operation {op_id} success"
|
|
|
|
# Execute operations
|
|
for i in range(context.operation_count):
|
|
with contextlib.suppress(builtins.BaseException):
|
|
operation_with_jitter(i)
|
|
|
|
|
|
@then("the retries should have random delays")
|
|
def step_verify_random_delays(context):
|
|
"""Verify retries have random delays (jitter)."""
|
|
# The jitter is applied internally by tenacity
|
|
# We can verify that operations were attempted
|
|
assert len(context.operation_times) > 0
|
|
|
|
|
|
@then("operations should not retry simultaneously")
|
|
def step_verify_no_simultaneous_retry(context):
|
|
"""Verify operations don't all share the exact same timestamp.
|
|
|
|
We use monotonic_ns() for high-resolution timestamps and count the
|
|
number of *distinct* timestamps. On every platform (even fast CI
|
|
runners with ``time.sleep`` patched to a no-op in a preceding
|
|
scenario) the sequential loop overhead ensures at least two distinct
|
|
nanosecond readings among the five operations, so we only assert
|
|
``>= 2``. The real jitter behaviour is validated by tenacity's own
|
|
test-suite; this scenario confirms the decorator wires up correctly.
|
|
"""
|
|
unique_times = len(set(t for _, t in context.operation_times))
|
|
assert unique_times >= 2, (
|
|
f"Expected at least 2 distinct timestamps, got {unique_times}"
|
|
)
|
|
|
|
|
|
# Retry on result tests
|
|
@given("I have sequential result payloads requiring a retry")
|
|
def step_setup_sequential_retry_results(context):
|
|
"""Create a deterministic sequence of results for retry_on_result tests."""
|
|
context.call_count = 0
|
|
context.result_sequence = [
|
|
{"retry": True, "value": "try-again"},
|
|
{"retry": False, "value": "done"},
|
|
]
|
|
|
|
def result_function():
|
|
context.call_count += 1
|
|
index = min(context.call_count - 1, len(context.result_sequence) - 1)
|
|
return context.result_sequence[index]
|
|
|
|
context.test_function = result_function
|
|
|
|
|
|
@given("I have a retry predicate that checks for retry flag")
|
|
def step_setup_retry_predicate(context):
|
|
"""Define predicate used by retry_on_result decorator."""
|
|
|
|
def predicate(result):
|
|
return isinstance(result, dict) and bool(result.get("retry"))
|
|
|
|
context.retry_predicate = predicate
|
|
|
|
|
|
@when("I apply retry on result with max {max_attempts:d} attempts")
|
|
def step_apply_retry_on_result(context, max_attempts):
|
|
"""Apply retry_on_result decorator and capture the outcome."""
|
|
decorated_func = retry_on_result(
|
|
context.retry_predicate, max_attempts=max_attempts
|
|
)(context.test_function)
|
|
|
|
try:
|
|
context.retry_on_result_output = decorated_func()
|
|
context.retry_on_result_succeeded = True
|
|
except Exception as exc: # pragma: no cover - defensive safety
|
|
context.retry_on_result_output = exc
|
|
context.retry_on_result_succeeded = False
|
|
|
|
|
|
@then("the decorated function should return the successful payload")
|
|
def step_verify_retry_on_result_payload(context):
|
|
"""Ensure retry_on_result eventually returns the non-retry payload."""
|
|
assert context.retry_on_result_succeeded, (
|
|
f"Retry on result failed: {context.retry_on_result_output}"
|
|
)
|
|
assert isinstance(context.retry_on_result_output, dict)
|
|
assert context.retry_on_result_output.get("retry") is False
|
|
|
|
|
|
# Auto-debug retry tests
|
|
@given("I have a function that fails with errors")
|
|
def step_create_failing_function_with_errors(context):
|
|
"""Create a function that fails with specific errors."""
|
|
context.failure_count = 0
|
|
|
|
async def failing_func():
|
|
context.failure_count += 1
|
|
if context.failure_count <= 2:
|
|
raise ValueError(f"Error {context.failure_count}")
|
|
return {"success": True, "message": "Fixed after debug"}
|
|
|
|
context.test_function = failing_func
|
|
|
|
|
|
@given("I have a debug callback that fixes issues")
|
|
def step_create_debug_callback(context):
|
|
"""Create a debug callback function."""
|
|
|
|
async def debug_callback(error, attempt):
|
|
# Simulate fixing the issue
|
|
if attempt >= 1:
|
|
return {"fixed": True, "message": f"Fixed error: {error}"}
|
|
return {"fixed": False}
|
|
|
|
context.debug_callback = debug_callback
|
|
|
|
|
|
@when("I apply auto-debug retry")
|
|
def step_apply_auto_debug_retry(context):
|
|
"""Apply auto-debug retry pattern."""
|
|
decorated_func = retry_auto_debug(
|
|
max_debug_attempts=3, debug_callback=context.debug_callback
|
|
)(context.test_function)
|
|
|
|
async def run_auto_debug():
|
|
try:
|
|
result = await decorated_func()
|
|
return result, True
|
|
except Exception as e:
|
|
return e, False
|
|
|
|
loop = asyncio.new_event_loop()
|
|
result, succeeded = loop.run_until_complete(run_auto_debug())
|
|
loop.close()
|
|
|
|
context.debug_result = result
|
|
context.debug_succeeded = succeeded
|
|
|
|
|
|
@then("the function should retry with debug fixes")
|
|
def step_verify_debug_retry(context):
|
|
"""Verify function retried with debug fixes."""
|
|
assert context.failure_count > 1, "Should have failed at least once"
|
|
|
|
|
|
@then("eventually succeed after debugging")
|
|
def step_verify_debug_success(context):
|
|
"""Verify function succeeded after debugging."""
|
|
assert context.debug_succeeded, f"Auto-debug failed: {context.debug_result}"
|
|
assert context.debug_result["success"] is True
|
|
|
|
|
|
# Retry categories tests
|
|
@given("I have the retry categories defined")
|
|
def step_verify_retry_categories(context):
|
|
"""Verify retry categories are defined."""
|
|
context.categories = RETRY_CATEGORIES
|
|
assert "network" in context.categories
|
|
assert "provider" in context.categories
|
|
assert "database" in context.categories
|
|
assert "file_operation" in context.categories
|
|
|
|
|
|
@then("network operations should retry {expected:d} times")
|
|
def step_verify_network_retry_count(context, expected):
|
|
"""Verify network retry configuration."""
|
|
assert context.categories["network"]["max_attempts"] == expected
|
|
|
|
|
|
@then("provider operations should retry {expected:d} times")
|
|
def step_verify_provider_retry_count(context, expected):
|
|
"""Verify provider retry configuration."""
|
|
assert context.categories["provider"]["max_attempts"] == expected
|
|
|
|
|
|
@then("database operations should retry {expected:d} times")
|
|
def step_verify_database_retry_count(context, expected):
|
|
"""Verify database retry configuration."""
|
|
assert context.categories["database"]["max_attempts"] == expected
|
|
|
|
|
|
@then("file operations should retry {expected:d} times")
|
|
def step_verify_file_retry_count(context, expected):
|
|
"""Verify file operation retry configuration."""
|
|
assert context.categories["file_operation"]["max_attempts"] == expected
|
|
|
|
|
|
@given("the retry logger rejects keyword arguments")
|
|
def step_retry_logger_rejects_keyword_arguments(context):
|
|
"""Configure a logger that rejects keyword arguments."""
|
|
context.logged_messages = []
|
|
|
|
class KeywordRejectingLogger:
|
|
def __init__(self, messages):
|
|
self.messages = messages
|
|
|
|
def debug(self, message, *args, **kwargs):
|
|
if kwargs:
|
|
raise TypeError("Keyword arguments not supported")
|
|
self.messages.append(message)
|
|
|
|
def info(self, *args, **kwargs):
|
|
return None
|
|
|
|
def warning(self, *args, **kwargs):
|
|
return None
|
|
|
|
def error(self, *args, **kwargs):
|
|
return None
|
|
|
|
context.original_logger = retry_patterns_module.logger
|
|
retry_patterns_module.logger = KeywordRejectingLogger(context.logged_messages)
|
|
|
|
def cleanup_logger():
|
|
retry_patterns_module.logger = context.original_logger
|
|
|
|
context.add_cleanup(cleanup_logger)
|
|
|
|
|
|
@when("I invoke logging callbacks for failed and successful attempts")
|
|
def step_invoke_logging_callbacks(context):
|
|
"""Trigger logging callbacks to exercise fallback paths."""
|
|
failure_state = SimpleNamespace(
|
|
attempt_number=1,
|
|
outcome=SimpleNamespace(
|
|
failed=True,
|
|
exception=lambda: ValueError("simulated failure"),
|
|
),
|
|
next_action=SimpleNamespace(sleep=0.5),
|
|
)
|
|
success_state = SimpleNamespace(
|
|
attempt_number=2,
|
|
outcome=SimpleNamespace(failed=False),
|
|
next_action=None,
|
|
)
|
|
|
|
log_before_retry(failure_state)
|
|
log_after_retry(failure_state)
|
|
log_after_retry(success_state)
|
|
|
|
|
|
@then("the fallback logging messages should be recorded")
|
|
def step_verify_fallback_logging_messages(context):
|
|
"""Verify fallback logging produced the expected messages."""
|
|
expected_messages = {
|
|
"Starting attempt 1",
|
|
"Attempt 1 failed, will retry",
|
|
"Attempt 2 succeeded",
|
|
}
|
|
recorded = set(context.logged_messages)
|
|
missing = expected_messages - recorded
|
|
assert not missing, f"Missing fallback messages: {missing}"
|
|
|
|
|
|
@given("I have a half-open circuit breaker with threshold {threshold:d}")
|
|
def step_half_open_circuit_breaker(context, threshold):
|
|
"""Prepare a circuit breaker in half-open state."""
|
|
context.circuit_breaker = CircuitBreaker(
|
|
failure_threshold=threshold,
|
|
recovery_timeout=0.1,
|
|
expected_exception=Exception,
|
|
)
|
|
context.circuit_breaker.state = "half-open"
|
|
context.circuit_breaker.success_count_half_open = 0
|
|
context.circuit_breaker.failure_count = 0
|
|
|
|
|
|
@when("I record a successful half-open call")
|
|
def step_record_half_open_success(context):
|
|
"""Record the first successful call while half-open."""
|
|
context.last_half_open_result = context.circuit_breaker.call(
|
|
lambda: "half-open success"
|
|
)
|
|
|
|
|
|
@when("I record another successful half-open call")
|
|
def step_record_second_half_open_success(context):
|
|
"""Record the second successful call while half-open."""
|
|
context.second_half_open_result = context.circuit_breaker.call(
|
|
lambda: "half-open success"
|
|
)
|
|
|
|
|
|
@then("the circuit breaker should close after consecutive successes")
|
|
def step_verify_circuit_closes_after_successes(context):
|
|
"""Ensure the circuit closes after consecutive successes."""
|
|
assert context.circuit_breaker.state == "closed", "Circuit should be closed"
|
|
assert context.circuit_breaker.success_count_half_open == 0
|
|
assert context.second_half_open_result == "half-open success"
|
|
|
|
|
|
@when("I simulate a half-open failure")
|
|
def step_simulate_half_open_failure(context):
|
|
"""Simulate a failure while half-open to reopen the circuit."""
|
|
context.circuit_breaker.state = "half-open"
|
|
context.circuit_breaker.success_count_half_open = 0
|
|
|
|
def failing_call():
|
|
raise Exception("half-open failure")
|
|
|
|
try:
|
|
context.circuit_breaker.call(failing_call)
|
|
except Exception:
|
|
context.half_open_failure_caught = True
|
|
else:
|
|
context.half_open_failure_caught = False
|
|
|
|
|
|
@then("the circuit breaker should reopen after half-open failure")
|
|
def step_verify_circuit_reopens_after_failure(context):
|
|
"""Verify the circuit reopens after a half-open failure."""
|
|
assert context.half_open_failure_caught, "Failure should have been raised"
|
|
assert context.circuit_breaker.state == "open"
|
|
assert context.circuit_breaker.success_count_half_open == 0
|
|
|
|
|
|
@given("I have an async circuit breaker with threshold {threshold:d}")
|
|
def step_create_async_circuit_breaker(context, threshold):
|
|
"""Create a circuit breaker for async operations."""
|
|
context.circuit_breaker = CircuitBreaker(
|
|
failure_threshold=threshold,
|
|
recovery_timeout=0.1, # Use shorter timeout — the sleep step waits 0.2s
|
|
expected_exception=Exception,
|
|
)
|
|
context.async_loop = asyncio.new_event_loop()
|
|
context.add_cleanup(context.async_loop.close)
|
|
|
|
|
|
@when("I trigger an async failure on the circuit breaker")
|
|
def step_trigger_async_failure(context):
|
|
"""Trigger an async failure to open the circuit."""
|
|
|
|
async def failing_call():
|
|
raise Exception("async breaker failure")
|
|
|
|
async def run_failure():
|
|
try:
|
|
await context.circuit_breaker.async_call(failing_call)
|
|
except Exception:
|
|
return True
|
|
return False
|
|
|
|
context.async_failure_caught = context.async_loop.run_until_complete(run_failure())
|
|
assert context.async_failure_caught, "Async failure should have been caught"
|
|
assert context.circuit_breaker.state == "open"
|
|
|
|
|
|
@then("the async circuit breaker should raise immediately while open")
|
|
def step_verify_async_open_state(context):
|
|
"""Ensure async calls are blocked while circuit remains open."""
|
|
|
|
async def success_call():
|
|
return "async success"
|
|
|
|
async def run_open_call():
|
|
try:
|
|
await context.circuit_breaker.async_call(success_call)
|
|
except CircuitBreakerOpen:
|
|
return True
|
|
return False
|
|
|
|
context.async_open_blocked = context.async_loop.run_until_complete(run_open_call())
|
|
assert context.async_open_blocked, "Circuit breaker should prevent async call"
|
|
|
|
|
|
@when("I perform two async successful calls after recovery")
|
|
def step_async_success_calls(context):
|
|
"""Execute two successful async calls after recovery."""
|
|
|
|
async def success_call():
|
|
return "async success"
|
|
|
|
async def run_success_calls():
|
|
first = await context.circuit_breaker.async_call(success_call)
|
|
second = await context.circuit_breaker.async_call(success_call)
|
|
return first, second
|
|
|
|
context.async_success_results = context.async_loop.run_until_complete(
|
|
run_success_calls()
|
|
)
|
|
|
|
|
|
@then("the async circuit breaker should close after async successes")
|
|
def step_verify_async_circuit_closed(context):
|
|
"""Verify the circuit closes after successful async calls."""
|
|
assert context.async_success_results == ("async success", "async success")
|
|
assert context.circuit_breaker.state == "closed"
|
|
|
|
|
|
@given('I have a response payload with error "{message}"')
|
|
def step_response_payload_with_error(context, message):
|
|
"""Create a response payload containing an error message."""
|
|
context.result_payload = {"error": message}
|
|
|
|
|
|
@given("I have a response payload with status code {status:d}")
|
|
def step_response_payload_with_status(context, status):
|
|
"""Create a response payload containing a status code."""
|
|
context.result_payload = {"status_code": status}
|
|
|
|
|
|
@given("I have a result payload set to None")
|
|
def step_result_payload_none(context):
|
|
"""Set the result payload to None."""
|
|
context.result_payload = None
|
|
|
|
|
|
@given('I have a result payload set to the string "{value}"')
|
|
def step_result_payload_string(context, value):
|
|
"""Set the result payload to a literal string value."""
|
|
context.result_payload = value
|
|
|
|
|
|
@when("I evaluate the retry decision")
|
|
def step_evaluate_retry_decision(context):
|
|
"""Evaluate whether the result should trigger a retry."""
|
|
context.retry_decision = should_retry_result(context.result_payload)
|
|
|
|
|
|
@then("the retry decision should be True")
|
|
def step_verify_retry_decision_true(context):
|
|
"""Confirm the retry decision is positive."""
|
|
assert context.retry_decision is True
|
|
|
|
|
|
@then("the retry decision should be False")
|
|
def step_verify_retry_decision_false(context):
|
|
"""Confirm the retry decision is negative."""
|
|
assert context.retry_decision is False
|
|
|
|
|
|
@when('I request a retry decorator for "{category}"')
|
|
def step_request_retry_decorator(context, category):
|
|
"""Request a retry decorator for the given category."""
|
|
context.retry_decorator = get_retry_decorator(category)
|
|
context.decorated_value = None
|
|
|
|
|
|
@when('I decorate a function returning "{value}"')
|
|
def step_decorate_function_with_retry(context, value):
|
|
"""Decorate a function and execute it to capture the result."""
|
|
|
|
def sample_function():
|
|
return value
|
|
|
|
decorated_function = context.retry_decorator(sample_function)
|
|
context.decorated_value = value
|
|
context.decorated_result = decorated_function()
|
|
|
|
|
|
@then("the decorated function should execute successfully")
|
|
def step_verify_decorated_function_execution(context):
|
|
"""Ensure the decorated function returns the expected value."""
|
|
assert context.decorated_result == context.decorated_value
|
|
|
|
|
|
@when("I configure retry logging to {level_name} level")
|
|
def step_configure_retry_logging_level(context, level_name):
|
|
"""Configure retry logging and capture existing levels for restoration."""
|
|
level_attr = level_name.upper()
|
|
assert hasattr(logging, level_attr), f"Unknown log level: {level_name}"
|
|
level_value = getattr(logging, level_attr)
|
|
|
|
before_logger = logging.getLogger("tenacity.before")
|
|
after_logger = logging.getLogger("tenacity.after")
|
|
context.original_tenacity_levels = (before_logger.level, after_logger.level)
|
|
|
|
def restore_log_levels(): # pragma: no cover - cleanup path
|
|
before_logger.setLevel(context.original_tenacity_levels[0])
|
|
after_logger.setLevel(context.original_tenacity_levels[1])
|
|
|
|
context.add_cleanup(restore_log_levels)
|
|
configure_retry_logging(level_value)
|
|
context.configured_log_level = level_value
|
|
|
|
|
|
@then("the tenacity loggers should be set to {level_name}")
|
|
def step_verify_tenacity_log_levels(context, level_name):
|
|
"""Verify tenacity loggers were set to the expected log level."""
|
|
expected_level = getattr(logging, level_name.upper())
|
|
before_level = logging.getLogger("tenacity.before").level
|
|
after_level = logging.getLogger("tenacity.after").level
|
|
assert before_level == expected_level
|
|
assert after_level == expected_level
|
|
assert context.configured_log_level == expected_level
|
|
|
|
|
|
@given("I have a function that returns persistent error payloads")
|
|
def step_function_with_persistent_errors(context):
|
|
"""Create a function that always returns error payloads."""
|
|
context.persistent_error_count = 0
|
|
|
|
async def persistent_error_function():
|
|
context.persistent_error_count += 1
|
|
return {"error": f"Persistent failure {context.persistent_error_count}"}
|
|
|
|
context.test_function = persistent_error_function
|
|
|
|
|
|
@given("I have a debug callback that never fixes issues")
|
|
def step_debug_callback_never_fixes(context):
|
|
"""Create a debug callback that never fixes the issue."""
|
|
|
|
async def never_fix(_error, _attempt):
|
|
return {"fixed": False}
|
|
|
|
context.debug_callback = never_fix
|
|
|
|
|
|
@given("I have a debug callback that raises errors")
|
|
def step_debug_callback_raises_errors(context):
|
|
"""Create a debug callback that raises when invoked."""
|
|
context.debug_callback_failure_count = 0
|
|
|
|
async def raising_callback(_error, _attempt):
|
|
context.debug_callback_failure_count += 1
|
|
raise RuntimeError("debug callback failure")
|
|
|
|
context.debug_callback = raising_callback
|
|
|
|
|
|
@when("I apply auto-debug retry expecting failure")
|
|
def step_apply_auto_debug_retry_failure(context):
|
|
"""Run auto-debug retry expecting it to raise the last error message."""
|
|
decorated_func = retry_auto_debug(
|
|
max_debug_attempts=3, debug_callback=context.debug_callback
|
|
)(context.test_function)
|
|
|
|
async def run_auto_debug():
|
|
try:
|
|
result = await decorated_func()
|
|
return result, True
|
|
except Exception as exc: # pragma: no cover - defensive branch
|
|
return exc, False
|
|
|
|
loop = asyncio.new_event_loop()
|
|
try:
|
|
result, succeeded = loop.run_until_complete(run_auto_debug())
|
|
finally:
|
|
loop.close()
|
|
|
|
context.debug_result = result
|
|
context.debug_succeeded = succeeded
|
|
context.expected_last_error = f"Persistent failure {context.persistent_error_count}"
|
|
|
|
|
|
@then("the auto-debug retry should raise the last error message")
|
|
def step_verify_auto_debug_raises_last_error(context):
|
|
"""Ensure the auto-debug retry raises the final error string."""
|
|
assert not context.debug_succeeded, "Auto-debug retry unexpectedly succeeded"
|
|
assert isinstance(context.debug_result, Exception)
|
|
assert str(context.debug_result) == context.expected_last_error
|
|
|
|
|
|
@then("the auto-debug retry should raise the debug callback failure")
|
|
def step_verify_auto_debug_raises_callback_failure(context):
|
|
"""Ensure the debug callback failure bubble is surfaced."""
|
|
assert not context.debug_succeeded, "Auto-debug retry unexpectedly succeeded"
|
|
assert isinstance(context.debug_result, Exception)
|
|
assert str(context.debug_result) == "debug callback failure"
|
|
|
|
|
|
@then("the debug callback failures should be counted")
|
|
def step_verify_debug_callback_failures(context):
|
|
"""Ensure the debug callback failure path was exercised."""
|
|
failure_count = getattr(context, "debug_callback_failure_count", 0)
|
|
assert failure_count >= 1, "Expected debug callback to fail at least once"
|