test(e2e): tolerate provider-side flakes on five full-suite cells (#42628)

* test(e2e): tolerate provider-side flakes on five full-suite cells

Mistral OCR retries a provider-relayed 429 with backoff, the Vertex vision
probe turns reasoning off so its 32 tokens go to the answer, the Vertex
cache cell spaces eight never-seen prefixes 15s apart around Google's
nondeterministic minimum-token rejection and prices the cached tokens
instead of prompt_tokens, and the Azure content-policy cell resends the
jailbreak prompt while Azure skips its filter

* test(e2e): shorten the new helper docstrings

* test(e2e): accept a relayed provider 429 on the rust OCR cells

The gateway already retries a provider 429 three times per call and the
Mistral key is shared across pipelines, so a throttle can hold across all
four attempts of the OCR cell. After the bounded retries the cell now
accepts the gateway's faithful relay of the provider's 429 (throttling_error,
code 429) as its second expected outcome; the gateway's own 429 and any
other error still fail the cell at once.

* test(e2e): drop the harness unit tests, the live cells cover the helpers

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>

tests/e2e/e2e_http.py

@@ -345,6 +345,50 @@ def request_with_retry[T: RetryableResponse](
     return issue()
 
 
+PROVIDER_RATE_LIMIT_MARKER: Final = "litellm.RateLimitError"
+PROVIDER_RATE_LIMIT_ATTEMPTS: Final = 4
+PROVIDER_RATE_LIMIT_BACKOFF_SECONDS: Final = 5.0
+
+
+def tolerate_provider_rate_limit[R: BaseModel](
+    issue: Callable[[], Result[R]],
+    *,
+    attempts: int = PROVIDER_RATE_LIMIT_ATTEMPTS,
+    sleep: Callable[[float], None] = time.sleep,
+) -> Result[R]:
+    """Retry a call up to `attempts` times while the proxy relays the provider's own 429;
+    any other outcome, the proxy's own 429 included, comes back at once."""
+    for attempt in range(1, attempts):
+        match issue():
+            case RateLimitedError(body=body, retry_after_seconds=retry_after) if PROVIDER_RATE_LIMIT_MARKER in body:
+                delay = retry_after or PROVIDER_RATE_LIMIT_BACKOFF_SECONDS * (1 << (attempt - 1))
+                print(
+                    f"e2e-http: provider rate limit relayed by the proxy; retry {attempt}/{attempts - 1} in {delay}s",
+                    flush=True,
+                )
+                sleep(delay)
+            case result:
+                return result
+    return issue()
+
+
+class ProxyErrorDetail(BaseModel):
+    message: str
+    type: str
+    code: str
+
+
+class _ProxyErrorBody(BaseModel):
+    error: ProxyErrorDetail
+
+
+def relayed_provider_rate_limit(outcome: RateLimitedError) -> ProxyErrorDetail | None:
+    """The provider's own 429 as the proxy relayed it, or None when the 429 is the proxy's own."""
+    if PROVIDER_RATE_LIMIT_MARKER not in outcome.body:
+        return None
+    return _ProxyErrorBody.model_validate_json(outcome.body).error
+
+
 class ClassifiableResponse(Protocol):
     """What classifying an outcome reads off a response. requests.Response satisfies
     it, and so does a fake, so the classification rules are testable on their own."""
@@ -931,7 +975,10 @@ class PreparedForward:
 
 
 def prepare_forward(
-    method: str, url: str, headers: dict[str, str], body: bytes | None,
+    method: str,
+    url: str,
+    headers: dict[str, str],
+    body: bytes | None,
 ) -> PreparedForward | NetworkError:
     try:
         with requests.Session() as session:
@@ -950,7 +997,8 @@ def forward_prepared_stream(prepared: PreparedForward, timeout: float) -> Stream
     except requests.RequestException as exc:
         return NetworkError(message=str(exc))
     return StreamHead(
-        resp.status_code, {name.lower(): value for name, value in resp.headers.items()},
+        resp.status_code,
+        {name.lower(): value for name, value in resp.headers.items()},
         primed_steps(_stream_steps(resp)),
     )
 

tests/e2e/llm_translation/test_cache_control.py

@@ -13,7 +13,13 @@ Each case asserts the feature actually happened, not just a 200. Coverage matrix
   call, so a never-seen prefix must come back cached on its very first call
   (Gemini's implicit caching cannot hit a cold prefix), the cached count must
   cover the marked block, and the spend row must be billed below the uncached
-  price of the prompt.
+  price of the cached tokens. Vertex's cache create is nondeterministic:
+  identical bodies come back 200 or with the minimum-token 400 ("The cached
+  content is of 1 tokens"), in failure bursts of 45 seconds and more, so up to
+  eight never-seen prefixes are tried with a pause after each rejection. The
+  billing check prices the cached tokens rather than prompt_tokens, which
+  Vertex reports inclusive of the cached prefix on some calls and exclusive
+  of it on others.
 - Anthropic (claude-haiku-4-5, direct): the same ``cache_control`` prefix over
   the OpenAI-compatible route; the second call must report cache-read tokens > 0.
 - OpenAI (gpt-5.6): automatic prompt caching needs no request marker, so the
@@ -50,7 +56,8 @@ VERTEX_MODEL = "vertex_ai/gemini-2.5-flash"
 ANTHROPIC_MODEL = "anthropic/claude-haiku-4-5-20251001"
 OPENAI_MODEL = "openai/gpt-5.6"
 VERTEX_CACHE_TTL: Final = "300s"
-VERTEX_COLD_CALL_ATTEMPTS: Final = 3
+VERTEX_COLD_CALL_ATTEMPTS: Final = 8
+VERTEX_COLD_CALL_PAUSE_SECONDS: Final = 15.0
 VERTEX_MINIMUM_CACHED_TOKENS: Final = 1024
 CACHED_SHARE_OF_PROMPT: Final = 0.9
 VERTEX_CACHE_REJECTION_MARKER: Final = "minimum token count to start explicit caching"
@@ -156,26 +163,36 @@ def _cold_cache_call(send: Callable[[str], Result[ChatResponse]]) -> ChatRespons
             return unwrap(result)
 
 
+def _first_engaged_cold_call(send: Callable[[str], Result[ChatResponse]]) -> ChatResponse | None:
+    for attempt in range(1, VERTEX_COLD_CALL_ATTEMPTS + 1):
+        candidate = _cold_cache_call(send)
+        if candidate is not None and _cached_read_tokens(candidate.usage) >= VERTEX_MINIMUM_CACHED_TOKENS:
+            return candidate
+        if attempt < VERTEX_COLD_CALL_ATTEMPTS:
+            print(
+                f"cache_control: vertex did not engage the cache on cold attempt {attempt}/{VERTEX_COLD_CALL_ATTEMPTS}; "
+                f"pausing {VERTEX_COLD_CALL_PAUSE_SECONDS}s before the next never-seen prefix",
+                flush=True,
+            )
+            time.sleep(VERTEX_COLD_CALL_PAUSE_SECONDS)
+    return None
+
+
 def _first_cold_call_reads_cache(model: str, send: Callable[[str], Result[ChatResponse]]) -> ChatResponse:
-    completion: Final = next(
-        (
-            candidate
-            for candidate in (_cold_cache_call(send) for _ in range(VERTEX_COLD_CALL_ATTEMPTS))
-            if candidate is not None and _cached_read_tokens(candidate.usage) >= VERTEX_MINIMUM_CACHED_TOKENS
-        ),
-        None,
-    )
+    completion: Final = _first_engaged_cold_call(send)
     assert completion is not None, (
-        f"{model}: {VERTEX_COLD_CALL_ATTEMPTS} never-seen prompts marked with cache_control were each either "
-        f"rejected by Vertex's minimum-token check or served with fewer than {VERTEX_MINIMUM_CACHED_TOKENS} "
-        "cached tokens on their first call; explicit context caching did not engage"
+        f"{model}: {VERTEX_COLD_CALL_ATTEMPTS} never-seen prompts marked with cache_control, spread over "
+        f"{VERTEX_COLD_CALL_PAUSE_SECONDS * (VERTEX_COLD_CALL_ATTEMPTS - 1):.0f}s, were each either rejected by "
+        f"Vertex's minimum-token check or served with fewer than {VERTEX_MINIMUM_CACHED_TOKENS} cached tokens on "
+        "their first call; explicit context caching did not engage"
     )
     assert completion.choices, f"{model}: cached call returned no choices: {completion}"
     usage: Final = completion.usage
     cached: Final = _cached_read_tokens(usage)
-    assert usage and usage.prompt_tokens and cached >= CACHED_SHARE_OF_PROMPT * usage.prompt_tokens, (
-        f"{model}: only {cached} of {usage.prompt_tokens if usage else None} prompt tokens were served from the "
-        "cache; the cache_control block was not cached whole"
+    assert usage and usage.prompt_tokens, f"{model}: cached completion carried no prompt_tokens: {usage}"
+    assert cached >= CACHED_SHARE_OF_PROMPT * usage.prompt_tokens, (
+        f"{model}: only {cached} of {usage.prompt_tokens} prompt tokens were served from the cache; the "
+        "cache_control block was not cached whole"
     )
     return completion
 
@@ -196,10 +213,11 @@ def _assert_billed_below_uncached_prompt(client: PassthroughClient, model: str,
     assert row.prompt_tokens == usage.prompt_tokens, (
         f"{model}: spend row prompt_tokens {row.prompt_tokens} != response prompt_tokens {usage.prompt_tokens}"
     )
-    uncached_prompt_cost: Final = usage.prompt_tokens * _input_rate(client, model)
-    assert row.spend is not None and row.spend < uncached_prompt_cost, (
-        f"{model}: spend {row.spend} is not below the uncached price of the prompt alone ({uncached_prompt_cost} for "
-        f"{usage.prompt_tokens} tokens); cache-read pricing was not applied"
+    cached: Final = _cached_read_tokens(usage)
+    uncached_read_cost: Final = cached * _input_rate(client, model)
+    assert row.spend is not None and row.spend < uncached_read_cost, (
+        f"{model}: spend {row.spend} is not below the uncached price of the {cached} tokens read from the cache "
+        f"({uncached_read_cost}); cache-read pricing was not applied"
     )
 
 

tests/e2e/llm_translation/test_chat_completions_regression_e2e.py

@@ -110,10 +110,11 @@ def _vision_messages() -> list[ChatMessage]:
 
 def _assert_describes_cat(response: ChatResponse) -> None:
     assert response.choices, f"vision returned no choices: {response}"
-    message = response.choices[0].message
-    content = (message.content if message else None) or ""
+    choice = response.choices[0]
+    content = (choice.message.content if choice.message else None) or ""
     assert any(term in content.lower() for term in ("cat", "feline", "kitten", "kitty")), (
-        f"vision response did not describe the image: {content[:200]}"
+        f"vision response did not describe the image: {content[:200]!r} "
+        f"(finish_reason={choice.finish_reason!r}, usage={response.usage})"
     )
 
 
@@ -421,7 +422,11 @@ class TestVertexChatCompletions:
         model = self._register(client, resources, "e2e-vertex-vision")
         key = resources.key()
 
-        response = unwrap(client.proxy.chat(key, ChatBody(model=model, messages=_vision_messages(), max_tokens=32)))
+        response = unwrap(
+            client.proxy.chat(
+                key, ChatBody(model=model, messages=_vision_messages(), max_tokens=32, reasoning_effort="none")
+            )
+        )
         _assert_describes_cat(response)
 
     @pytest.mark.covers(

tests/e2e/llm_translation/test_ocr_rust_e2e.py

@@ -11,16 +11,28 @@ well-formed OCR document comes back. Per the e2e hard-fail contract, a case
 fails when no proxy answers and also fails once a request reaches it: the proxy
 fetches each provider's referenced secrets, so a
 missing credential surfaces as a live provider error rather than silent green.
+
+The provider keys are shared with other pipelines, so a provider's rate limit can
+hold across the bounded retries. The case then accepts the gateway's faithful relay
+of that 429 (throttling_error, code 429) as its second expected outcome; any other
+non-success still fails at once.
 """
 
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Protocol
+from typing import Final, Protocol
 
 import pytest
 from e2e_config import unique_marker
-from e2e_http import assert_client_error, unwrap
+from e2e_http import (
+    PROVIDER_RATE_LIMIT_ATTEMPTS,
+    RateLimitedError,
+    Success,
+    assert_client_error,
+    relayed_provider_rate_limit,
+    tolerate_provider_rate_limit,
+)
 from lifecycle import ResourceManager
 from models import LiteLLMParamsBody, OcrBody, OcrDocument, OcrResponse
 from proxy_client import ProxyClient
@@ -146,24 +158,36 @@ def _assert_ocr_document(response: OcrResponse) -> None:
     assert response.pages[0].markdown is not None, "first page has no markdown"
 
 
+def _assert_provider_rate_limit_relayed(model: str, outcome: RateLimitedError) -> None:
+    detail: Final = relayed_provider_rate_limit(outcome)
+    assert detail is not None, f"{model}: the 429 is the gateway's own, not the provider's: {outcome.body}"
+    assert (detail.type, detail.code) == ("throttling_error", "429"), f"{model}: provider 429 relayed as {detail!r}"
+    print(
+        f"{model}: the provider's rate limit held across {PROVIDER_RATE_LIMIT_ATTEMPTS} attempts; "
+        f"the gateway relayed it as {detail.type} {detail.code}",
+        flush=True,
+    )
+
+
 class TestRustOcrGateway:
     @pytest.mark.parametrize("case", RUST_OCR_CASES, ids=_CASE_IDS)
-    def test_rust_ocr_response(
-        self, proxy: ProxyClient, resources: ResourceManager, case: _OcrCase
-    ) -> None:
+    def test_rust_ocr_response(self, proxy: ProxyClient, resources: ResourceManager, case: _OcrCase) -> None:
         model = f"rust-ocr-{case.suffix}-{unique_marker()}"
         model_id = proxy.create_model(model, case.provider.litellm_params())
         resources.defer(lambda: proxy.delete_model(model_id))
         key = resources.key()
 
-        response = unwrap(proxy.ocr(key, OcrBody(model=model, document=case.document)))
-        _assert_ocr_document(response)
+        match tolerate_provider_rate_limit(lambda: proxy.ocr(key, OcrBody(model=model, document=case.document))):
+            case Success(data=response):
+                _assert_ocr_document(response)
+            case RateLimitedError() as outcome:
+                _assert_provider_rate_limit_relayed(model, outcome)
+            case outcome:
+                pytest.fail(f"{model}: {outcome!r}")
 
     @pytest.mark.skip(reason="stage red: product gap, /v1/ocr 500s (aocr TypeError) on missing document instead of 400")
     @pytest.mark.covers("llm.ocr.openai.input_validation.nonstream.works")
-    def test_missing_document_returns_error(
-        self, proxy: ProxyClient, resources: ResourceManager
-    ) -> None:
+    def test_missing_document_returns_error(self, proxy: ProxyClient, resources: ResourceManager) -> None:
         model = f"rust-ocr-val-{unique_marker()}"
         model_id = proxy.create_model(model, MistralOcr().litellm_params())
         resources.defer(lambda: proxy.delete_model(model_id))
@@ -174,4 +198,3 @@ class TestRustOcrGateway:
             json=_OptionalOcrBody(model=model),
         )
         assert_client_error(result, "ocr missing document")
-

tests/e2e/router/reliability_support.py

@@ -15,9 +15,10 @@ reliability behavior.
 
 from __future__ import annotations
 
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
+from typing import Final
 
-from pydantic import ValidationError
+from pydantic import BaseModel, ValidationError
 
 from proxy_client import ProxyClient
 from e2e_config import CHEAP_OPENAI_MODEL, PROXY_BASE_URL, unique_marker
@@ -367,6 +368,26 @@ def model_id_of(resp: StreamingResponse) -> str | None:
     return resp.headers.get("x-litellm-model-id")
 
 
+class _AzurePromptFilterResult(BaseModel):
+    content_filter_results: Mapping[str, object] | None = None
+
+
+class _AzureAnnotatedChatBody(BaseModel):
+    prompt_filter_results: Sequence[_AzurePromptFilterResult] | None = None
+
+
+def azure_prompt_filter_skipped(resp: StreamingResponse) -> bool:
+    """True when Azure's 200 recorded no prompt-filter verdict (every `content_filter_results`
+    empty), so the prompt has to be sent again."""
+    try:
+        annotated: Final = _AzureAnnotatedChatBody.model_validate_json(resp.body)
+    except ValidationError:
+        return False
+    if not annotated.prompt_filter_results:
+        return False
+    return all(not entry.content_filter_results for entry in annotated.prompt_filter_results)
+
+
 def _parsed(resp: StreamingResponse) -> ChatResponse | None:
     try:
         return ChatResponse.model_validate_json(resp.body)

tests/e2e/router/test_reliability_fallbacks_e2e.py

@@ -16,10 +16,16 @@ failure: the provider refuses the prompt itself, on length or on policy, and
 reroute those, not `fallbacks`. The policy refusal is a real one, from an Azure
 OpenAI content filter rejecting a jailbreak prompt, and a control call first
 proves the refusal reaches the customer as a 400 when no reroute is configured.
+Azure intermittently answers without running its prompt filter at all (the
+body's prompt_filter_results carry no verdict), which is not a pass, so both
+calls send the prompt again, a bounded number of times, until the filter ran.
 """
 
 from __future__ import annotations
 
+from collections.abc import Callable
+from typing import Final
+
 import pytest
 
 from complexity_router_client import ComplexityRouterClient
@@ -29,6 +35,7 @@ from lifecycle import ResourceManager
 from models import RouterSettingsOverride
 from reliability_support import (
     CONTENT_POLICY_PROMPT,
+    azure_prompt_filter_skipped,
     chat_override,
     completion_tokens_of,
     content_of,
@@ -64,6 +71,28 @@ def _assert_served_by_fallback(resp: StreamingResponse) -> None:
     assert int(attempted) >= 1, f"x-litellm-attempted-fallbacks should be >= 1, got {attempted!r}"
 
 
+AZURE_FILTER_ATTEMPTS: Final = 3
+
+
+def _chat_once_azure_runs_its_filter(send: Callable[[], StreamingResponse]) -> StreamingResponse:
+    for attempt in range(1, AZURE_FILTER_ATTEMPTS):
+        resp = send()
+        if not azure_prompt_filter_skipped(resp):
+            return resp
+        print(
+            "e2e: azure answered without running its prompt filter; sending the jailbreak prompt again "
+            f"({attempt}/{AZURE_FILTER_ATTEMPTS - 1})",
+            flush=True,
+        )
+    return send()
+
+
+def _filter_verdict(resp: StreamingResponse) -> str:
+    if azure_prompt_filter_skipped(resp):
+        return "azure skipped its prompt filter on every attempt"
+    return "the filter ran and let the prompt through"
+
+
 class TestReliabilityFallbacks:
     @pytest.mark.covers("reliability.fallback.5xx.routes_to_fallback")
     def test_5xx_routes_to_fallback(
@@ -124,17 +153,21 @@ class TestReliabilityFallbacks:
         model_id = create_content_filtered_deployment(client.proxy, primary)
         resources.defer(lambda: client.proxy.delete_model(model_id))
 
-        refused = chat_override(client.proxy, scoped_key, primary, f"{CONTENT_POLICY_PROMPT} {unique_marker()}")
+        refused = _chat_once_azure_runs_its_filter(
+            lambda: chat_override(client.proxy, scoped_key, primary, f"{CONTENT_POLICY_PROMPT} {unique_marker()}")
+        )
         assert refused.status_code == 400, (
-            f"the content filter should have refused the jailbreak prompt with a 400, got {refused.status_code}: "
-            f"{refused.body[:300]}"
+            f"the content filter should have refused the jailbreak prompt with a 400, got {refused.status_code} "
+            f"({_filter_verdict(refused)}): {refused.body[:300]}"
         )
 
-        resp = chat_override(
-            client.proxy,
-            scoped_key,
-            primary,
-            f"{CONTENT_POLICY_PROMPT} {unique_marker()}",
-            override=RouterSettingsOverride(content_policy_fallbacks=[{primary: ["gpt-5.5"]}]),
+        resp = _chat_once_azure_runs_its_filter(
+            lambda: chat_override(
+                client.proxy,
+                scoped_key,
+                primary,
+                f"{CONTENT_POLICY_PROMPT} {unique_marker()}",
+                override=RouterSettingsOverride(content_policy_fallbacks=[{primary: ["gpt-5.5"]}]),
+            )
         )
         _assert_served_by_fallback(resp)