starmen1265
c9ca1b538d
- Remove sys.path manipulation from kb_workflow.py - Update CLI imports to use kb_workflow.pipeline prefix - Update all test file imports and patch paths Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
136 lines
4.0 KiB
Python
136 lines
4.0 KiB
Python
"""Tests for retry mechanism module
|
|
|
|
Tests for retry_on_network_error decorator using tenacity.
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
import time
|
|
|
|
from kb_workflow.pipeline.errors import NetworkError, ValidationError, ErrorType
|
|
from kb_workflow.pipeline.retry import retry_on_network_error
|
|
|
|
|
|
class TestRetryOnNetworkError:
|
|
"""Tests for retry_on_network_error decorator"""
|
|
|
|
def test_decorator_retries_on_network_error(self):
|
|
"""Test that NetworkError triggers retry"""
|
|
call_count = 0
|
|
|
|
@retry_on_network_error
|
|
def failing_then_success():
|
|
nonlocal call_count
|
|
call_count += 1
|
|
if call_count < 3:
|
|
raise NetworkError(
|
|
step_name="test",
|
|
input_summary="test",
|
|
status_code=500
|
|
)
|
|
return "success"
|
|
|
|
result = failing_then_success()
|
|
assert result == "success"
|
|
assert call_count == 3
|
|
|
|
def test_decorator_retries_up_to_3_times(self):
|
|
"""Test that decorator retries max 3 times"""
|
|
call_count = 0
|
|
|
|
@retry_on_network_error
|
|
def always_fail():
|
|
nonlocal call_count
|
|
call_count += 1
|
|
raise NetworkError(
|
|
step_name="test",
|
|
input_summary="test",
|
|
status_code=500
|
|
)
|
|
|
|
with pytest.raises(NetworkError):
|
|
always_fail()
|
|
|
|
# 3 attempts total
|
|
assert call_count == 3
|
|
|
|
def test_decorator_does_not_retry_non_network_error(self):
|
|
"""Test that non-NetworkError is not retried"""
|
|
call_count = 0
|
|
|
|
@retry_on_network_error
|
|
def validation_fail():
|
|
nonlocal call_count
|
|
call_count += 1
|
|
raise ValidationError(
|
|
step_name="test",
|
|
input_summary="test"
|
|
)
|
|
|
|
with pytest.raises(ValidationError):
|
|
validation_fail()
|
|
|
|
# Only 1 attempt, no retry
|
|
assert call_count == 1
|
|
|
|
def test_decorator_reraises_after_max_attempts(self):
|
|
"""Test that original error is reraised after max attempts"""
|
|
@retry_on_network_error
|
|
def always_fail():
|
|
raise NetworkError(
|
|
step_name="test_step",
|
|
input_summary="url=http://example.com",
|
|
status_code=403
|
|
)
|
|
|
|
with pytest.raises(NetworkError) as exc_info:
|
|
always_fail()
|
|
|
|
assert exc_info.value.step_name == "test_step"
|
|
assert exc_info.value.status_code == 403
|
|
|
|
def test_decorator_preserves_return_value(self):
|
|
"""Test that successful return value is preserved"""
|
|
@retry_on_network_error
|
|
def successful_func():
|
|
return {"status": "ok", "data": [1, 2, 3]}
|
|
|
|
result = successful_func()
|
|
assert result == {"status": "ok", "data": [1, 2, 3]}
|
|
|
|
def test_decorator_uses_exponential_backoff(self):
|
|
"""Test that retry uses exponential backoff (2s, 4s max 10s)"""
|
|
# This test verifies the configuration, not exact timing
|
|
# The actual timing is handled by tenacity
|
|
from tenacity import stop_after_attempt, wait_exponential
|
|
|
|
# Verify the decorator is configured correctly
|
|
# by checking it's a retry wrapper
|
|
@retry_on_network_error
|
|
def dummy():
|
|
pass
|
|
|
|
# The function should be wrapped
|
|
assert hasattr(dummy, "__wrapped__") or callable(dummy)
|
|
|
|
|
|
class TestRetryWithCause:
|
|
"""Tests for retry with original exception as cause"""
|
|
|
|
def test_retry_preserves_cause(self):
|
|
"""Test that original exception is preserved as cause"""
|
|
original = ValueError("Connection refused")
|
|
|
|
@retry_on_network_error
|
|
def fail_with_cause():
|
|
raise NetworkError(
|
|
step_name="connect",
|
|
input_summary="host=localhost",
|
|
cause=original
|
|
)
|
|
|
|
with pytest.raises(NetworkError) as exc_info:
|
|
fail_with_cause()
|
|
|
|
assert exc_info.value.cause is original
|