Spaces:
Sleeping
Sleeping
File size: 3,594 Bytes
69fb140 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 | """
測試 core/retry.py 重試機制
"""
import pytest
import asyncio
from core.retry import (
RetryConfig,
retry_async,
with_retry,
default_retry_config,
api_retry_config,
)
class TestRetryConfig:
"""測試重試配置"""
def test_default_config(self):
"""測試預設配置"""
config = RetryConfig()
assert config.max_retries == 3
assert config.base_delay == 1.0
assert config.max_delay == 30.0
def test_custom_config(self):
"""測試自訂配置"""
config = RetryConfig(max_retries=5, base_delay=0.5)
assert config.max_retries == 5
assert config.base_delay == 0.5
def test_calculate_delay(self):
"""測試延遲計算"""
config = RetryConfig(base_delay=1.0, jitter=False)
# 第一次重試
delay0 = config.calculate_delay(0)
assert delay0 == 1.0
# 第二次重試（指數退避）
delay1 = config.calculate_delay(1)
assert delay1 == 2.0
def test_max_delay_cap(self):
"""測試最大延遲上限"""
config = RetryConfig(base_delay=10.0, max_delay=15.0, jitter=False)
delay = config.calculate_delay(5) # 10 * 2^5 = 320，應被限制
assert delay == 15.0
class TestRetryAsync:
"""測試異步重試"""
@pytest.mark.asyncio
async def test_success_no_retry(self):
"""測試成功時不重試"""
call_count = 0
async def success_func():
nonlocal call_count
call_count += 1
return "success"
result = await retry_async(success_func)
assert result == "success"
assert call_count == 1
@pytest.mark.asyncio
async def test_retry_then_success(self):
"""測試重試後成功"""
call_count = 0
async def fail_then_success():
nonlocal call_count
call_count += 1
if call_count < 3:
raise ValueError("temporary error")
return "success"
config = RetryConfig(max_retries=3, base_delay=0.01)
result = await retry_async(fail_then_success, config=config)
assert result == "success"
assert call_count == 3
@pytest.mark.asyncio
async def test_all_retries_fail(self):
"""測試所有重試都失敗"""
call_count = 0
async def always_fail():
nonlocal call_count
call_count += 1
raise ValueError("always fails")
config = RetryConfig(max_retries=2, base_delay=0.01)
with pytest.raises(ValueError):
await retry_async(always_fail, config=config)
assert call_count == 3 # 初始 + 2 次重試
class TestWithRetryDecorator:
"""測試重試裝飾器"""
@pytest.mark.asyncio
async def test_decorator_success(self):
"""測試裝飾器成功"""
@with_retry(max_retries=2, base_delay=0.01)
async def decorated_func():
return "decorated"
result = await decorated_func()
assert result == "decorated"
class TestPredefinedConfigs:
"""測試預定義配置"""
def test_default_config_exists(self):
"""測試預設配置存在"""
assert default_retry_config is not None
assert default_retry_config.max_retries == 3
def test_api_config_exists(self):
"""測試 API 配置存在"""
assert api_retry_config is not None
assert api_retry_config.max_retries == 2
assert api_retry_config.base_delay == 0.5
|