From b23e6585b81b411b82b9131e1f24f717f59869f1 Mon Sep 17 00:00:00 2001 From: agustin18 Date: Sun, 27 Sep 2026 23:26:33 +0000 Subject: [PATCH] test(timeout): test real timeout enforcement without wall-clock assertions Signed-off-by: agustin18 --- tests/unit/test_timeout.py | 100 ++++++++++++------------------------- 1 file changed, 31 insertions(+), 69 deletions(-) diff --git a/tests/unit/test_timeout.py b/tests/unit/test_timeout.py index e6f1d915ce2..160dd49f3eb 100644 --- a/tests/unit/test_timeout.py +++ b/tests/unit/test_timeout.py @@ -1,6 +1,7 @@ """Unit tests for litellm.timeout decorator.""" import asyncio +import time from typing import Final import pytest @@ -11,87 +12,48 @@ from litellm.timeout import timeout @pytest.mark.asyncio @pytest.mark.parametrize("timeout_arg", ["request_timeout", "force_timeout"]) -async def test_async_timeout_decorator_passes_per_call_timeout( - monkeypatch: pytest.MonkeyPatch, timeout_arg: str -) -> None: - captured_timeout: Final[list[float | None]] = [] - - async def fake_wait_for(fut, timeout): - captured_timeout.append(timeout) - return await fut - - monkeypatch.setattr("litellm.timeout.asyncio.wait_for", fake_wait_for) - - @timeout(timeout_duration=5.0) - async def sample_func(**kwargs): - return "ok" - - result: Final = await sample_func(**{timeout_arg: 0.05, "model": "test-model"}) - assert result == "ok" - assert captured_timeout == [0.05] - - -@pytest.mark.asyncio -@pytest.mark.parametrize("timeout_arg", ["request_timeout", "force_timeout"]) -async def test_async_timeout_decorator_raises_with_per_call_timeout_message( - monkeypatch: pytest.MonkeyPatch, timeout_arg: str -) -> None: - async def fake_wait_for_timeout(fut, timeout): - fut.close() - raise asyncio.TimeoutError() - - monkeypatch.setattr("litellm.timeout.asyncio.wait_for", fake_wait_for_timeout) - - @timeout(timeout_duration=5.0) - async def sample_func(**kwargs): - return "ok" +async def test_async_timeout_decorator_enforces_per_call_timeout(timeout_arg: str) -> None: + @timeout(timeout_duration=60.0) + async def hung_func(**kwargs): + await asyncio.sleep(10.0) + return "never" with pytest.raises(Timeout) as exc_info: - await sample_func(**{timeout_arg: 0.05, "model": "test-model"}) + await hung_func(**{timeout_arg: 0.001, "model": "test-model"}) - assert "0.05 second(s)" in str(exc_info.value) + assert "0.001 second(s)" in str(exc_info.value) @pytest.mark.asyncio -async def test_async_timeout_decorator_handles_none_force_timeout(monkeypatch: pytest.MonkeyPatch) -> None: - captured_timeout: Final[list[float | None]] = [] +async def test_async_timeout_decorator_extends_duration_with_per_call_timeout() -> None: + @timeout(timeout_duration=0.001) + async def fast_func(**kwargs): + return "completed" - async def fake_wait_for(fut, timeout): - captured_timeout.append(timeout) - return await fut - - monkeypatch.setattr("litellm.timeout.asyncio.wait_for", fake_wait_for) - - @timeout(timeout_duration=5.0) - async def sample_func(**kwargs): - return "ok" - - await sample_func(force_timeout=None, model="test-model") - assert captured_timeout == [5.0] + result: Final = await fast_func(request_timeout=60.0, model="test-model") + assert result == "completed" @pytest.mark.asyncio -async def test_async_timeout_decorator_uses_default_timeout(monkeypatch: pytest.MonkeyPatch) -> None: - captured_timeout: Final[list[float | None]] = [] +async def test_async_timeout_decorator_handles_none_force_timeout() -> None: + @timeout(timeout_duration=0.001) + async def hung_func(**kwargs): + await asyncio.sleep(10.0) + return "never" - async def fake_wait_for(fut, timeout): - captured_timeout.append(timeout) - return await fut + with pytest.raises(Timeout) as exc_info: + await hung_func(force_timeout=None, model="test-model") - monkeypatch.setattr("litellm.timeout.asyncio.wait_for", fake_wait_for) - - @timeout(timeout_duration=42.0) - async def sample_func(**kwargs): - return "ok" - - await sample_func(model="test-model") - assert captured_timeout == [42.0] + assert "0.001 second(s)" in str(exc_info.value) -def test_sync_timeout_decorator_runs_without_clock() -> None: - @timeout(timeout_duration=5.0) - def sample_sync_func(**kwargs): - return "sync_ok" +def test_sync_timeout_decorator_enforces_per_call_timeout() -> None: + @timeout(timeout_duration=60.0) + def hung_sync_func(**kwargs): + time.sleep(10.0) + return "never" - result: Final = sample_sync_func(request_timeout=0.05, model="test-model") - assert result == "sync_ok" + with pytest.raises(Timeout) as exc_info: + hung_sync_func(request_timeout=0.001, model="test-model") + + assert "0.001 second(s)" in str(exc_info.value)