88 lines
2.4 KiB
Python
88 lines
2.4 KiB
Python
"""``retry_on_retryable_error`` 装饰器：覆盖同步/异步、重试次数、退避语义。"""
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from kilostar.utils.error import NonRetryableError, RetryableError
|
|
from kilostar.utils.retry import retry_on_retryable_error
|
|
|
|
|
|
def test_sync_retries_until_success(monkeypatch):
|
|
sleep_calls: list[float] = []
|
|
monkeypatch.setattr("time.sleep", lambda s: sleep_calls.append(s))
|
|
|
|
counter = {"n": 0}
|
|
|
|
@retry_on_retryable_error(max_retries=3, base_delay=1)
|
|
def flaky() -> str:
|
|
counter["n"] += 1
|
|
if counter["n"] < 3:
|
|
raise RetryableError("temp")
|
|
return "ok"
|
|
|
|
assert flaky() == "ok"
|
|
assert counter["n"] == 3
|
|
# 第 1、2 次失败前各 sleep 一次：base * 2**0 = 1, base * 2**1 = 2
|
|
assert sleep_calls == [1, 2]
|
|
|
|
|
|
def test_sync_reraises_after_exhaustion(monkeypatch):
|
|
monkeypatch.setattr("time.sleep", lambda _s: None)
|
|
|
|
@retry_on_retryable_error(max_retries=2, base_delay=1)
|
|
def always_fail() -> None:
|
|
raise RetryableError("nope")
|
|
|
|
with pytest.raises(RetryableError):
|
|
always_fail()
|
|
|
|
|
|
def test_sync_does_not_retry_on_non_retryable(monkeypatch):
|
|
sleep_calls: list[float] = []
|
|
monkeypatch.setattr("time.sleep", lambda s: sleep_calls.append(s))
|
|
|
|
@retry_on_retryable_error(max_retries=5, base_delay=1)
|
|
def boom() -> None:
|
|
raise NonRetryableError("hard")
|
|
|
|
with pytest.raises(NonRetryableError):
|
|
boom()
|
|
assert sleep_calls == []
|
|
|
|
|
|
async def test_async_retries_until_success(monkeypatch):
|
|
sleep_calls: list[float] = []
|
|
|
|
async def fake_sleep(s: float) -> None:
|
|
sleep_calls.append(s)
|
|
|
|
monkeypatch.setattr(asyncio, "sleep", fake_sleep)
|
|
|
|
counter = {"n": 0}
|
|
|
|
@retry_on_retryable_error(max_retries=4, base_delay=2)
|
|
async def flaky() -> int:
|
|
counter["n"] += 1
|
|
if counter["n"] < 2:
|
|
raise RetryableError("temp")
|
|
return 42
|
|
|
|
assert await flaky() == 42
|
|
assert counter["n"] == 2
|
|
assert sleep_calls == [2] # base * 2**0
|
|
|
|
|
|
async def test_async_reraises_after_exhaustion(monkeypatch):
|
|
async def fake_sleep(_s: float) -> None:
|
|
return None
|
|
|
|
monkeypatch.setattr(asyncio, "sleep", fake_sleep)
|
|
|
|
@retry_on_retryable_error(max_retries=2, base_delay=1)
|
|
async def boom() -> None:
|
|
raise RetryableError("nope")
|
|
|
|
with pytest.raises(RetryableError):
|
|
await boom()
|