test(e2e): harden the suite against response-cache cross-talk, slow providers and single upstream blips

tests/e2e/e2e_config.py

@@ -78,6 +78,7 @@
 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

tests/e2e/llm_translation/endpoints_client.py

@@ -12,6 +12,7 @@
 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,38 +75,43 @@ 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):
     model: str
     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):
     model: str
     query: str
     documents: list[str]
     top_n: int
+    cache: dict[str, bool] | None = {"no-cache": True}
 
 
 class SpeechRequest(BaseModel):
@@ -446,6 +452,7 @@ def image_edit(
             file_content_type="image/png",
             file_field="image",
             response_type=ImagesResult,
+            timeout=SLOW_PROVIDER_TIMEOUT_SECONDS,
         )
 
     def generate_content(

tests/e2e/models.py

@@ -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):

tests/e2e/otel_client.py

@@ -24,7 +24,7 @@
 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 @@ def poll_traces_for_call(
         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
 
 

tests/e2e/proxy_client.py

@@ -70,6 +70,7 @@
     POLL_TIMEOUT,
     PROXY_BASE_URL,
     REQUEST_TIMEOUT,
+    SLOW_PROVIDER_TIMEOUT_SECONDS,
     settle_propagation,
 )
 from transport import HttpTransport, SplitTransport, Transport
@@ -425,6 +426,7 @@ def ocr(self, key: str, body: OcrBody) -> Result[OcrResponse]:
             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]:

tests/e2e/pytest.ini

@@ -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

tests/e2e/quota_management/budgets/test_spend_counter_reseed_e2e.py

@@ -39,6 +39,7 @@
 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 one(_: int) -> StreamingResponse:
 
     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

tests/e2e/quota_management/spend_tracking/spend_e2e_client.py

@@ -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 @@ def chat(
         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(

tests/e2e/quota_management/spend_tracking/test_spend_tracking_e2e.py

@@ -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,

tests/e2e/router/reliability_support.py

@@ -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,
     )

tests/e2e/router/test_reliability_cache_e2e.py

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

tests/e2e/transport.py

@@ -25,7 +25,13 @@
 
 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 @@ def upload[R: BaseModel](
         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 @@ def master(self) -> AuthHeaders:
         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 @@ def upload[R: BaseModel](
         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 @@ def upload[R: BaseModel](
             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 @@ def master(self) -> AuthHeaders:
         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 @@ def upload[R: BaseModel](
         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 @@ def upload[R: BaseModel](
             file_field=file_field,
             params=params,
             response_type=response_type,
+            timeout=timeout,
         )
 
     def download(self, path: str, *, headers: BaseModel) -> StreamingResponse: