Spaces:
Sleeping
Sleeping
File size: 2,233 Bytes
5db30ac | 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 | import asyncio
import sys
import unittest
from pathlib import Path
from unittest.mock import AsyncMock, patch
import httpx
ROOT = Path(__file__).resolve().parents[1]
SRC = ROOT / "src"
if str(SRC) not in sys.path:
sys.path.insert(0, str(SRC))
from elbench.execution.retry import is_retryable_exception, sleep_before_retry # noqa: E402
from elbench.schemas.config import RetryConfig # noqa: E402
class RetryDelayTest(unittest.TestCase):
def test_retryable_false_marker_disables_retry(self) -> None:
exc = RuntimeError("known non-retryable failure")
exc.retryable = False
self.assertFalse(is_retryable_exception(exc, retryable_status_codes={500, 504}))
def test_429_waits_for_minute_window_when_retry_after_is_absent(self) -> None:
request = httpx.Request("POST", "https://api.innospark.cn/v1/chat/completions")
response = httpx.Response(429, request=request)
exc = httpx.HTTPStatusError("too many requests", request=request, response=response)
policy = RetryConfig(
initial_delay_seconds=1,
max_delay_seconds=20,
backoff_multiplier=2,
jitter_ratio=0,
)
async def run() -> float:
with patch("asyncio.sleep", new=AsyncMock()) as sleep:
delay = await sleep_before_retry(policy, 1, exc)
sleep.assert_awaited_once_with(60.0)
return delay
self.assertEqual(asyncio.run(run()), 60.0)
def test_429_honors_retry_after_when_longer_than_default_window(self) -> None:
request = httpx.Request("POST", "https://api.innospark.cn/v1/chat/completions")
response = httpx.Response(429, headers={"Retry-After": "90"}, request=request)
exc = httpx.HTTPStatusError("too many requests", request=request, response=response)
policy = RetryConfig(jitter_ratio=0)
async def run() -> float:
with patch("asyncio.sleep", new=AsyncMock()) as sleep:
delay = await sleep_before_retry(policy, 1, exc)
sleep.assert_awaited_once_with(90.0)
return delay
self.assertEqual(asyncio.run(run()), 90.0)
if __name__ == "__main__":
unittest.main()
|