118 lines
3.8 KiB
Python
118 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import time
|
|
from typing import Any
|
|
|
|
import anthropic
|
|
import httpx
|
|
import openai
|
|
import pytest
|
|
|
|
from src.llm.request_builder import execute_completion
|
|
|
|
from .conftest import make_backend, require_provider_key, wrap_async_method
|
|
from .model_matrix import LiveModelSpec, ProviderName, get_live_model_specs
|
|
|
|
pytestmark = [pytest.mark.live_llm]
|
|
|
|
GENEROUS_TIMEOUT_SECONDS = 120
|
|
TIGHT_TIMEOUT_SECONDS = 0.01
|
|
# Well under the 600s client default; generous enough to absorb SDK retries.
|
|
TIGHT_TIMEOUT_WALL_CLOCK_LIMIT_SECONDS = 30
|
|
|
|
TIMEOUT_EXCEPTIONS: dict[ProviderName, tuple[type[BaseException], ...]] = {
|
|
"anthropic": (anthropic.APITimeoutError,),
|
|
"openai": (openai.APITimeoutError,),
|
|
# google-genai raises httpx or aiohttp timeouts depending on its transport;
|
|
# aiohttp surfaces as asyncio.TimeoutError (== builtins.TimeoutError).
|
|
"gemini": (httpx.TimeoutException, TimeoutError),
|
|
}
|
|
|
|
PROVIDER_MARKS = {
|
|
"anthropic": pytest.mark.requires_anthropic,
|
|
"openai": pytest.mark.requires_openai,
|
|
"gemini": pytest.mark.requires_gemini,
|
|
}
|
|
|
|
|
|
def representative_specs() -> list[Any]:
|
|
"""One spec per provider — timeout plumbing is transport-level, not model-level."""
|
|
params: list[Any] = []
|
|
for provider in ("anthropic", "openai", "gemini"):
|
|
specs = get_live_model_specs(provider=provider)
|
|
if not specs:
|
|
continue
|
|
params.append(
|
|
pytest.param(specs[0], marks=PROVIDER_MARKS[provider], id=specs[0].id)
|
|
)
|
|
return params
|
|
|
|
|
|
def assert_timeout_reached_sdk(
|
|
model_spec: LiveModelSpec, call_kwargs: dict[str, Any], timeout_seconds: float
|
|
) -> None:
|
|
if model_spec.provider == "gemini":
|
|
http_options = call_kwargs["config"]["http_options"]
|
|
assert http_options.timeout == int(timeout_seconds * 1000)
|
|
else:
|
|
assert call_kwargs["timeout"] == timeout_seconds
|
|
|
|
|
|
def sdk_call_target(backend: Any, model_spec: LiveModelSpec) -> tuple[Any, str]:
|
|
if model_spec.provider == "gemini":
|
|
return backend._client.aio.models, "generate_content"
|
|
if model_spec.provider == "anthropic":
|
|
return backend._client.messages, "create"
|
|
return backend._client.chat.completions, "create"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("model_spec", representative_specs())
|
|
async def test_live_provider_timeout_reaches_the_wire(
|
|
model_spec: LiveModelSpec,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
require_provider_key(model_spec)
|
|
backend, config = make_backend(
|
|
model_spec, provider_params={"timeout": GENEROUS_TIMEOUT_SECONDS}
|
|
)
|
|
target, attribute = sdk_call_target(backend, model_spec)
|
|
calls = wrap_async_method(monkeypatch, target, attribute)
|
|
|
|
result = await execute_completion(
|
|
backend,
|
|
config,
|
|
messages=[{"role": "user", "content": "Reply with the single word: ok"}],
|
|
max_tokens=256,
|
|
)
|
|
|
|
assert isinstance(result.content, str)
|
|
assert result.content.strip()
|
|
assert len(calls) == 1
|
|
assert_timeout_reached_sdk(model_spec, calls[0]["kwargs"], GENEROUS_TIMEOUT_SECONDS)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("model_spec", representative_specs())
|
|
async def test_live_tight_provider_timeout_aborts_request(
|
|
model_spec: LiveModelSpec,
|
|
) -> None:
|
|
require_provider_key(model_spec)
|
|
backend, config = make_backend(
|
|
model_spec, provider_params={"timeout": TIGHT_TIMEOUT_SECONDS}
|
|
)
|
|
|
|
started = time.monotonic()
|
|
with pytest.raises(TIMEOUT_EXCEPTIONS[model_spec.provider]):
|
|
await execute_completion(
|
|
backend,
|
|
config,
|
|
messages=[{"role": "user", "content": "Reply with the single word: ok"}],
|
|
max_tokens=256,
|
|
)
|
|
elapsed = time.monotonic() - started
|
|
|
|
assert (
|
|
elapsed < TIGHT_TIMEOUT_WALL_CLOCK_LIMIT_SECONDS
|
|
), f"tight timeout took {elapsed:.1f}s — per-request timeout likely not applied"
|