tests/test_runtime_utils.py

"""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")