| """Tests for runtime_utils.py.""" |
|
|
| import httpx |
| import pytest |
| from huggingface_hub.errors import HfHubHTTPError |
|
|
| from discord_support_issues.hf_runtime import should_retry_hf_exception |
| from discord_support_issues.runtime_utils import ( |
| ProcessLockError, |
| async_retry_call, |
| exclusive_process_lock, |
| retry_call, |
| shutdown_controller, |
| ) |
|
|
|
|
| def test_retry_call_retries_transient_errors(): |
| shutdown_controller.reset() |
| attempts = 0 |
|
|
| def flaky(): |
| nonlocal attempts |
| attempts += 1 |
| if attempts < 3: |
| raise OSError("temporary") |
| return "ok" |
|
|
| result = retry_call( |
| "sync operation", |
| flaky, |
| attempts=3, |
| base_delay=0, |
| max_delay=0, |
| jitter=0, |
| retry_exceptions=(OSError,), |
| ) |
|
|
| assert result == "ok" |
| assert attempts == 3 |
|
|
|
|
| @pytest.mark.asyncio |
| async def test_async_retry_call_retries_transient_errors(): |
| shutdown_controller.reset() |
| attempts = 0 |
|
|
| async def flaky(): |
| nonlocal attempts |
| attempts += 1 |
| if attempts < 2: |
| raise TimeoutError("temporary") |
| return "ok" |
|
|
| result = await async_retry_call( |
| "async operation", |
| flaky, |
| attempts=2, |
| base_delay=0, |
| max_delay=0, |
| jitter=0, |
| retry_exceptions=(TimeoutError,), |
| ) |
|
|
| assert result == "ok" |
| assert attempts == 2 |
|
|
|
|
| def test_should_retry_hf_exception_distinguishes_transient_failures(): |
| request = httpx.Request("GET", "https://huggingface.co") |
| retryable_response = httpx.Response(status_code=503, request=request) |
| non_retryable_response = httpx.Response(status_code=400, request=request) |
|
|
| assert should_retry_hf_exception(OSError("temporary")) |
| assert should_retry_hf_exception(TimeoutError("temporary")) |
| assert should_retry_hf_exception(httpx.ReadTimeout("request timed out")) |
| assert should_retry_hf_exception(RuntimeError("gateway timeout")) |
| assert should_retry_hf_exception( |
| HfHubHTTPError("service unavailable", response=retryable_response) |
| ) |
| assert not should_retry_hf_exception( |
| HfHubHTTPError("bad request", response=non_retryable_response) |
| ) |
|
|
|
|
| def test_exclusive_process_lock_rejects_an_active_lock(tmp_path): |
| lock_path = tmp_path / "sync.lock" |
|
|
| with ( |
| exclusive_process_lock(lock_path, operation_name="sync"), |
| pytest.raises(ProcessLockError, match="sync is already running"), |
| exclusive_process_lock(lock_path, operation_name="sync"), |
| ): |
| pytest.fail("nested lock acquisition should not succeed") |
|
|
