From 4587dbeae98b798be6776e5b2a47d5eebea50705 Mon Sep 17 00:00:00 2001 From: Yuneng Jiang Date: Tue, 15 Sep 2026 18:14:12 -0700 Subject: [PATCH 1/6] feat(e2e): cache exact provider responses for 24 hours --- .github/workflows/test-provider-cache.yml | 63 +++ .../test_provider_cache.py | 409 ++++++++++++++++++ tests/e2e/PROVIDER_CACHE.md | 27 ++ tests/e2e/conftest.py | 6 +- tests/e2e/e2e_http.py | 41 +- .../e2e/llm_translation/test_messages_e2e.py | 3 +- .../llm_translation/test_together_ai_e2e.py | 1 + tests/e2e/provider_cache.py | 297 +++++++++++++ tests/e2e/provider_cache_redis.py | 154 +++++++ tests/e2e/provider_cache_routing.py | 23 + tests/e2e/provider_edge.py | 61 ++- tests/e2e/proxy_client.py | 8 +- 12 files changed, 1077 insertions(+), 16 deletions(-) create mode 100644 .github/workflows/test-provider-cache.yml create mode 100644 tests/code_coverage_tests/test_provider_cache.py create mode 100644 tests/e2e/PROVIDER_CACHE.md create mode 100644 tests/e2e/provider_cache.py create mode 100644 tests/e2e/provider_cache_redis.py create mode 100644 tests/e2e/provider_cache_routing.py diff --git a/.github/workflows/test-provider-cache.yml b/.github/workflows/test-provider-cache.yml new file mode 100644 index 00000000000..29b9d08a267 --- /dev/null +++ b/.github/workflows/test-provider-cache.yml @@ -0,0 +1,63 @@ +name: Provider cache contracts + +on: + pull_request: + paths: + - 'tests/e2e/**' + - 'tests/code_coverage_tests/test_provider_cache.py' + - 'tests/code_coverage_tests/test_provider_replay_harness.py' + - '.github/workflows/test-provider-cache.yml' + - 'pyproject.toml' + - 'uv.lock' + workflow_dispatch: + +permissions: + contents: read + +jobs: + provider-cache: + runs-on: ubuntu-latest + timeout-minutes: 20 + services: + redis: + image: redis@sha256:e2debfb7956fa12c7ddc79d7e645c8cf26b30c99a6e9161ea9bf4171e1668a5f + ports: + - 6379:6379 + options: >- + --health-cmd 'redis-cli ping' + --health-interval 5s + --health-timeout 3s + --health-retries 10 + env: + PYTHONDONTWRITEBYTECODE: '1' + PYTHONPATH: tests/e2e + E2E_CACHE_TEST_REDIS_URL: redis://127.0.0.1:6379/0 + E2E_PROVIDER_CACHE: '0' + E2E_FIXTURE_MODE: live + steps: + - uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 + with: + persist-credentials: false + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 + with: + python-version: '3.13' + - uses: ./.github/actions/setup-uv-with-retries + with: + version: '0.10.9' + - uses: ./.github/actions/cache-cargo-build + - name: Install locked dependencies + run: .github/scripts/uv_sync_with_retries.sh --frozen --group proxy-dev --group e2e-dev + - name: Verify cache and existing edge contracts + run: >- + uv run --no-sync pytest -c /dev/null -p no:cacheprovider + tests/code_coverage_tests/test_provider_cache.py + tests/code_coverage_tests/test_provider_replay_harness.py + tests/e2e/test_provider_edge.py + -q --junitxml=provider-cache-results.xml + - name: Save test results + if: always() + uses: actions/upload-artifact@4cec3d8aa04e39d1a68397de0c4cd6fb9dce8ec1 + with: + name: provider-cache-results + path: provider-cache-results.xml + if-no-files-found: warn diff --git a/tests/code_coverage_tests/test_provider_cache.py b/tests/code_coverage_tests/test_provider_cache.py new file mode 100644 index 00000000000..7edca8151af --- /dev/null +++ b/tests/code_coverage_tests/test_provider_cache.py @@ -0,0 +1,409 @@ +from __future__ import annotations + +import os +import shutil +import socket +import subprocess +import threading +import time +import uuid +from collections.abc import Generator +from concurrent.futures import ThreadPoolExecutor +from contextlib import contextmanager +from dataclasses import dataclass, replace +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Final + +import pytest +from e2e_http import NetworkError, PreparedForward, RawResponse, StreamChunk, StreamHead, forward, prepare_forward +from models import LiteLLMParamsBody +from provider_cache import CacheEdge, CacheHit, CaptureLease, exact_key, successful_response +from provider_cache_redis import PUBLISH, RedisCommands, RedisResponseStore, configured_cache, redis_store +from provider_cache_routing import LIVE_PROVIDER_REQUIRED, route_cache_model +from provider_edge import configured_cache_backend, start_provider_edge +from redis.exceptions import ConnectionError as RedisConnectionError + +SECRET: Final = b"synthetic-cache-hmac-key-for-tests" +BODY: Final = b'{"model":"test","messages":[{"role":"user","content":"hello"}]}' +SUCCESS: Final = b'{"id":"provider-fixed-id","choices":[{"message":{"content":"hello"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2}}' +HEADERS: Final = {"content-type": "application/json", "authorization": "Bearer synthetic-account-one"} + + +class Provider(ThreadingHTTPServer): + hits: tuple[tuple[str, bytes], ...] = () + response: bytes = SUCCESS + status: int = 200 + delay: float = 0 + stream: bool = False + truncated: bool = False + + +class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def do_POST(self) -> None: + server: Final = self.server + assert isinstance(server, Provider) + body: Final = self.rfile.read(int(self.headers.get("content-length", "0"))) + server.hits += ((self.path, body),) + time.sleep(server.delay) + self.send_response(server.status) + if server.stream: + self.send_header("content-type", "text/event-stream") + self.send_header("transfer-encoding", "chunked") + self.end_headers() + self.wfile.write(b"%x\r\n%s\r\n" % (len(server.response), server.response)) + if server.truncated: + self.close_connection = True + return + self.wfile.write(b"0\r\n\r\n") + return + self.send_header("content-type", "application/json") + self.send_header("content-length", str(len(server.response))) + self.end_headers() + self.wfile.write(server.response) + + def log_message(self, format: str, *args: object) -> None: + pass + + +@pytest.fixture +def provider() -> Generator[Provider, None, None]: + server: Final = Provider(("127.0.0.1", 0), Handler) + thread: Final = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield server + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +@pytest.fixture(scope="module") +def redis_url(tmp_path_factory: pytest.TempPathFactory) -> Generator[str, None, None]: + configured: Final = os.environ.get("E2E_CACHE_TEST_REDIS_URL") + if configured: + yield configured + return + binary: Final = shutil.which("redis-server") + assert binary is not None, "Set E2E_CACHE_TEST_REDIS_URL or install Redis for cache integration checks" + root: Final = tmp_path_factory.mktemp("provider-cache-redis") + with socket.socket() as probe: + probe.bind(("127.0.0.1", 0)) + port: Final = probe.getsockname()[1] + with (root / "redis.log").open("wb") as log: + process: Final = subprocess.Popen( + [binary, "--bind", "127.0.0.1", "--port", str(port), "--save", "", "--appendonly", "no", "--dir", str(root)], + stdout=log, stderr=subprocess.STDOUT, + ) + try: + deadline: Final = time.monotonic() + 5 + while True: + try: + with socket.create_connection(("127.0.0.1", port), timeout=0.1): + break + except OSError: + assert process.poll() is None and time.monotonic() < deadline + time.sleep(0.02) + yield f"redis://127.0.0.1:{port}/0" + finally: + process.terminate() + process.wait(timeout=5) + + +@pytest.fixture +def store(redis_url: str) -> RedisResponseStore: + return redis_store(redis_url, "test-" + uuid.uuid4().hex) + + +@contextmanager +def edge(cache: CacheEdge, provider: Provider) -> Generator[str, None, None]: + upstream: Final = f"http://127.0.0.1:{provider.server_port}" + running: Final = start_provider_edge(cache, mounts={"openai": upstream}) + try: + yield running.edge.api_base("openai") + "/v1/chat/completions" + finally: + running.shutdown() + + +def call(url: str, body: bytes = BODY, headers: dict[str, str] = HEADERS) -> RawResponse: + result: Final = forward("POST", url, headers=headers, body=body, timeout=5) + assert isinstance(result, RawResponse), result + return result + + +def test_success_is_reusable_across_fresh_edges(store: RedisResponseStore, provider: Provider) -> None: + with edge(CacheEdge(store, SECRET), provider) as url: + assert call(url).body == SUCCESS + assert call(url).body == SUCCESS + with edge(CacheEdge(store, SECRET), provider) as other: + assert call(other).body == SUCCESS + assert len(provider.hits) == 1 + + +@pytest.mark.parametrize("body", [BODY + b" ", BODY.replace(b"hello", b"Hello"), BODY.replace(b"test", b"test2")]) +def test_any_body_change_calls_live(store: RedisResponseStore, provider: Provider, body: bytes) -> None: + with edge(CacheEdge(store, SECRET), provider) as url: + call(url) + call(url, body) + call(url, body) + assert len(provider.hits) == 2 + + +@pytest.mark.parametrize("name,value", [("authorization", "Bearer another-account"), ("x-request-id", "one"), ("anthropic-version", "new")]) +def test_changed_header_cannot_reuse(store: RedisResponseStore, provider: Provider, name: str, value: str) -> None: + with edge(CacheEdge(store, SECRET), provider) as url: + call(url) + call(url, headers=HEADERS | {name: value}) + call(url + "?x=1") + assert len(provider.hits) == 3 + + +@pytest.mark.parametrize("status,response", [(429, b'{"error":"rate limited"}'), (500, b'failed'), (200, b'{"error":"bad"}'), (200, b'not json')]) +def test_failed_provider_responses_never_enter_cache(store: RedisResponseStore, provider: Provider, status: int, response: bytes) -> None: + provider.status = status + provider.response = response + with edge(CacheEdge(store, SECRET), provider) as url: + assert call(url).status_code == status + assert call(url).body == response + assert len(provider.hits) == 2 + + +def test_expiry_does_not_slide(store: RedisResponseStore, provider: Provider) -> None: + short: Final = replace(store, lifetime_ms=250) + with edge(CacheEdge(short, SECRET), provider) as url: + call(url) + call(url) + time.sleep(0.3) + call(url) + call(url) + assert len(provider.hits) == 2 + + +def test_concurrent_requests_publish_atomically(store: RedisResponseStore, provider: Provider) -> None: + provider.delay = 0.15 + with edge(CacheEdge(store, SECRET), provider) as url: + with ThreadPoolExecutor(max_workers=5) as executor: + replies: Final = tuple(executor.map(lambda _: call(url).body, range(5))) + assert replies == (SUCCESS,) * 5 + assert len(provider.hits) == 1 + + +@pytest.mark.parametrize("truncated", [False, True]) +def test_stream_completion_controls_publication(store: RedisResponseStore, provider: Provider, truncated: bool) -> None: + provider.stream = True + provider.truncated = truncated + provider.response = b'data: {"choices":[{"index":0,"delta":{"content":"hello"},"finish_reason":"stop"}]}\n\ndata: [DONE]\n\n' + with edge(CacheEdge(store, SECRET), provider) as url: + for _ in range(2): + result: Final = forward("POST", url, headers=HEADERS, body=BODY, timeout=5) + if truncated: + assert isinstance(result, NetworkError) + else: + assert isinstance(result, RawResponse) and result.body == provider.response + assert len(provider.hits) == (2 if truncated else 1) + + +def test_store_outage_preserves_provider_success(provider: Provider) -> None: + with socket.socket() as probe: + probe.bind(("127.0.0.1", 0)) + port: Final = probe.getsockname()[1] + unavailable: Final = redis_store(f"redis://127.0.0.1:{port}/0", "unavailable") + with edge(CacheEdge(unavailable, SECRET), provider) as url: + assert call(url).body == SUCCESS + assert call(url).body == SUCCESS + assert len(provider.hits) == 2 + + +def test_old_lease_cannot_overwrite_new_owner(store: RedisResponseStore) -> None: + short: Final = replace(store, lease_ms=50) + old: Final = short.lookup("key") + assert isinstance(old, CaptureLease) + time.sleep(0.08) + current: Final = short.lookup("key") + assert isinstance(current, CaptureLease) + assert not short.publish("key", old, b"old") + assert short.publish("key", current, b"new") + hit: Final = short.lookup("key") + assert isinstance(hit, CacheHit) and hit.payload == b"new" + + +def test_identity_preserves_values_and_never_contains_credentials() -> None: + variants: Final = (b'{}', b'{"a":null}', b'{"a":false}', b'{"a":0}', b'{"a":0.0}', b'{"a":"0"}', b' { }', None, b'') + keys: Final = tuple(exact_key(SECRET, "POST", "https://example.invalid/v1/chat/completions", HEADERS, body) for body in variants) + assert len(set(keys)) == len(variants) + assert all(len(key) == 64 and "synthetic-account" not in key for key in keys) + + +@pytest.mark.parametrize("payload", [b"corrupt response", '{"response":"{}","signature":"é"}'.encode()]) +def test_corrupt_entry_is_replaced_by_same_successful_request(store: RedisResponseStore, provider: Provider, payload: bytes) -> None: + upstream: Final = f"http://127.0.0.1:{provider.server_port}/v1/chat/completions" + prepared: Final = prepare_forward("POST", upstream, HEADERS, BODY) + assert isinstance(prepared, PreparedForward) + key: Final = exact_key(SECRET, "POST", upstream, prepared.headers, BODY) + lease: Final = store.lookup(key) + assert isinstance(lease, CaptureLease) + assert store.publish(key, lease, payload) + cache: Final = CacheEdge(store, SECRET) + for _ in range(2): + head = cache.forward("POST", upstream, HEADERS, BODY, 5) + assert isinstance(head, StreamHead) + assert b"".join(step.data for step in head.steps if isinstance(step, StreamChunk)) == SUCCESS + assert len(provider.hits) == 1 + assert dict(cache.counters.counts) == { + "corrupt": 1, "misses": 1, "upstream_attempts": 1, "writes": 1, "hits": 1, + } + + +@pytest.mark.parametrize("payload", [ + b'data: {}\n\ndata: [DONE]\n\n', + b'data: {"choices":[{"index":0,"delta":{}}]}\n\ndata: [DONE]\n\n', + b'data: {"choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}\n\ndata: [DONE]', + b'data: {"error":{"message":"failed"}}\n\ndata: [DONE]\n\n', +]) +def test_malformed_success_stream_is_never_cached(store: RedisResponseStore, provider: Provider, payload: bytes) -> None: + provider.stream = True + provider.response = payload + with edge(CacheEdge(store, SECRET), provider) as url: + assert call(url).body == payload + assert call(url).body == payload + assert len(provider.hits) == 2 + + +def test_anthropic_stream_requires_start_finish_and_stop() -> None: + start: Final = b'data: {"type":"message_start","message":{}}\n\n' + finish: Final = b'data: {"type":"message_delta","delta":{"stop_reason":"end_turn"}}\n\n' + stop: Final = b'data: {"type":"message_stop"}\n\n' + url: Final = "https://example.invalid/v1/messages" + headers: Final = {"content-type": "text/event-stream"} + assert successful_response(url, 200, headers, start + finish + stop) + assert not successful_response(url, 200, headers, start + stop) + assert not successful_response(url, 200, headers, finish + stop) + assert not successful_response(url, 200, headers, start + finish) + + +@pytest.mark.parametrize("provider,suffix", [("openai", "/v1"), ("anthropic", "")]) +def test_normal_registration_routes_supported_providers(provider: str, suffix: str) -> None: + params: Final = LiteLLMParamsBody(model=f"{provider}/test", api_key="os.environ/SYNTHETIC_KEY", timeout=12) + routed: Final = route_cache_model(params, lambda mount: f"http://edge.invalid/{mount}", enabled=True) + assert routed.api_base == f"http://edge.invalid/{provider}{suffix}" + assert routed.model_dump(exclude={"api_base"}) == params.model_dump(exclude={"api_base"}) + assert params.api_base is None + + +@pytest.mark.parametrize("params", [ + LiteLLMParamsBody(model="bedrock/test"), + LiteLLMParamsBody(model="azure/test"), + LiteLLMParamsBody(model="openai/test", api_base="https://custom.invalid/v1"), + LiteLLMParamsBody(model="openai/test", api_base=""), + LiteLLMParamsBody(model="openai/test", litellm_credential_name="named-credential"), + LiteLLMParamsBody(model="openai/test", mock_response="synthetic"), +]) +def test_registration_preserves_unsupported_or_explicit_routes(params: LiteLLMParamsBody) -> None: + def unexpected_edge(mount: str) -> str: + pytest.fail(f"should not start edge for {mount}") + assert route_cache_model(params, unexpected_edge, enabled=True) is params + + +def test_rollback_and_live_only_policy_keep_direct_provider_route() -> None: + params: Final = LiteLLMParamsBody(model="openai/test") + assert route_cache_model(params, lambda _: "http://edge.invalid", enabled=False) is params + assert route_cache_model(params, lambda _: "http://edge.invalid", enabled=True, mode="realtime") is params + token: Final = LIVE_PROVIDER_REQUIRED.set(True) + try: + assert route_cache_model(params, lambda _: "http://edge.invalid", enabled=True) is params + finally: + LIVE_PROVIDER_REQUIRED.reset(token) + assert route_cache_model(params, lambda _: "http://edge.invalid", enabled=True).api_base == "http://edge.invalid/v1" + + +@dataclass(frozen=True) +class PublishOutage: + client: RedisCommands + + def eval(self, script: str, numkeys: int, *args: str | bytes | int) -> object: + if script == PUBLISH: + raise RedisConnectionError("synthetic publication outage") + return self.client.eval(script, numkeys, *args) + + +def test_write_outage_preserves_success_without_hidden_retry(store: RedisResponseStore, provider: Provider) -> None: + unavailable: Final = replace(store, client=PublishOutage(store.client)) + cache: Final = CacheEdge(unavailable, SECRET) + with edge(cache, provider) as url: + assert call(url).body == SUCCESS + assert call(url).body == SUCCESS + assert len(provider.hits) == 2 + assert dict(cache.counters.counts)["write_failures"] == 2 + with edge(CacheEdge(store, SECRET), provider) as url: + assert call(url).body == SUCCESS + assert call(url).body == SUCCESS + assert len(provider.hits) == 3 + + +def test_connection_failure_releases_capture_lease(store: RedisResponseStore) -> None: + with socket.socket() as unavailable: + unavailable.bind(("127.0.0.1", 0)) + url: Final = f"http://127.0.0.1:{unavailable.getsockname()[1]}/v1/chat/completions" + cache: Final = CacheEdge(store, SECRET) + assert isinstance(cache.forward("POST", url, HEADERS, BODY, 0.2), NetworkError) + prepared: Final = prepare_forward("POST", url, HEADERS, BODY) + assert isinstance(prepared, PreparedForward) + key: Final = exact_key(SECRET, "POST", url, prepared.headers, BODY) + slot: Final = store.lookup(key) + assert isinstance(slot, CaptureLease) + assert store.release(key, slot) + assert dict(cache.counters.counts)["rejected"] == 1 + + +def test_close_before_first_chunk_releases_lease(store: RedisResponseStore, provider: Provider) -> None: + url: Final = f"http://127.0.0.1:{provider.server_port}/v1/chat/completions" + cache: Final = CacheEdge(store, SECRET) + head: Final = cache.forward("POST", url, HEADERS, BODY, 5) + assert isinstance(head, StreamHead) + head.steps.close() + prepared: Final = prepare_forward("POST", url, HEADERS, BODY) + assert isinstance(prepared, PreparedForward) + key: Final = exact_key(SECRET, "POST", url, prepared.headers, BODY) + slot: Final = store.lookup(key) + assert isinstance(slot, CaptureLease) + assert store.release(key, slot) + + +def test_effective_account_change_cannot_reuse_cache( + store: RedisResponseStore, provider: Provider, monkeypatch: pytest.MonkeyPatch, tmp_path, +) -> None: + url: Final = f"http://127.0.0.1:{provider.server_port}/v1/chat/completions" + cache: Final = CacheEdge(store, SECRET) + for account in ("account-a", "account-b", "account-b"): + netrc = tmp_path / account + netrc.write_text(f"machine 127.0.0.1 login {account} password synthetic\n") + monkeypatch.setenv("NETRC", str(netrc)) + head = cache.forward("POST", url, HEADERS, BODY, 5) + assert isinstance(head, StreamHead) + assert b"".join(step.data for step in head.steps if isinstance(step, StreamChunk)) == SUCCESS + assert len(provider.hits) == 2 + assert dict(cache.counters.counts)["hits"] == 1 + + +def test_enabled_environment_reuses_store_across_fresh_backends( + redis_url: str, provider: Provider, monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("E2E_PROVIDER_CACHE", "1") + monkeypatch.setenv("E2E_PROVIDER_CACHE_REDIS_URL", redis_url) + monkeypatch.setenv("E2E_PROVIDER_CACHE_HMAC_KEY", SECRET.decode()) + monkeypatch.setenv("E2E_PROVIDER_CACHE_NAMESPACE", "environment-" + uuid.uuid4().hex) + configured_cache.cache_clear() + try: + for _ in range(2): + backend = configured_cache_backend() + assert isinstance(backend, CacheEdge) + with edge(backend, provider) as url: + assert call(url).body == SUCCESS + configured_cache.cache_clear() + assert len(provider.hits) == 1 + monkeypatch.setenv("E2E_PROVIDER_CACHE", "0") + assert configured_cache_backend() is None + finally: + configured_cache.cache_clear() diff --git a/tests/e2e/PROVIDER_CACHE.md b/tests/e2e/PROVIDER_CACHE.md new file mode 100644 index 00000000000..280132d0940 --- /dev/null +++ b/tests/e2e/PROVIDER_CACHE.md @@ -0,0 +1,27 @@ +# Shared provider-response cache + +`E2E_PROVIDER_CACHE=1` enables automatic response reuse in the live E2E mode. Standard OpenAI and Anthropic model registrations use the provider edge. Existing custom API bases, named credentials, mocked models and realtime WebSocket deployments keep their existing routing. Other provider protocols remain live + +The edge caches complete successful POST responses for `/v1/chat/completions` and `/v1/messages`, including streams. Unsupported endpoints pass through. It matches the method, original URL, effective outbound headers (including authentication and HTTP-library defaults), body presence and exact body bytes using a full keyed digest. It sends the same prepared request used for matching. No prompts, random markers, JSON values or credentials are normalized away + +An eligible miss calls the provider. A complete successful response is stored immediately even if a later test assertion fails. Provider errors, malformed responses, truncated streams and cancelled captures are not stored. Cache reads, writes and lease failures fall through to normal provider behavior; they introduce no provider retry. An already-started response cannot be restarted after a delivery failure + +Recordings are shared across workers and builds through dedicated Redis, separate from the candidate's own cache. They expire 86,400 seconds after capture starts, based on Redis time. Reads never extend expiry. There is no scheduled recapture: the next miss calls the provider again. Bounded coordination reduces duplicate concurrent calls, but slow or failed captures may lead to extra live calls after the wait expires + +## Configuration + +The trusted runner receives: + +- `E2E_PROVIDER_CACHE`: `1` to enable, `0` to use the normal live path +- `E2E_PROVIDER_CACHE_REDIS_URL`: authenticated dedicated Redis URL +- `E2E_PROVIDER_CACHE_HMAC_KEY`: dedicated secret containing at least 32 bytes +- `E2E_PROVIDER_CACHE_NAMESPACE`: shared environment namespace, independent of build and candidate revision +- `E2E_PROVIDER_CACHE_METRICS_DIR`: optional per-process counter artifact directory + +Do not give cache credentials to candidate deployments. Counter artifacts contain no recorded payloads or credentials. Hits count shared-cache responses; upstream attempts count actual forwards from the edge. Existing application-cache observations still count requests arriving at the edge, including shared-cache hits + +Tests that require real provider timing, limits or state use `@pytest.mark.provider_live`. The marker keeps newly registered models on live routes without weakening their assertions. Ordinary assertion failures still fail E2E. The shared cache does not modify provider response IDs or make the proxy aware of replay + +## Qualification + +`tests/code_coverage_tests/test_provider_cache.py` exercises local HTTP providers and disposable real Redis. CI runs these checks with the existing provider-edge and replay harness tests. These component checks do not establish Buildkite deployment, full-suite cross-build reuse or a genuine 24-hour expiry observation; those require separate runtime evidence diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py index b1a75d5f862..829c84910a9 100644 --- a/tests/e2e/conftest.py +++ b/tests/e2e/conftest.py @@ -22,7 +22,6 @@ from typing import Final import pytest import requests - from e2e_config import ( CONTROL_PLANE_BASE_URL, FIXTURE_DIR, @@ -41,6 +40,7 @@ from idp import Identity, Keycloak, keycloak_from_env from junit_properties import attach_result_properties from lifecycle import ProxyClientProvider, ResourceManager from models import TeamNewBody, UserNewBody, UserNewResponse +from provider_cache_routing import LIVE_PROVIDER_REQUIRED from provider_edge import replay_leftover_error from proxy_client import ProxyClient, build_proxy_client @@ -85,6 +85,7 @@ def jwt_identity(idp: Keycloak, resources: ResourceManager, proxy: ProxyClient) def pytest_configure(config: pytest.Config) -> None: + config.addinivalue_line("markers", "provider_live: requires actual provider timing, limits or state; bypass shared cache") config.addinivalue_line( "markers", "e2e: live test that requires a running proxy and real provider keys", @@ -192,11 +193,13 @@ def _proxy_fail_reason() -> str | None: return None +@pytest.hookimpl(tryfirst=True) def pytest_runtest_setup(item: pytest.Item) -> None: """Hard-fail `e2e`-marked tests unless a proxy answers its liveness probe. Unmarked tests (unit coverage of the harness) don't touch the proxy, so they run even when none is up. Never skip for a missing proxy. Replay mode needs the proxy too: only provider-bound traffic replays from the bundle.""" + LIVE_PROVIDER_REQUIRED.set(item.get_closest_marker("provider_live") is not None) if item.get_closest_marker("e2e") is None: return reason = _proxy_fail_reason() @@ -235,6 +238,7 @@ def pytest_runtest_teardown(item: pytest.Item) -> Generator[None, None, None]: yield so fixture finalizers replay their recorded calls first. Failed tests are left alone - their own failure already explains any unconsumed tail.""" result = yield + LIVE_PROVIDER_REQUIRED.set(False) if not item.stash.get(_CALL_PASSED, False): return result reason = replay_leftover_error( diff --git a/tests/e2e/e2e_http.py b/tests/e2e/e2e_http.py index 1992f419823..4184b6cbefc 100644 --- a/tests/e2e/e2e_http.py +++ b/tests/e2e/e2e_http.py @@ -853,6 +853,7 @@ def _stream_steps(resp: requests.Response) -> Generator[StreamStep, None, None]: the chunks already delivered are exactly what makes a mid-stream failure different from a request that never streamed at all.""" try: + yield StreamChunk(b"") for piece in cast("Iterator[bytes]", resp.iter_content(chunk_size=None)): if piece: yield StreamChunk(data=piece) @@ -862,6 +863,44 @@ def _stream_steps(resp: requests.Response) -> Generator[StreamStep, None, None]: resp.close() +def primed_steps(steps: Generator[StreamStep, None, None]) -> Generator[StreamStep, None, None]: + first: Final = next(steps) + assert isinstance(first, StreamChunk) and first.data == b"" + return steps + + +@dataclass(frozen=True, slots=True, repr=False) +class PreparedForward: + request: requests.PreparedRequest + url: str + headers: dict[str, str] + + +def prepare_forward( + method: str, url: str, headers: dict[str, str], body: bytes | None, +) -> PreparedForward | NetworkError: + try: + with requests.Session() as session: + request: Final = session.prepare_request(requests.Request(method, url, headers=headers, data=body)) + except requests.RequestException as exc: + return NetworkError(message=str(exc)) + assert request.url is not None + return PreparedForward(request, request.url, dict(request.headers)) + + +def forward_prepared_stream(prepared: PreparedForward, timeout: float) -> StreamHead | NetworkError: + try: + with requests.Session() as session: + settings: Final = session.merge_environment_settings(prepared.url, {}, True, None, None) + resp: Final = session.send(prepared.request, timeout=timeout, allow_redirects=False, **settings) + 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()}, + primed_steps(_stream_steps(resp)), + ) + + def open_stream(url: URL, *, headers: BaseModel, json: BaseModel, timeout: float = 60.0) -> StreamHead | NetworkError: """POST a streaming request and return the moment its response head arrives, leaving the body unread behind ``StreamHead.steps``. For a test that must keep @@ -907,5 +946,5 @@ def forward_stream( return StreamHead( status_code=resp.status_code, headers={name.lower(): value for name, value in resp.headers.items()}, - steps=_stream_steps(resp), + steps=primed_steps(_stream_steps(resp)), ) diff --git a/tests/e2e/llm_translation/test_messages_e2e.py b/tests/e2e/llm_translation/test_messages_e2e.py index ca58c30d40c..44c416a3e78 100644 --- a/tests/e2e/llm_translation/test_messages_e2e.py +++ b/tests/e2e/llm_translation/test_messages_e2e.py @@ -24,10 +24,10 @@ from models import ( AnthropicAssistantTurn, AnthropicContentBlock, AnthropicCustomTool, + AnthropicMessagesBody, AnthropicToolChoice, AnthropicToolResultBlock, AnthropicToolResultTurn, - AnthropicMessagesBody, ChatMessage, JsonSchemaProperty, LiteLLMParamsBody, @@ -165,6 +165,7 @@ class TestAnthropicMessages: ) @pytest.mark.covers("llm.messages.anthropic.basic.stream.works") + @pytest.mark.provider_live def test_messages_streams_completion( self, endpoints_client: EndpointsClient, resources: ResourceManager ) -> None: diff --git a/tests/e2e/llm_translation/test_together_ai_e2e.py b/tests/e2e/llm_translation/test_together_ai_e2e.py index 31e74c22e17..8dd7e7c1a31 100644 --- a/tests/e2e/llm_translation/test_together_ai_e2e.py +++ b/tests/e2e/llm_translation/test_together_ai_e2e.py @@ -744,6 +744,7 @@ class TestTogetherMessages: assert "22" in text, f"the model never saw the tool result: {response.content}" @pytest.mark.covers("llm.messages.together_ai.basic.stream.works") + @pytest.mark.provider_live def test_streams_text_deltas( self, client: PassthroughClient, resources: ResourceManager, reasoning_tool_backend: str ) -> None: diff --git a/tests/e2e/provider_cache.py b/tests/e2e/provider_cache.py new file mode 100644 index 00000000000..50b336f5263 --- /dev/null +++ b/tests/e2e/provider_cache.py @@ -0,0 +1,297 @@ +from __future__ import annotations + +import base64 +import hashlib +import hmac +import io +import threading +import time +from collections.abc import Callable, Generator, Mapping +from contextlib import closing +from dataclasses import dataclass, field +from typing import Final, Literal, Protocol +from urllib.parse import urlsplit + +from e2e_http import ( + NetworkError, + StreamChunk, + StreamHead, + StreamStep, + StreamTruncation, + forward_prepared_stream, + forward_stream, + prepare_forward, + primed_steps, +) +from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter, ValidationError + +LIFETIME_SECONDS: Final = 86_400 +MAX_REQUEST_BYTES: Final = 256 * 1024 +MAX_RESPONSE_BYTES: Final = 8 * 1024 * 1024 +JSON_VALUE: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue) + + +@dataclass(frozen=True, slots=True) +class CacheHit: + payload: bytes + valid_until: float + + +@dataclass(frozen=True, slots=True) +class CaptureLease: + token: str + captured_at_ms: int + expires_at_ms: int + + +@dataclass(frozen=True, slots=True) +class CacheBusy: + pass + + +@dataclass(frozen=True, slots=True) +class CacheUnavailable: + pass + + +type CacheLookup = CacheHit | CaptureLease | CacheBusy | CacheUnavailable + + +class ResponseStore(Protocol): + def lookup(self, key: str) -> CacheLookup: ... + + def publish(self, key: str, lease: CaptureLease, payload: bytes) -> bool: ... + + def release(self, key: str, lease: CaptureLease) -> bool: ... + + def discard(self, key: str, payload: bytes) -> bool: ... + + +class CachedResponse(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid", strict=True) + format_version: Literal[1] = 1 + request_key: str + status_code: int + headers: dict[str, str] + chunks: tuple[str, ...] + + +class SignedResponse(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid", strict=True) + response: str + signature: str + + +def exact_key(secret: bytes, method: str, url: str, headers: Mapping[str, str], body: bytes | None) -> str: + fields: Final = ( + b"provider-cache-exact-v1", method.encode(), url.encode(), + *(part.encode() for pair in sorted(headers.items()) for part in pair), + b"no-body" if body is None else b"body", b"" if body is None else body, + ) + encoded: Final = b"".join(len(part).to_bytes(8, "big") + part for part in fields) + return hmac.new(secret, encoded, hashlib.sha256).hexdigest() + + +def cacheable_endpoint(method: str, url: str, body: bytes | None) -> bool: + return ( + method == "POST" + and urlsplit(url).path in {"/v1/chat/completions", "/v1/messages"} + and body is not None + and len(body) <= MAX_REQUEST_BYTES + ) + + +def successful_response(url: str, status: int, headers: Mapping[str, str], body: bytes) -> bool: + if not 200 <= status < 300 or len(body) > MAX_RESPONSE_BYTES: + return False + if any(name.lower() == "set-cookie" for name in headers): + return False + streaming: Final = "text/event-stream" in headers.get("content-type", "").lower() + if streaming: + try: + text: Final = body.decode("utf-8").replace("\r\n", "\n") + if not text.endswith("\n\n"): + return False + events: Final = tuple( + "\n".join(line[5:].removeprefix(" ") for line in event.split("\n") if line.startswith("data:")) + for event in text.split("\n\n") if any(line.startswith("data:") for line in event.split("\n")) + ) + values: Final = tuple(JSON_VALUE.validate_json(event) for event in events if event != "[DONE]") + except (UnicodeDecodeError, ValidationError): + return False + if not values or any(not isinstance(value, dict) or "error" in value or value.get("type") == "error" for value in values): + return False + if urlsplit(url).path == "/v1/chat/completions": + return events[-1] == "[DONE]" and "[DONE]" not in events[:-1] and complete_chat_stream(values) + return ( + "[DONE]" not in events + and isinstance(values[0], dict) and values[0].get("type") == "message_start" + and isinstance(values[-1], dict) and values[-1].get("type") == "message_stop" + and any( + isinstance(value, dict) and value.get("type") == "message_delta" + and isinstance(delta := value.get("delta"), dict) and isinstance(delta.get("stop_reason"), str) + for value in values + ) + ) + try: + value: Final = JSON_VALUE.validate_json(body) + except ValidationError: + return False + if not isinstance(value, dict) or "error" in value: + return False + if urlsplit(url).path == "/v1/messages": + return value.get("type") == "message" and isinstance(value.get("content"), list) and isinstance(value.get("stop_reason"), str) + choices: Final = value.get("choices") + return isinstance(choices, list) and bool(choices) and all( + isinstance(choice, dict) and isinstance(choice.get("message"), dict) and isinstance(choice.get("finish_reason"), str) + for choice in choices + ) + + +def complete_chat_stream(values: tuple[JsonValue, ...]) -> bool: + if any(not isinstance(value, dict) or not isinstance(value.get("choices"), list) for value in values): + return False + choices: Final = tuple( + choice for value in values if isinstance(value, dict) + if isinstance(items := value.get("choices"), list) for choice in items + ) + if not choices or any( + not isinstance(choice, dict) or type(choice.get("index")) is not int + or not isinstance(choice.get("delta"), dict) + for choice in choices + ): + return False + indices: Final = frozenset(choice["index"] for choice in choices if isinstance(choice, dict)) + return all( + isinstance(tuple(choice for choice in choices if isinstance(choice, dict) and choice["index"] == index)[-1].get("finish_reason"), str) + for index in indices + ) + + +def encode_response(secret: bytes, response: CachedResponse) -> bytes: + raw: Final = response.model_dump_json() + return SignedResponse(response=raw, signature=hmac.new(secret, raw.encode(), hashlib.sha256).hexdigest()).model_dump_json().encode() + + +def decode_response(secret: bytes, key: str, payload: bytes, url: str) -> CachedResponse | None: + if len(payload) > 2 * MAX_RESPONSE_BYTES: + return None + try: + signed: Final = SignedResponse.model_validate_json(payload) + if not hmac.compare_digest(signed.signature.encode(), hmac.new(secret, signed.response.encode(), hashlib.sha256).hexdigest().encode()): + return None + response: Final = CachedResponse.model_validate_json(signed.response) + chunks: Final = tuple(base64.b64decode(chunk, validate=True) for chunk in response.chunks) + except (ValidationError, ValueError): + return None + if response.request_key != key or not successful_response(url, response.status_code, response.headers, b"".join(chunks)): + return None + return response + + +@dataclass(slots=True) +class CacheCounters: + counts: tuple[tuple[str, int], ...] = () + lock: threading.Lock = field(default_factory=threading.Lock) + + def increment(self, name: str) -> None: + with self.lock: + current: Final = dict(self.counts) + self.counts = tuple((current | {name: current.get(name, 0) + 1}).items()) + + +@dataclass(slots=True) +class ResponseCapture: + buffer: io.BytesIO = field(default_factory=io.BytesIO) + size: int = 0 + eligible: bool = True + + def observe(self, step: StreamStep) -> None: + if not self.eligible: + return + if isinstance(step, StreamTruncation) or self.size + len(step.data) + 8 > MAX_RESPONSE_BYTES: + self.eligible = False + self.buffer.close() + return + self.buffer.write(len(step.data).to_bytes(8, "big")) + self.buffer.write(step.data) + self.size += len(step.data) + 8 + + def chunks(self) -> tuple[bytes, ...]: + self.buffer.seek(0) + return tuple(self.buffer.read(int.from_bytes(size, "big")) for size in iter(lambda: self.buffer.read(8), b"")) + + +def response_steps(response: CachedResponse) -> Generator[StreamStep, None, None]: + for chunk in response.chunks: + yield StreamChunk(base64.b64decode(chunk, validate=True)) + + +@dataclass(frozen=True, slots=True) +class CacheEdge: + store: ResponseStore + secret: bytes = field(repr=False) + counters: CacheCounters = field(default_factory=CacheCounters) + wait_seconds: float = 2.0 + clock: Callable[[], float] = time.monotonic + sleep: Callable[[float], None] = time.sleep + + def lookup(self, key: str) -> CacheLookup: + deadline: Final = self.clock() + self.wait_seconds + while isinstance(result := self.store.lookup(key), CacheBusy) and self.clock() < deadline: + self.sleep(min(0.05, max(0, deadline - self.clock()))) + return result + + def forward(self, method: str, url: str, headers: dict[str, str], body: bytes | None, timeout: float) -> StreamHead | NetworkError: + if not cacheable_endpoint(method, url, body): + self.counters.increment("bypass") + self.counters.increment("upstream_attempts") + return forward_stream(method, url, headers=headers, body=body, timeout=timeout) + prepared: Final = prepare_forward(method, url, headers, body) + if isinstance(prepared, NetworkError): + self.counters.increment("rejected") + return prepared + key: Final = exact_key(self.secret, method, url, prepared.headers, body) + found: Final = self.lookup(key) + if isinstance(found, CacheHit): + response: Final = decode_response(self.secret, key, found.payload, url) + if response is not None and self.clock() < found.valid_until: + self.counters.increment("hits") + return StreamHead(response.status_code, response.headers, response_steps(response)) + self.counters.increment("corrupt" if response is None else "expired") + self.store.discard(key, found.payload) + capture_slot: Final = self.lookup(key) if isinstance(found, CacheHit) else found + self.counters.increment("misses") + if isinstance(capture_slot, CacheUnavailable): + self.counters.increment("cache_errors") + self.counters.increment("upstream_attempts") + head: Final = forward_prepared_stream(prepared, timeout) + if not isinstance(capture_slot, CaptureLease): + return head + if isinstance(head, NetworkError): + self.store.release(key, capture_slot) + self.counters.increment("rejected") + return head + return StreamHead(head.status_code, head.headers, primed_steps(self.capture(key, capture_slot, url, head))) + + def capture(self, key: str, lease: CaptureLease, url: str, head: StreamHead) -> Generator[StreamStep, None, None]: + capture: Final = ResponseCapture() + try: + with closing(head.steps): + yield StreamChunk(b"") + for step in head.steps: + yield step + capture.observe(step) + chunks: Final = capture.chunks() if capture.eligible else () + if not capture.eligible or not successful_response(url, head.status_code, head.headers, b"".join(chunks)): + self.counters.increment("rejected") + return + response: Final = CachedResponse( + request_key=key, status_code=head.status_code, headers=head.headers, + chunks=tuple(base64.b64encode(chunk).decode("ascii") for chunk in chunks), + ) + published: Final = self.store.publish(key, lease, encode_response(self.secret, response)) + self.counters.increment("writes" if published else "write_failures") + finally: + self.store.release(key, lease) + capture.buffer.close() diff --git a/tests/e2e/provider_cache_redis.py b/tests/e2e/provider_cache_redis.py new file mode 100644 index 00000000000..be4e31b2c49 --- /dev/null +++ b/tests/e2e/provider_cache_redis.py @@ -0,0 +1,154 @@ +from __future__ import annotations + +import atexit +import functools +import json +import logging +import os +import re +import time +import uuid +from dataclasses import dataclass +from pathlib import Path +from typing import Final, Protocol, cast + +from provider_cache import LIFETIME_SECONDS, CacheBusy, CacheEdge, CacheHit, CacheLookup, CacheUnavailable, CaptureLease +from pydantic import TypeAdapter, ValidationError +from redis import Redis +from redis.exceptions import RedisError + +REDIS_ARRAY: Final[TypeAdapter[list[bytes]]] = TypeAdapter(list[bytes]) + +LOOKUP: Final = """ +local clock = redis.call('TIME') +local now = clock[1] * 1000 + math.floor(clock[2] / 1000) +local row = redis.call('HMGET', KEYS[1], 'captured', 'expires', 'payload') +if row[3] then + local captured = tonumber(row[1]) + local expires = tonumber(row[2]) + if captured and expires and captured <= now and expires > now + and expires - captured == tonumber(ARGV[2]) then + return {'hit', row[3], tostring(expires - now)} + end + redis.call('DEL', KEYS[1]) +end +if redis.call('SET', KEYS[2], ARGV[1], 'NX', 'PX', ARGV[3]) then + return {'lease', tostring(now), tostring(now + tonumber(ARGV[2]))} +end +return {'busy'} +""" + +PUBLISH: Final = """ +if redis.call('GET', KEYS[2]) ~= ARGV[1] then return 0 end +local clock = redis.call('TIME') +local now = clock[1] * 1000 + math.floor(clock[2] / 1000) +local captured = tonumber(ARGV[2]) +local expires = tonumber(ARGV[3]) +if captured > now or expires <= now or expires - captured ~= tonumber(ARGV[5]) then return 0 end +if redis.call('EXISTS', KEYS[1]) == 1 then return 0 end +redis.call('HSET', KEYS[1], 'captured', ARGV[2], 'expires', ARGV[3], 'payload', ARGV[4]) +redis.call('PEXPIREAT', KEYS[1], expires) +redis.call('DEL', KEYS[2]) +return 1 +""" + +RELEASE: Final = """ +if redis.call('GET', KEYS[1]) ~= ARGV[1] then return 0 end +return redis.call('DEL', KEYS[1]) +""" + +DISCARD: Final = """ +if redis.call('HGET', KEYS[1], 'payload') ~= ARGV[1] then return 0 end +return redis.call('DEL', KEYS[1]) +""" + + +class RedisCommands(Protocol): + def eval(self, script: str, numkeys: int, *args: str | bytes | int) -> object: ... + + +@dataclass(frozen=True, slots=True) +class RedisResponseStore: + client: RedisCommands + namespace: str + lifetime_ms: int = LIFETIME_SECONDS * 1000 + lease_ms: int = 120_000 + + def keys(self, key: str) -> tuple[str, str]: + prefix: Final = f"e2e-provider-cache:v1:{self.namespace}:{{{key}}}" + return prefix + ":response", prefix + ":lease" + + def lookup(self, key: str) -> CacheLookup: + token: Final = uuid.uuid4().hex + started: Final = time.monotonic() + try: + result: Final = self.client.eval(LOOKUP, 2, *self.keys(key), token, self.lifetime_ms, self.lease_ms) + except (RedisError, OSError): + return CacheUnavailable() + try: + parts: Final = tuple(REDIS_ARRAY.validate_python(result, strict=True)) + except ValidationError: + return CacheUnavailable() + if len(parts) == 3 and parts[0] == b"hit" and parts[2].isdigit(): + return CacheHit(parts[1], started + int(parts[2]) / 1000) + if len(parts) == 3 and parts[0] == b"lease" and parts[1].isdigit() and parts[2].isdigit(): + return CaptureLease(token, int(parts[1]), int(parts[2])) + if parts == (b"busy",): + return CacheBusy() + return CacheUnavailable() + + def publish(self, key: str, lease: CaptureLease, payload: bytes) -> bool: + try: + result: Final = self.client.eval( + PUBLISH, 2, *self.keys(key), lease.token, lease.captured_at_ms, lease.expires_at_ms, payload, self.lifetime_ms, + ) + except (RedisError, OSError): + return False + return result == 1 + + def release(self, key: str, lease: CaptureLease) -> bool: + try: + result: Final = self.client.eval(RELEASE, 1, self.keys(key)[1], lease.token) + except (RedisError, OSError): + return False + return result == 1 + + def discard(self, key: str, payload: bytes) -> bool: + try: + result: Final = self.client.eval(DISCARD, 1, self.keys(key)[0], payload) + except (RedisError, OSError): + return False + return result == 1 + + +def redis_store(url: str, namespace: str) -> RedisResponseStore: + client: Final = Redis.from_url(url, socket_timeout=0.25, socket_connect_timeout=0.25, decode_responses=False) + return RedisResponseStore(cast(RedisCommands, client), namespace) + + +def write_metrics(cache: CacheEdge) -> None: + report: Final = json.dumps({"provider_cache": dict(cache.counters.counts)}) + directory: Final = os.environ.get("E2E_PROVIDER_CACHE_METRICS_DIR") + if directory: + try: + root: Final = Path(directory) + root.mkdir(parents=True, exist_ok=True) + (root / f"{os.getpid()}.json").write_text(report + "\n") + except OSError: + logging.getLogger(__name__).warning("provider cache metrics artifact unavailable") + logging.getLogger(__name__).info("%s", report) + + +@functools.lru_cache(maxsize=1) +def configured_cache() -> CacheEdge | None: + if os.environ.get("E2E_PROVIDER_CACHE", "0") == "0": + return None + if os.environ.get("E2E_PROVIDER_CACHE") != "1": + raise ValueError("E2E_PROVIDER_CACHE must be 0 or 1") + secret: Final = os.environ.get("E2E_PROVIDER_CACHE_HMAC_KEY", "").encode() + namespace: Final = os.environ.get("E2E_PROVIDER_CACHE_NAMESPACE", "") + if len(secret) < 32 or re.fullmatch(r"[a-zA-Z0-9_-]{1,64}", namespace) is None: + raise ValueError("provider cache requires a dedicated key and namespace") + cache: Final = CacheEdge(redis_store(os.environ["E2E_PROVIDER_CACHE_REDIS_URL"], namespace), secret) + atexit.register(write_metrics, cache) + return cache diff --git a/tests/e2e/provider_cache_routing.py b/tests/e2e/provider_cache_routing.py new file mode 100644 index 00000000000..24599b5a313 --- /dev/null +++ b/tests/e2e/provider_cache_routing.py @@ -0,0 +1,23 @@ +from __future__ import annotations + +from collections.abc import Callable +from contextvars import ContextVar +from typing import Final + +from models import LiteLLMParamsBody, ModelMode + +LIVE_PROVIDER_REQUIRED: Final[ContextVar[bool]] = ContextVar("live_provider_required", default=False) + + +def route_cache_model( + params: LiteLLMParamsBody, base_for: Callable[[str], str | None], *, enabled: bool, mode: ModelMode | None = None, +) -> LiteLLMParamsBody: + if not enabled or mode == "realtime" or LIVE_PROVIDER_REQUIRED.get() or params.api_base is not None or params.mock_response is not None: + return params + provider: Final = params.model.partition("/")[0] + if provider not in {"openai", "anthropic"} or params.litellm_credential_name is not None: + return params + base: Final = base_for(provider) + if base is None: + return params + return params.model_copy(update={"api_base": f"{base}/v1" if provider == "openai" else base}) diff --git a/tests/e2e/provider_edge.py b/tests/e2e/provider_edge.py index de36895ebb6..49d574ee957 100644 --- a/tests/e2e/provider_edge.py +++ b/tests/e2e/provider_edge.py @@ -42,6 +42,7 @@ import base64 import difflib import functools import hashlib +import os import re import threading from collections import deque @@ -93,6 +94,8 @@ from fixture_mode import ( parse_fixture_mode, ) from fixture_profile import IneligibleRequest, MatchProfile, match_profile, strict_identity +from provider_cache import CacheEdge +from provider_cache_routing import LIVE_PROVIDER_REQUIRED from pydantic import JsonValue, TypeAdapter EDGE_MOUNTS: Final[Mapping[str, str]] = MappingProxyType( @@ -506,7 +509,7 @@ class LiveEdge: pass -type EdgeBackend = RecordEdge | ReplayEdge | LiveEdge +type EdgeBackend = RecordEdge | ReplayEdge | LiveEdge | CacheEdge @dataclass(slots=True) @@ -750,12 +753,16 @@ def _handle_record( def _handle_live( - method: str, url: str, headers: Mapping[str, str], body: bytes | None, timeout: float + method: str, url: str, headers: Mapping[str, str], body: bytes | None, timeout: float, + cache: CacheEdge | None = None, ) -> EdgeOutcome: forwarded: Final = { name: value for name, value in headers.items() if name.lower() not in _REQUEST_DROPPED_HEADERS } - head: Final = forward_stream(method, url, headers=forwarded, body=body, timeout=timeout) + head: Final = ( + forward_stream(method, url, headers=forwarded, body=body, timeout=timeout) + if cache is None else cache.forward(method, url, forwarded, body, timeout) + ) match head: case NetworkError(message=message): return _recorded_outcome(_network_error_response(message)) @@ -821,6 +828,10 @@ def handle_edge_request( else edge_request(method, split.path, split.query, body, _header_value(headers, "content-type")) ) match backend: + case CacheEdge(): + return _handle_live( + method, _upstream_url(upstream_base, upstream_path, split.query), headers, body, timeout, backend, + ) case LiveEdge(): return _handle_live( method, _upstream_url(upstream_base, upstream_path, split.query), headers, body, timeout @@ -871,11 +882,17 @@ class _EdgeHandler(BaseHTTPRequestHandler): or isinstance(edge_server.backend, ReplayEdge) and edge_server.backend.source.bundle.manifest.match_profile == "stateless_v1" ) - if strict and len({name.lower() for name in self.headers.keys()}) != len(self.headers): + if strict and len({name.lower() for name in self.headers}) != len(self.headers): self._write_reply(_text_reply(REPLAY_MISS_STATUS, "stateless_v1 eligibility error: duplicate headers")) return + duplicate_headers: Final = len({name.lower() for name in self.headers}) != len(self.headers) + selected_backend: Final = ( + LiveEdge() if isinstance(edge_server.backend, CacheEdge) and duplicate_headers else edge_server.backend + ) + if isinstance(edge_server.backend, CacheEdge) and duplicate_headers: + edge_server.backend.counters.increment("duplicate_header_bypass") outcome: Final = handle_edge_request( - edge_server.backend, + selected_backend, edge_server.mounts, self.command, self.path, @@ -908,12 +925,12 @@ class _EdgeHandler(BaseHTTPRequestHandler): shuts down write-side first: the proxy sees a graceful close mid-message, which is the incomplete chunked read a provider hanging up produces, and not the reset that could discard the chunks already in flight.""" - self.send_response(stream.status_code) - for name, value in stream.headers.items(): - self.send_header(name, value) - self.send_header("transfer-encoding", "chunked") - self.end_headers() with closing(stream.steps) as steps: + self.send_response(stream.status_code) + for name, value in stream.headers.items(): + self.send_header(name, value) + self.send_header("transfer-encoding", "chunked") + self.end_headers() for step in steps: match step: case StreamChunk(data=data): @@ -923,7 +940,7 @@ class _EdgeHandler(BaseHTTPRequestHandler): return case _: assert_never(step) - self.wfile.write(b"0\r\n\r\n") + self.wfile.write(b"0\r\n\r\n") def log_message(self, format: str, *args: object) -> None: """Silence the per-request stderr line BaseHTTPRequestHandler emits.""" @@ -1056,6 +1073,8 @@ def provider_edge_api_base( case InvalidFixtureMode(value=value): raise ValueError(f"E2E_FIXTURE_MODE={value!r} is not one of {', '.join(FIXTURE_MODES)}") case "live": + if configured_cache_backend() is not None: + return _shared_cache_edge(bind_host, advertise_host, forward_timeout).api_base(mount) return None case "record" | "replay": if mount not in EDGE_MOUNTS: @@ -1073,7 +1092,7 @@ def _observed_backend(mode_raw: str, bundle_dir: Path) -> EdgeBackend: case InvalidFixtureMode(value=value): raise ValueError(f"E2E_FIXTURE_MODE={value!r} is not one of {', '.join(FIXTURE_MODES)}") case "live": - return LiveEdge() + return configured_cache_backend() or LiveEdge() case "record": return RecordEdge(_shared_recorder(bundle_dir, match_profile()), threading.Lock()) case "replay": @@ -1082,6 +1101,24 @@ def _observed_backend(mode_raw: str, bundle_dir: Path) -> EdgeBackend: assert_never(mode) +def configured_cache_backend() -> CacheEdge | None: + if LIVE_PROVIDER_REQUIRED.get() or os.environ.get("E2E_PROVIDER_CACHE", "0") == "0": + return None + from provider_cache_redis import configured_cache + + return configured_cache() + + +@functools.lru_cache(maxsize=8) +def _shared_cache_edge(bind_host: str, advertise_host: str, forward_timeout: float) -> ProviderEdge: + backend: Final = configured_cache_backend() + assert backend is not None + return start_provider_edge( + backend, mounts=EDGE_MOUNTS, bind_host=bind_host, + advertise_host=advertise_host, forward_timeout=forward_timeout, + ).edge + + @contextmanager def observed_provider_edge( observation: ProviderRequestObservation, diff --git a/tests/e2e/proxy_client.py b/tests/e2e/proxy_client.py index 3f7fba5ffec..f8ed8843461 100644 --- a/tests/e2e/proxy_client.py +++ b/tests/e2e/proxy_client.py @@ -8,6 +8,7 @@ ProxyClient's key/customer methods for cleanup. Read-backs are eventually consis from __future__ import annotations +import os import time import warnings from collections.abc import Callable, Mapping @@ -26,6 +27,7 @@ from e2e_config import ( PROXY_REPLICA_URLS, REQUEST_TIMEOUT, SLOW_PROVIDER_TIMEOUT_SECONDS, + provider_edge_base, settle_propagation, ) from e2e_http import ( @@ -93,6 +95,7 @@ from models import ( UserDeleteBody, UserDeleteResponse, ) +from provider_cache_routing import route_cache_model from pydantic import BaseModel from transport import HttpTransport, SplitTransport, Transport, is_control_plane_path @@ -645,7 +648,10 @@ class ProxyClient: self.transport.post( "/model/new", headers=self.management_headers(), - json=body, + json=body.model_copy(update={"litellm_params": route_cache_model( + body.litellm_params, provider_edge_base, + enabled=os.environ.get("E2E_PROVIDER_CACHE", "0") == "1", mode=body.model_info.mode, + )}), response_type=ModelNewResponse, ) ).model_id From 45d5e6b8336fa4383e0615d2c1020ebeca34aab6 Mon Sep 17 00:00:00 2001 From: Yuneng Jiang Date: Tue, 15 Sep 2026 18:26:44 -0700 Subject: [PATCH 2/6] fix(e2e): start cache CI service and count bypass calls --- .github/workflows/test-provider-cache.yml | 8 +++---- .../test_provider_cache.py | 23 +++++++++++++++++++ tests/e2e/provider_edge.py | 1 + 3 files changed, 28 insertions(+), 4 deletions(-) diff --git a/.github/workflows/test-provider-cache.yml b/.github/workflows/test-provider-cache.yml index 29b9d08a267..0507f90cf89 100644 --- a/.github/workflows/test-provider-cache.yml +++ b/.github/workflows/test-provider-cache.yml @@ -24,7 +24,7 @@ jobs: ports: - 6379:6379 options: >- - --health-cmd 'redis-cli ping' + --health-cmd "redis-cli ping" --health-interval 5s --health-timeout 3s --health-retries 10 @@ -35,10 +35,10 @@ jobs: E2E_PROVIDER_CACHE: '0' E2E_FIXTURE_MODE: live steps: - - uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 + - uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0 with: persist-credentials: false - - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0 with: python-version: '3.13' - uses: ./.github/actions/setup-uv-with-retries @@ -56,7 +56,7 @@ jobs: -q --junitxml=provider-cache-results.xml - name: Save test results if: always() - uses: actions/upload-artifact@4cec3d8aa04e39d1a68397de0c4cd6fb9dce8ec1 + uses: actions/upload-artifact@4cec3d8aa04e39d1a68397de0c4cd6fb9dce8ec1 # v4.6.1 with: name: provider-cache-results path: provider-cache-results.xml diff --git a/tests/code_coverage_tests/test_provider_cache.py b/tests/code_coverage_tests/test_provider_cache.py index 7edca8151af..d0e8c5296dc 100644 --- a/tests/code_coverage_tests/test_provider_cache.py +++ b/tests/code_coverage_tests/test_provider_cache.py @@ -11,8 +11,10 @@ from collections.abc import Generator from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager from dataclasses import dataclass, replace +from http.client import HTTPConnection from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Final +from urllib.parse import urlsplit import pytest from e2e_http import NetworkError, PreparedForward, RawResponse, StreamChunk, StreamHead, forward, prepare_forward @@ -407,3 +409,24 @@ def test_enabled_environment_reuses_store_across_fresh_backends( assert configured_cache_backend() is None finally: configured_cache.cache_clear() + + +def test_duplicate_headers_bypass_cache_and_count_live_calls(store: RedisResponseStore, provider: Provider) -> None: + cache: Final = CacheEdge(store, SECRET) + with edge(cache, provider) as url: + parsed: Final = urlsplit(url) + for _ in range(2): + connection = HTTPConnection(str(parsed.hostname), parsed.port, timeout=5) + try: + connection.putrequest("POST", parsed.path) + connection.putheader("content-length", str(len(BODY))) + connection.putheader("content-type", "application/json") + connection.putheader("x-duplicate", "first") + connection.putheader("x-duplicate", "second") + connection.endheaders(BODY) + assert connection.getresponse().read() == SUCCESS + finally: + connection.close() + assert len(provider.hits) == 2 + assert dict(cache.counters.counts)["duplicate_header_bypass"] == 2 + assert dict(cache.counters.counts)["upstream_attempts"] == 2 diff --git a/tests/e2e/provider_edge.py b/tests/e2e/provider_edge.py index 49d574ee957..053a2b5c0d3 100644 --- a/tests/e2e/provider_edge.py +++ b/tests/e2e/provider_edge.py @@ -891,6 +891,7 @@ class _EdgeHandler(BaseHTTPRequestHandler): ) if isinstance(edge_server.backend, CacheEdge) and duplicate_headers: edge_server.backend.counters.increment("duplicate_header_bypass") + edge_server.backend.counters.increment("upstream_attempts") outcome: Final = handle_edge_request( selected_backend, edge_server.mounts, From aa15e9f23f2d5a1fe46ed782d1605c3840a28d04 Mon Sep 17 00:00:00 2001 From: Yuneng Jiang Date: Tue, 15 Sep 2026 18:45:43 -0700 Subject: [PATCH 3/6] test(e2e): reject expired cache entries before physical eviction --- .../test_provider_cache.py | 23 +++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/tests/code_coverage_tests/test_provider_cache.py b/tests/code_coverage_tests/test_provider_cache.py index d0e8c5296dc..9a4cba7f322 100644 --- a/tests/code_coverage_tests/test_provider_cache.py +++ b/tests/code_coverage_tests/test_provider_cache.py @@ -192,6 +192,29 @@ def test_concurrent_requests_publish_atomically(store: RedisResponseStore, provi assert len(provider.hits) == 1 +@pytest.mark.parametrize("age_past_expiry_ms", [0, 1]) +def test_expired_response_is_rejected_without_physical_eviction( + store: RedisResponseStore, age_past_expiry_ms: int, +) -> None: + response_key: Final = store.keys("expired")[0] + retained: Final = store.client.eval( + """ +local clock = redis.call('TIME') +local expires = clock[1] * 1000 + math.floor(clock[2] / 1000) - tonumber(ARGV[1]) +redis.call('HSET', KEYS[1], 'captured', expires - 86400000, 'expires', expires, 'payload', 'old-response') +return redis.call('PTTL', KEYS[1]) +""", + 1, response_key, age_past_expiry_ms, + ) + assert retained == -1 + replacement: Final = store.lookup("expired") + assert isinstance(replacement, CaptureLease) + assert replacement.expires_at_ms - replacement.captured_at_ms == 86_400_000 + assert store.publish("expired", replacement, b"fresh-response") + hit: Final = store.lookup("expired") + assert isinstance(hit, CacheHit) and hit.payload == b"fresh-response" + + @pytest.mark.parametrize("truncated", [False, True]) def test_stream_completion_controls_publication(store: RedisResponseStore, provider: Provider, truncated: bool) -> None: provider.stream = True From d94b9d117d70e42866948c2986a9ea0c3a4533e6 Mon Sep 17 00:00:00 2001 From: Yuneng Jiang Date: Tue, 15 Sep 2026 19:16:45 -0700 Subject: [PATCH 4/6] docs(e2e): clarify recorded IDs and quota header semantics --- tests/e2e/PROVIDER_CACHE.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/tests/e2e/PROVIDER_CACHE.md b/tests/e2e/PROVIDER_CACHE.md index 280132d0940..efd9eb66daa 100644 --- a/tests/e2e/PROVIDER_CACHE.md +++ b/tests/e2e/PROVIDER_CACHE.md @@ -22,6 +22,12 @@ Do not give cache credentials to candidate deployments. Counter artifacts contai Tests that require real provider timing, limits or state use `@pytest.mark.provider_live`. The marker keeps newly registered models on live routes without weakening their assertions. Ordinary assertion failures still fail E2E. The shared cache does not modify provider response IDs or make the proxy aware of replay +## Recorded response semantics + +Replay preserves the original response ID, usage and end-to-end headers. The proxy can therefore deduplicate repeated provider IDs when storing spend-log rows, just as it does when a live upstream returns the same ID twice. One spend-log row per invocation is not guaranteed for identical recorded responses. Existing spend reconciliation requests use distinct prompt markers and retain their distinct-ID and row-count assertions; accounting tests are not automatically excluded from caching + +Provider remaining-quota headers describe the captured response. Metrics derived from them are historical on a cache hit, not a measurement of current provider capacity. Gateway-generated API-key quota headers are a separate contract. A test of fresh provider quota or timing must use the live-provider policy; replay can still exercise how the proxy processes the recorded headers + ## Qualification `tests/code_coverage_tests/test_provider_cache.py` exercises local HTTP providers and disposable real Redis. CI runs these checks with the existing provider-edge and replay harness tests. These component checks do not establish Buildkite deployment, full-suite cross-build reuse or a genuine 24-hour expiry observation; those require separate runtime evidence From 3dda798c8a24f255219a2808eccbd2f66434f681 Mon Sep 17 00:00:00 2001 From: Yuneng Jiang Date: Tue, 15 Sep 2026 19:28:50 -0700 Subject: [PATCH 5/6] ci(e2e): consolidate cache contracts in filtered CircleCI job --- .circleci/config.yml | 17 ++++- .circleci/scripts/classify_changes.sh | 11 +++- .circleci/scripts/path_filter.sh | 4 +- .github/workflows/test-provider-cache.yml | 63 ------------------- .../test_litellm/test_circleci_path_filter.py | 14 +++++ 5 files changed, 40 insertions(+), 69 deletions(-) delete mode 100644 .github/workflows/test-provider-cache.yml diff --git a/.circleci/config.yml b/.circleci/config.yml index bb4ad0f4019..e2102a9ae91 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -9,7 +9,7 @@ commands: parameters: category: type: enum - enum: ["backend", "client"] + enum: ["backend", "client", "provider-harness"] default: "backend" steps: - run: @@ -2918,19 +2918,30 @@ jobs: provider_replay_harness: docker: - *python312_image + - image: redis@sha256:e2debfb7956fa12c7ddc79d7e645c8cf26b30c99a6e9161ea9bf4171e1668a5f working_directory: ~/project resource_class: medium + environment: + E2E_CACHE_TEST_REDIS_URL: redis://127.0.0.1:6379/0 + E2E_PROVIDER_CACHE: "0" + E2E_FIXTURE_MODE: live steps: + - checkout + - skip_if_unrelated_changes: + category: provider-harness - setup_litellm_test_deps + - wait_for_service: + url: tcp://localhost:6379 - run: - name: Test provider replay harness + name: Test provider capture and replay harness command: | mkdir -p test-results/provider-replay-harness uv run --no-sync pytest -q --noconftest -o addopts= -o pythonpath=tests/e2e -p no:rerunfailures \ --junitxml=test-results/provider-replay-harness/junit.xml \ tests/e2e/test_provider_edge.py tests/e2e/test_fixture_bundle.py \ tests/e2e/test_fixture_canonical.py tests/e2e/test_fixture_mode.py \ - tests/code_coverage_tests/test_provider_replay_harness.py + tests/code_coverage_tests/test_provider_replay_harness.py \ + tests/code_coverage_tests/test_provider_cache.py - store_test_results: path: test-results/provider-replay-harness diff --git a/.circleci/scripts/classify_changes.sh b/.circleci/scripts/classify_changes.sh index 7aa0c3544ee..9dc7b76b23f 100755 --- a/.circleci/scripts/classify_changes.sh +++ b/.circleci/scripts/classify_changes.sh @@ -1,13 +1,19 @@ #!/usr/bin/env bash set -uo pipefail -category="${1:?usage: classify_changes.sh }" +category="${1:?usage: classify_changes.sh }" has_client=false has_backend=false has_ci=false +has_provider_harness=false while IFS= read -r file || [ -n "$file" ]; do [ -n "$file" ] || continue + case "$file" in + tests/e2e/*/*.py) : ;; + tests/e2e/*.py | tests/code_coverage_tests/test_provider_cache.py | tests/code_coverage_tests/test_provider_replay_harness.py | tests/test_litellm/test_circleci_path_filter.py | .circleci/* | pyproject.toml | uv.lock) + has_provider_harness=true ;; + esac case "$file" in ui/* | tests/e2e/ui/*) has_client=true ;; docs/* | *.md | *.mdx) : ;; @@ -17,6 +23,9 @@ while IFS= read -r file || [ -n "$file" ]; do done case "$category" in + provider-harness) + [ "$has_provider_harness" = true ] && echo run || echo skip + ;; backend) [ "$has_backend" = true ] && echo run || echo skip ;; diff --git a/.circleci/scripts/path_filter.sh b/.circleci/scripts/path_filter.sh index dcf64a24399..cdadde732bd 100755 --- a/.circleci/scripts/path_filter.sh +++ b/.circleci/scripts/path_filter.sh @@ -1,7 +1,7 @@ #!/usr/bin/env bash set -uo pipefail -category="${1:?usage: path_filter.sh }" +category="${1:?usage: path_filter.sh }" here="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" run_full() { @@ -36,5 +36,5 @@ if [ "$decision" = run ]; then run_full "$category-relevant changes detected" fi -echo "path-filter[$category]: only unrelated (docs/client) changes detected; halting job as successful" +echo "path-filter[$category]: only unrelated changes detected; halting job as successful" circleci-agent step halt diff --git a/.github/workflows/test-provider-cache.yml b/.github/workflows/test-provider-cache.yml deleted file mode 100644 index 0507f90cf89..00000000000 --- a/.github/workflows/test-provider-cache.yml +++ /dev/null @@ -1,63 +0,0 @@ -name: Provider cache contracts - -on: - pull_request: - paths: - - 'tests/e2e/**' - - 'tests/code_coverage_tests/test_provider_cache.py' - - 'tests/code_coverage_tests/test_provider_replay_harness.py' - - '.github/workflows/test-provider-cache.yml' - - 'pyproject.toml' - - 'uv.lock' - workflow_dispatch: - -permissions: - contents: read - -jobs: - provider-cache: - runs-on: ubuntu-latest - timeout-minutes: 20 - services: - redis: - image: redis@sha256:e2debfb7956fa12c7ddc79d7e645c8cf26b30c99a6e9161ea9bf4171e1668a5f - ports: - - 6379:6379 - options: >- - --health-cmd "redis-cli ping" - --health-interval 5s - --health-timeout 3s - --health-retries 10 - env: - PYTHONDONTWRITEBYTECODE: '1' - PYTHONPATH: tests/e2e - E2E_CACHE_TEST_REDIS_URL: redis://127.0.0.1:6379/0 - E2E_PROVIDER_CACHE: '0' - E2E_FIXTURE_MODE: live - steps: - - uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0 - with: - persist-credentials: false - - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0 - with: - python-version: '3.13' - - uses: ./.github/actions/setup-uv-with-retries - with: - version: '0.10.9' - - uses: ./.github/actions/cache-cargo-build - - name: Install locked dependencies - run: .github/scripts/uv_sync_with_retries.sh --frozen --group proxy-dev --group e2e-dev - - name: Verify cache and existing edge contracts - run: >- - uv run --no-sync pytest -c /dev/null -p no:cacheprovider - tests/code_coverage_tests/test_provider_cache.py - tests/code_coverage_tests/test_provider_replay_harness.py - tests/e2e/test_provider_edge.py - -q --junitxml=provider-cache-results.xml - - name: Save test results - if: always() - uses: actions/upload-artifact@4cec3d8aa04e39d1a68397de0c4cd6fb9dce8ec1 # v4.6.1 - with: - name: provider-cache-results - path: provider-cache-results.xml - if-no-files-found: warn diff --git a/tests/test_litellm/test_circleci_path_filter.py b/tests/test_litellm/test_circleci_path_filter.py index b427a1a3bd8..af0e932400f 100644 --- a/tests/test_litellm/test_circleci_path_filter.py +++ b/tests/test_litellm/test_circleci_path_filter.py @@ -49,6 +49,20 @@ CI = [".github/workflows/test-litellm-ui-unit.yml"] @pytest.mark.parametrize( "category,changed,expected", [ + ("provider-harness", ["tests/e2e/provider_cache.py"], "run"), + ("provider-harness", ["tests/e2e/conftest.py"], "run"), + ("provider-harness", ["tests/e2e/e2e_http.py"], "run"), + ("provider-harness", ["tests/code_coverage_tests/test_provider_cache.py"], "run"), + ("provider-harness", ["tests/code_coverage_tests/test_provider_replay_harness.py"], "run"), + ("provider-harness", [".circleci/config.yml"], "run"), + ("provider-harness", [".circleci/scripts/classify_changes.sh"], "run"), + ("provider-harness", ["pyproject.toml"], "run"), + ("provider-harness", ["uv.lock"], "run"), + ("provider-harness", ["tests/e2e/PROVIDER_CACHE.md"], "skip"), + ("provider-harness", ["tests/e2e/ui/test_example.py"], "skip"), + ("provider-harness", ["tests/e2e/quota_management/test_quota.py"], "skip"), + ("provider-harness", ["litellm/main.py"], "skip"), + ("provider-harness", ["ui/litellm-dashboard/src/App.tsx"], "skip"), # docs-only: skip everything ("backend", DOCS, "skip"), ("client", DOCS, "skip"), From 917018308788e3abca5cbb10afdefaa05dcdb804 Mon Sep 17 00:00:00 2001 From: Yuneng Jiang Date: Tue, 15 Sep 2026 19:32:50 -0700 Subject: [PATCH 6/6] fix(e2e): exclude unknown routes from upstream counters --- tests/code_coverage_tests/test_provider_cache.py | 16 +++++++++++----- tests/e2e/provider_edge.py | 3 ++- 2 files changed, 13 insertions(+), 6 deletions(-) diff --git a/tests/code_coverage_tests/test_provider_cache.py b/tests/code_coverage_tests/test_provider_cache.py index 9a4cba7f322..9d89a5fe622 100644 --- a/tests/code_coverage_tests/test_provider_cache.py +++ b/tests/code_coverage_tests/test_provider_cache.py @@ -434,22 +434,28 @@ def test_enabled_environment_reuses_store_across_fresh_backends( configured_cache.cache_clear() -def test_duplicate_headers_bypass_cache_and_count_live_calls(store: RedisResponseStore, provider: Provider) -> None: +@pytest.mark.parametrize("known_mount", (True, False)) +def test_duplicate_headers_bypass_cache_and_count_live_calls( + store: RedisResponseStore, provider: Provider, known_mount: bool, +) -> None: cache: Final = CacheEdge(store, SECRET) with edge(cache, provider) as url: parsed: Final = urlsplit(url) for _ in range(2): connection = HTTPConnection(str(parsed.hostname), parsed.port, timeout=5) try: - connection.putrequest("POST", parsed.path) + connection.putrequest("POST", parsed.path if known_mount else "/unknown/v1/chat/completions") connection.putheader("content-length", str(len(BODY))) connection.putheader("content-type", "application/json") connection.putheader("x-duplicate", "first") connection.putheader("x-duplicate", "second") connection.endheaders(BODY) - assert connection.getresponse().read() == SUCCESS + response = connection.getresponse() + assert response.status == (200 if known_mount else 404) + payload = response.read() + assert payload == SUCCESS if known_mount else b"unknown provider mount" in payload finally: connection.close() - assert len(provider.hits) == 2 + assert len(provider.hits) == (2 if known_mount else 0) assert dict(cache.counters.counts)["duplicate_header_bypass"] == 2 - assert dict(cache.counters.counts)["upstream_attempts"] == 2 + assert dict(cache.counters.counts).get("upstream_attempts", 0) == (2 if known_mount else 0) diff --git a/tests/e2e/provider_edge.py b/tests/e2e/provider_edge.py index 053a2b5c0d3..dda9e6f8e4f 100644 --- a/tests/e2e/provider_edge.py +++ b/tests/e2e/provider_edge.py @@ -891,7 +891,8 @@ class _EdgeHandler(BaseHTTPRequestHandler): ) if isinstance(edge_server.backend, CacheEdge) and duplicate_headers: edge_server.backend.counters.increment("duplicate_header_bypass") - edge_server.backend.counters.increment("upstream_attempts") + if urlsplit(self.path).path.lstrip("/").partition("/")[0] in edge_server.mounts: + edge_server.backend.counters.increment("upstream_attempts") outcome: Final = handle_edge_request( selected_backend, edge_server.mounts,