test(timeout): test real timeout enforcement without wall-clock assertions

Signed-off-by: agustin18 <agustinsaiz02@gmail.com>

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)