mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-10 03:28:53 +00:00
test(e2e): harden the suite against response-cache cross-talk, slow providers and single upstream blips (#37957)
* test(e2e): send no-cache on every cacheable request body, opt in only where a hit is the assertion
The e2e proxy runs with the response cache on, so any test that re-sends an
identical chat, messages, responses, completions, embeddings or rerank body
reads back a redis copy of an earlier call instead of reaching the provider.
Five tests in the last week failed that way. Default cache: {"no-cache": true}
on those request models and pass cache=None only in the two tests whose
assertion is the cache hit itself.
* test(e2e): give image edits and OCR a 180s client timeout
Both routes wait on providers that can legitimately take longer than the
60s transport-wide request timeout (gpt-image edits, Azure Document
Intelligence), and a client-side read timeout there fails a green request.
post/upload now accept a per-call timeout like get already does; only those
two call sites use it.
* test(e2e): rerun once on network errors and upstream 5xx only
Assertion failures still fail on the first attempt; only an outcome whose
error string carries the e2e_http network kind or a 5xx status gets one
more try. Test Engine records every attempt, so the flake rate stays
visible while a single provider blip no longer reds the rc run.
* test(e2e): let the reseed burst survive one upstream failure and print why
The burst is the precondition, not the property: one 5xx among six
concurrent calls still leaves five workers racing the cold counter, which
is what the reseed assertion measures. Two or more failures still abort,
and the failing bodies are now in the message instead of only the status
codes.
* test(e2e): keep polling Jaeger through a transient query failure
poll_traces_for_call already waits up to POLL_TIMEOUT for spans to land,
but a single refused connection to the query API failed the test on the
spot. Jaeger restarted twice during today's gate runs (19:05 and 19:41
UTC, each under a minute) and took ten and three otel tests with it while
the same tests passed on the rc build minutes later. A network failure
now counts as not-yet inside the same deadline; if Jaeger is still
unreachable when the deadline passes the test fails with that error, and
any non-network failure still fails immediately.
This commit is contained in:
parent
98dfb789cb
commit
7aef79b774
12 changed files with 87 additions and 26 deletions
|
|
@ -78,6 +78,7 @@ DD_SEARCH_INTERVAL = float(os.environ.get("E2E_DD_SEARCH_INTERVAL", "10"))
|
|||
POLL_TIMEOUT = float(os.environ.get("E2E_POLL_TIMEOUT", "120"))
|
||||
POLL_INTERVAL = float(os.environ.get("E2E_POLL_INTERVAL", "5"))
|
||||
REQUEST_TIMEOUT = float(os.environ.get("E2E_REQUEST_TIMEOUT", "60"))
|
||||
SLOW_PROVIDER_TIMEOUT_SECONDS = float(os.environ.get("E2E_SLOW_PROVIDER_TIMEOUT", "180"))
|
||||
|
||||
# How long a control-plane write (/model/new, /guardrails, /v1/agents) may take to
|
||||
# reach EVERY replica. Distinct from POLL_TIMEOUT, which is sized for spend-row
|
||||
|
|
|
|||
|
|
@ -12,6 +12,7 @@ from __future__ import annotations
|
|||
from dataclasses import dataclass
|
||||
from typing import Literal
|
||||
|
||||
from e2e_config import SLOW_PROVIDER_TIMEOUT_SECONDS
|
||||
from e2e_http import BinaryStream, Result, StreamingResponse
|
||||
from models import CacheControl, ChatMessage, LiteLLMParamsBody, RichMessage, TextBlock
|
||||
from proxy_client import ProxyClient
|
||||
|
|
@ -74,12 +75,14 @@ class ResponsesRequest(BaseModel):
|
|||
stream: bool = False
|
||||
tools: list[ResponsesFunctionTool] | None = None
|
||||
guardrails: list[str] | None = None
|
||||
cache: dict[str, bool] | None = {"no-cache": True}
|
||||
|
||||
|
||||
class MessagesRequest(BaseModel):
|
||||
model: str
|
||||
max_tokens: int
|
||||
messages: list[ChatMessage]
|
||||
cache: dict[str, bool] | None = {"no-cache": True}
|
||||
|
||||
|
||||
class RichMessagesRequest(BaseModel):
|
||||
|
|
@ -87,18 +90,20 @@ class RichMessagesRequest(BaseModel):
|
|||
max_tokens: int = 64
|
||||
system: list[TextBlock]
|
||||
messages: list[RichMessage]
|
||||
cache: dict[str, bool] = {"no-cache": True}
|
||||
cache: dict[str, bool] | None = {"no-cache": True}
|
||||
|
||||
|
||||
class CompletionsRequest(BaseModel):
|
||||
model: str
|
||||
prompt: str
|
||||
max_tokens: int = 32
|
||||
cache: dict[str, bool] | None = {"no-cache": True}
|
||||
|
||||
|
||||
class EmbeddingsRequest(BaseModel):
|
||||
model: str
|
||||
input: str
|
||||
cache: dict[str, bool] | None = {"no-cache": True}
|
||||
|
||||
|
||||
class RerankRequest(BaseModel):
|
||||
|
|
@ -106,6 +111,7 @@ class RerankRequest(BaseModel):
|
|||
query: str
|
||||
documents: list[str]
|
||||
top_n: int
|
||||
cache: dict[str, bool] | None = {"no-cache": True}
|
||||
|
||||
|
||||
class SpeechRequest(BaseModel):
|
||||
|
|
@ -446,6 +452,7 @@ class EndpointsClient:
|
|||
file_content_type="image/png",
|
||||
file_field="image",
|
||||
response_type=ImagesResult,
|
||||
timeout=SLOW_PROVIDER_TIMEOUT_SECONDS,
|
||||
)
|
||||
|
||||
def generate_content(
|
||||
|
|
|
|||
|
|
@ -233,6 +233,7 @@ class ChatBody(BaseModel):
|
|||
tool_choice: str | None = None
|
||||
guardrails: list[str] | None = None
|
||||
response_format: dict[str, object] | None = None
|
||||
cache: dict[str, bool] | None = {"no-cache": True}
|
||||
|
||||
|
||||
class RouterSettingsOverride(BaseModel):
|
||||
|
|
@ -431,6 +432,7 @@ class AnthropicMessagesBody(BaseModel):
|
|||
stream: bool | None = None
|
||||
tools: list[AnthropicTool] | None = None
|
||||
guardrails: list[str] | None = None
|
||||
cache: dict[str, bool] | None = {"no-cache": True}
|
||||
|
||||
|
||||
class CountTokensBody(BaseModel):
|
||||
|
|
@ -496,6 +498,7 @@ class McpServerInfo(BaseModel):
|
|||
class EmbedBody(BaseModel):
|
||||
model: str
|
||||
input: str
|
||||
cache: dict[str, bool] | None = {"no-cache": True}
|
||||
|
||||
|
||||
class EmbedResponse(BaseModel):
|
||||
|
|
|
|||
|
|
@ -24,7 +24,7 @@ import pytest
|
|||
from pydantic import BaseModel, ConfigDict, Field
|
||||
|
||||
from e2e_config import OTEL_QUERY_URL, POLL_INTERVAL, POLL_TIMEOUT
|
||||
from e2e_http import URL, NoBody, Success, get
|
||||
from e2e_http import URL, NetworkError, NoBody, Result, Success, get
|
||||
|
||||
#: OTEL resource service.name the proxy exports under (OTEL_SERVICE_NAME default).
|
||||
JAEGER_SERVICE = "litellm"
|
||||
|
|
@ -100,18 +100,20 @@ def _settled(trace: JaegerTrace, names: set[str], prefixes: set[str]) -> bool:
|
|||
class OtelReader:
|
||||
query_url: str
|
||||
|
||||
def traces_for_call(self, call_id: str) -> list[JaegerTrace]:
|
||||
"""Every trace holding a span tagged with this call id. Jaeger matches
|
||||
spans server-side and returns their full traces; more than one hit for
|
||||
one call IS the split-trace bug, so this never collapses to one."""
|
||||
result = get(
|
||||
def _query_traces(self, call_id: str) -> Result[JaegerTracesPage]:
|
||||
return get(
|
||||
URL(f"{self.query_url}/api/traces"),
|
||||
headers=NoBody(),
|
||||
params=_TracesQuery(service=JAEGER_SERVICE, tags=json.dumps({CALL_ID_TAG: call_id})),
|
||||
response_type=JaegerTracesPage,
|
||||
timeout=30.0,
|
||||
)
|
||||
match result:
|
||||
|
||||
def traces_for_call(self, call_id: str) -> list[JaegerTrace]:
|
||||
"""Every trace holding a span tagged with this call id. Jaeger matches
|
||||
spans server-side and returns their full traces; more than one hit for
|
||||
one call IS the split-trace bug, so this never collapses to one."""
|
||||
match self._query_traces(call_id):
|
||||
case Success(data=page):
|
||||
return page.data
|
||||
case failure:
|
||||
|
|
@ -128,11 +130,24 @@ class OtelReader:
|
|||
on a split trace this never settles and the orphan comes back."""
|
||||
deadline = time.monotonic() + POLL_TIMEOUT
|
||||
hits: list[JaegerTrace] = []
|
||||
unreachable: NetworkError | None = None
|
||||
while time.monotonic() < deadline:
|
||||
hits = self.traces_for_call(call_id)
|
||||
if len(hits) == 1 and _settled(hits[0], settled_names, settled_prefixes):
|
||||
return hits
|
||||
match self._query_traces(call_id):
|
||||
case Success(data=page):
|
||||
unreachable = None
|
||||
hits = page.data
|
||||
if len(hits) == 1 and _settled(hits[0], settled_names, settled_prefixes):
|
||||
return hits
|
||||
case NetworkError() as failure:
|
||||
unreachable = failure
|
||||
case failure:
|
||||
pytest.fail(f"Jaeger query API at {self.query_url} failed: {failure}")
|
||||
time.sleep(POLL_INTERVAL)
|
||||
if unreachable is not None:
|
||||
pytest.fail(
|
||||
f"Jaeger query API at {self.query_url} stayed unreachable until the "
|
||||
f"{POLL_TIMEOUT}s poll deadline: {unreachable}"
|
||||
)
|
||||
return hits
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -70,6 +70,7 @@ from e2e_config import (
|
|||
POLL_TIMEOUT,
|
||||
PROXY_BASE_URL,
|
||||
REQUEST_TIMEOUT,
|
||||
SLOW_PROVIDER_TIMEOUT_SECONDS,
|
||||
settle_propagation,
|
||||
)
|
||||
from transport import HttpTransport, SplitTransport, Transport
|
||||
|
|
@ -425,6 +426,7 @@ class ProxyClient:
|
|||
headers=self.transport.bearer(key),
|
||||
json=body,
|
||||
response_type=OcrResponse,
|
||||
timeout=SLOW_PROVIDER_TIMEOUT_SECONDS,
|
||||
)
|
||||
|
||||
def count_tokens(self, key: str, body: CountTokensBody) -> Result[CountTokensResponse]:
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@
|
|||
# Config when any e2e suite under tests/e2e/ is run directly, e.g.
|
||||
# uv run pytest tests/e2e/quota_management/spend_tracking/ -v
|
||||
# The e2e marker is also registered in conftest.py for runs rooted elsewhere.
|
||||
addopts = --strict-markers --strict-config
|
||||
addopts = --strict-markers --strict-config --reruns 1 --only-rerun "kind='network'" --only-rerun "status_code=5[0-9][0-9]"
|
||||
markers =
|
||||
e2e: live test that requires a running proxy and real provider keys
|
||||
load: heavy throughput/load test; collected last so it never perturbs latency-sensitive suites
|
||||
|
|
|
|||
|
|
@ -39,6 +39,7 @@ pytestmark = pytest.mark.e2e
|
|||
MODEL = "claude-haiku-4-5"
|
||||
ACCUMULATE_CALLS = 24
|
||||
BURST = 6
|
||||
BURST_TOLERATED_FAILURES = 1
|
||||
# proxy_batch_write_at (60s) flushes the spend to the DB and default_redis_ttl (20s)
|
||||
# expires the counter; this waits out both.
|
||||
COLD_WAIT_SECONDS = 80
|
||||
|
|
@ -174,9 +175,11 @@ def test_cold_counter_reseed_keeps_counter_equal_to_db_spend(
|
|||
|
||||
with ThreadPoolExecutor(max_workers=BURST) as pool:
|
||||
burst_results = list(pool.map(one, range(BURST)))
|
||||
assert all(r.ok for r in burst_results), (
|
||||
"some burst calls failed; cannot exercise concurrent reseed. "
|
||||
f"statuses={[r.status_code for r in burst_results]}"
|
||||
failed = [r for r in burst_results if not r.ok]
|
||||
assert len(failed) <= BURST_TOLERATED_FAILURES, (
|
||||
"too many burst calls failed; cannot exercise concurrent reseed. "
|
||||
f"statuses={[r.status_code for r in burst_results]} "
|
||||
f"bodies={[r.body[:300] for r in failed]}"
|
||||
)
|
||||
|
||||
counter: float | None = None
|
||||
|
|
|
|||
|
|
@ -65,6 +65,7 @@ def _chat_body(
|
|||
tags: list[str] | None = None,
|
||||
user: str | None = None,
|
||||
stream: bool = False,
|
||||
cache: dict[str, bool] | None = {"no-cache": True},
|
||||
) -> ChatBody:
|
||||
return ChatBody(
|
||||
model=model,
|
||||
|
|
@ -73,6 +74,7 @@ def _chat_body(
|
|||
stream=stream,
|
||||
user=user,
|
||||
metadata=ChatMetadata(tags=tags) if tags else None,
|
||||
cache=cache,
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -89,9 +91,11 @@ class SpendClient:
|
|||
max_tokens: int | None = None,
|
||||
tags: list[str] | None = None,
|
||||
user: str | None = None,
|
||||
cache: dict[str, bool] | None = {"no-cache": True},
|
||||
) -> Result[ChatResponse]:
|
||||
return self.proxy.chat(
|
||||
key, _chat_body(model, content, max_tokens=max_tokens, tags=tags, user=user)
|
||||
key,
|
||||
_chat_body(model, content, max_tokens=max_tokens, tags=tags, user=user, cache=cache),
|
||||
)
|
||||
|
||||
def chat_stream(
|
||||
|
|
|
|||
|
|
@ -226,8 +226,8 @@ def test_cache_hit_is_zero_cost_and_suffixed(
|
|||
# populated. The marker keeps each run isolated - a fixed prompt would persist
|
||||
# in the shared response cache across runs and make both calls hit (flaky).
|
||||
prompt = f"What is the capital of France? Answer in one word. {unique_marker()}"
|
||||
_ = unwrap(client.chat(scoped_key, "gemini-2.5-flash", prompt, max_tokens=16))
|
||||
_ = unwrap(client.chat(scoped_key, "gemini-2.5-flash", prompt, max_tokens=16))
|
||||
_ = unwrap(client.chat(scoped_key, "gemini-2.5-flash", prompt, max_tokens=16, cache=None))
|
||||
_ = unwrap(client.chat(scoped_key, "gemini-2.5-flash", prompt, max_tokens=16, cache=None))
|
||||
|
||||
rows = client.poll_logs_for_key(
|
||||
scoped_key,
|
||||
|
|
|
|||
|
|
@ -47,6 +47,7 @@ def chat_override(
|
|||
content: str,
|
||||
override: RouterSettingsOverride | None = None,
|
||||
stream: bool = False,
|
||||
cache: dict[str, bool] | None = {"no-cache": True},
|
||||
) -> StreamingResponse:
|
||||
"""POST /chat/completions with an optional per-request router_settings_override,
|
||||
returning the raw outcome so tests read status, body, and reliability headers."""
|
||||
|
|
@ -59,6 +60,7 @@ def chat_override(
|
|||
max_tokens=64,
|
||||
stream=stream,
|
||||
router_settings_override=override,
|
||||
cache=cache,
|
||||
),
|
||||
stream=stream,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -23,13 +23,13 @@ class TestReliabilityCache:
|
|||
def test_exact_cache_returns_cached(self, client: ComplexityRouterClient, scoped_key: str) -> None:
|
||||
prompt = f"cache probe {unique_marker()}"
|
||||
|
||||
first = chat_override(client.proxy, scoped_key, "gpt-5.5", prompt)
|
||||
first = chat_override(client.proxy, scoped_key, "gpt-5.5", prompt, cache=None)
|
||||
assert first.status_code == 200, f"first call should succeed, got {first.status_code}: {first.body[:300]}"
|
||||
assert "x-litellm-cache-key" not in first.headers, (
|
||||
"first (uncached) call must not report a cache-key header"
|
||||
)
|
||||
|
||||
second = chat_override(client.proxy, scoped_key, "gpt-5.5", prompt)
|
||||
second = chat_override(client.proxy, scoped_key, "gpt-5.5", prompt, cache=None)
|
||||
assert second.status_code == 200, f"second call should succeed, got {second.status_code}: {second.body[:300]}"
|
||||
assert "x-litellm-cache-key" in second.headers, (
|
||||
"second identical call should hit the response cache and report a cache-key header "
|
||||
|
|
|
|||
|
|
@ -25,7 +25,13 @@ from e2e_http import (
|
|||
|
||||
class Transport(Protocol):
|
||||
def post[R: BaseModel](
|
||||
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
|
||||
self,
|
||||
path: str,
|
||||
*,
|
||||
headers: BaseModel,
|
||||
json: BaseModel,
|
||||
response_type: type[R],
|
||||
timeout: float | None = None,
|
||||
) -> Result[R]: ...
|
||||
|
||||
def stream(
|
||||
|
|
@ -93,6 +99,7 @@ class Transport(Protocol):
|
|||
file_field: str = "file",
|
||||
params: BaseModel | None = None,
|
||||
response_type: type[R],
|
||||
timeout: float | None = None,
|
||||
) -> Result[R]: ...
|
||||
|
||||
def download(self, path: str, *, headers: BaseModel) -> StreamingResponse: ...
|
||||
|
|
@ -120,14 +127,22 @@ class HttpTransport:
|
|||
return self.bearer(self.master_key)
|
||||
|
||||
def post[R: BaseModel](
|
||||
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
|
||||
self,
|
||||
path: str,
|
||||
*,
|
||||
headers: BaseModel,
|
||||
json: BaseModel,
|
||||
response_type: type[R],
|
||||
timeout: float | None = None,
|
||||
) -> Result[R]:
|
||||
"""`timeout` overrides the transport-wide request_timeout for this call, for
|
||||
provider operations that legitimately outlive it (image edits, OCR)."""
|
||||
return e2e_http.post(
|
||||
self._url(path),
|
||||
headers=headers,
|
||||
json=json,
|
||||
response_type=response_type,
|
||||
timeout=self.request_timeout,
|
||||
timeout=self.request_timeout if timeout is None else timeout,
|
||||
)
|
||||
|
||||
def get[R: BaseModel](
|
||||
|
|
@ -250,6 +265,7 @@ class HttpTransport:
|
|||
file_field: str = "file",
|
||||
params: BaseModel | None = None,
|
||||
response_type: type[R],
|
||||
timeout: float | None = None,
|
||||
) -> Result[R]:
|
||||
return e2e_http.upload(
|
||||
self._url(path),
|
||||
|
|
@ -261,7 +277,7 @@ class HttpTransport:
|
|||
file_field=file_field,
|
||||
params=params,
|
||||
response_type=response_type,
|
||||
timeout=self.request_timeout,
|
||||
timeout=self.request_timeout if timeout is None else timeout,
|
||||
)
|
||||
|
||||
def download(self, path: str, *, headers: BaseModel) -> StreamingResponse:
|
||||
|
|
@ -327,10 +343,16 @@ class SplitTransport:
|
|||
return self.data.master
|
||||
|
||||
def post[R: BaseModel](
|
||||
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
|
||||
self,
|
||||
path: str,
|
||||
*,
|
||||
headers: BaseModel,
|
||||
json: BaseModel,
|
||||
response_type: type[R],
|
||||
timeout: float | None = None,
|
||||
) -> Result[R]:
|
||||
return self._route(path).post(
|
||||
path, headers=headers, json=json, response_type=response_type
|
||||
path, headers=headers, json=json, response_type=response_type, timeout=timeout
|
||||
)
|
||||
|
||||
def get[R: BaseModel](
|
||||
|
|
@ -426,6 +448,7 @@ class SplitTransport:
|
|||
file_field: str = "file",
|
||||
params: BaseModel | None = None,
|
||||
response_type: type[R],
|
||||
timeout: float | None = None,
|
||||
) -> Result[R]:
|
||||
return self._route(path).upload(
|
||||
path,
|
||||
|
|
@ -437,6 +460,7 @@ class SplitTransport:
|
|||
file_field=file_field,
|
||||
params=params,
|
||||
response_type=response_type,
|
||||
timeout=timeout,
|
||||
)
|
||||
|
||||
def download(self, path: str, *, headers: BaseModel) -> StreamingResponse:
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue