From 866d94ed23c8306907df4252debdcdb14cb204ea Mon Sep 17 00:00:00 2001 From: Yuneng Jiang Date: Fri, 11 Sep 2026 12:10:19 -0700 Subject: [PATCH] test(e2e): verify cached answers and upstream request count --- tests/e2e/provider_edge.py | 99 ++++++++++++++-- .../e2e/router/test_reliability_cache_e2e.py | 107 ++++++++++++++---- tests/e2e/test_provider_edge.py | 66 ++++++++++- 3 files changed, 239 insertions(+), 33 deletions(-) diff --git a/tests/e2e/provider_edge.py b/tests/e2e/provider_edge.py index ceb695ffcd6..6c87c7ef7ac 100644 --- a/tests/e2e/provider_edge.py +++ b/tests/e2e/provider_edge.py @@ -45,18 +45,16 @@ import hashlib import re import threading from collections import deque -from collections.abc import Mapping, Sequence -from contextlib import closing +from collections.abc import Generator, Mapping, Sequence +from contextlib import closing, contextmanager from dataclasses import dataclass, field from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from itertools import islice from pathlib import Path from types import MappingProxyType -from typing import Final, Generator, Literal, assert_never +from typing import Final, Literal, assert_never from urllib.parse import parse_qsl, urlsplit -from pydantic import JsonValue, TypeAdapter - from e2e_http import ( NetworkError, StreamChunk, @@ -94,6 +92,7 @@ from fixture_mode import ( current_test_key, parse_fixture_mode, ) +from pydantic import JsonValue, TypeAdapter EDGE_MOUNTS: Final[Mapping[str, str]] = MappingProxyType( { @@ -495,7 +494,29 @@ class ReplayEdge: source: ReplaySource -type EdgeBackend = RecordEdge | ReplayEdge +@dataclass(frozen=True, slots=True) +class LiveEdge: + pass + + +type EdgeBackend = RecordEdge | ReplayEdge | LiveEdge + + +@dataclass(slots=True) +class ProviderRequestObservation: + marker: str + _count: int = field(default=0, init=False) + _lock: threading.Lock = field(default_factory=threading.Lock, init=False) + + def observe(self, body: bytes | None) -> None: + if body is not None and self.marker.encode() in body: + with self._lock: + self._count += 1 + + @property + def count(self) -> int: + with self._lock: + return self._count @dataclass(frozen=True, slots=True) @@ -721,6 +742,24 @@ def _handle_record( assert_never(head) +def _handle_live( + method: str, url: str, headers: Mapping[str, str], body: bytes | None, timeout: float +) -> 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) + match head: + case NetworkError(message=message): + return _recorded_outcome(_network_error_response(message)) + case StreamHead() if _is_streamed(head.headers): + return EdgeStream(head.status_code, _filtered_response_headers(head.headers), head.steps) + case StreamHead(): + return _recorded_outcome(_drain_to_response(head)) + case _: + assert_never(head) + + def _handle_replay(source: ReplaySource, request: RecordedRequest) -> EdgeOutcome: try: interaction: Final = source.next_interaction(request) @@ -753,6 +792,10 @@ def handle_edge_request( method, split.path, split.query, body, _header_value(headers, "content-type") ) match backend: + case LiveEdge(): + return _handle_live( + method, _upstream_url(upstream_base, upstream_path, split.query), headers, body, timeout + ) case RecordEdge(): return _handle_record( backend, @@ -792,6 +835,8 @@ class _EdgeHandler(BaseHTTPRequestHandler): assert isinstance(edge_server, _EdgeHTTPServer) length: Final = int(self.headers.get("content-length") or "0") body: Final = self.rfile.read(length) if length else None + if edge_server.observation is not None: + edge_server.observation.observe(body) outcome: Final = handle_edge_request( edge_server.backend, edge_server.mounts, @@ -857,11 +902,13 @@ class _EdgeHTTPServer(ThreadingHTTPServer): backend: EdgeBackend, mounts: Mapping[str, str], forward_timeout: float, + observation: ProviderRequestObservation | None, ) -> None: super().__init__(bind, _EdgeHandler) self.backend: Final = backend self.mounts: Final = mounts self.forward_timeout: Final = forward_timeout + self.observation: Final = observation @dataclass(frozen=True, slots=True) @@ -890,13 +937,14 @@ def start_provider_edge( bind_host: str = "127.0.0.1", advertise_host: str | None = None, forward_timeout: float = 60.0, + observation: ProviderRequestObservation | None = None, ) -> RunningEdge: """Boot an edge server on an OS-assigned port in a daemon thread. ``advertise_host`` is what api_base URLs name (it differs from the bind host when the proxy runs in a container and reaches the host machine via a gateway address like host.docker.internal).""" server: Final = _EdgeHTTPServer( - (bind_host, 0), backend=backend, mounts=mounts, forward_timeout=forward_timeout + (bind_host, 0), backend=backend, mounts=mounts, forward_timeout=forward_timeout, observation=observation ) thread: Final = threading.Thread(target=server.serve_forever, name="e2e-provider-edge", daemon=True) thread.start() @@ -979,3 +1027,40 @@ def provider_edge_api_base( return _shared_edge(mode, bundle_dir, bind_host, advertise_host, forward_timeout).api_base(mount) case _: assert_never(mode) + + +def _observed_backend(mode_raw: str, bundle_dir: Path) -> EdgeBackend: + mode: Final = parse_fixture_mode(mode_raw) + match mode: + case InvalidFixtureMode(value=value): + raise ValueError(f"E2E_FIXTURE_MODE={value!r} is not one of {', '.join(FIXTURE_MODES)}") + case "live": + return LiveEdge() + case "record": + return RecordEdge(_shared_recorder(bundle_dir), threading.Lock()) + case "replay": + return ReplayEdge(_shared_replay_source(bundle_dir)) + case _: + assert_never(mode) + + +@contextmanager +def observed_provider_edge( + observation: ProviderRequestObservation, + *, + mode_raw: str, + bundle_dir: Path, + bind_host: str, + advertise_host: str, + forward_timeout: float = 60.0, + mounts: Mapping[str, str] = EDGE_MOUNTS, +) -> Generator[ProviderEdge, None, None]: + running: Final = start_provider_edge( + _observed_backend(mode_raw, bundle_dir), mounts=mounts, + bind_host=bind_host, advertise_host=advertise_host, + forward_timeout=forward_timeout, observation=observation, + ) + try: + yield running.edge + finally: + running.shutdown() diff --git a/tests/e2e/router/test_reliability_cache_e2e.py b/tests/e2e/router/test_reliability_cache_e2e.py index 4ea05a1ecca..f7a2f2ffeb7 100644 --- a/tests/e2e/router/test_reliability_cache_e2e.py +++ b/tests/e2e/router/test_reliability_cache_e2e.py @@ -1,37 +1,98 @@ -"""Live e2e: the response cache returns a cached answer on an exact repeat. +"""An exact cache hit preserves the full choices and usage without another provider call. -The same unique prompt is sent twice to the real `gpt-5.5` deployment under the -same key: the first call is a cache miss (the proxy computes and stores the entry, -and returns no x-litellm-cache-key), the second is an exact hit (the proxy serves -from cache and returns x-litellm-cache-key). This relies on the standard Redis -response cache being enabled on the proxy under test. +Response IDs, creation timestamps and proxy headers are transport metadata; +compare every field within choices and usage, including provider extensions. """ from __future__ import annotations +from typing import Final + import pytest - from complexity_router_client import ComplexityRouterClient -from e2e_config import unique_marker -from reliability_support import chat_override +from e2e_config import ( + FIXTURE_DIR, + FIXTURE_MODE_RAW, + PROVIDER_EDGE_ADVERTISE_HOST, + PROVIDER_EDGE_BIND_HOST, + REQUEST_TIMEOUT, + unique_marker, +) +from lifecycle import ResourceManager +from models import ChatBody, ChatMessage, ChatResponse, LiteLLMParamsBody +from provider_edge import ProviderRequestObservation, observed_provider_edge +from pydantic import BaseModel, JsonValue -pytestmark = pytest.mark.e2e +pytestmark = [pytest.mark.e2e, pytest.mark.replayable] + + +class _CacheChatBody(ChatBody): + ttl: int = 600 + + +class _CachedAnswer(BaseModel): + model: str + choices: tuple[dict[str, JsonValue], ...] + usage: dict[str, JsonValue] class TestReliabilityCache: @pytest.mark.covers("reliability.cache.exact.returns_cached") - def test_exact_cache_returns_cached(self, client: ComplexityRouterClient, scoped_key: str) -> None: - prompt = f"cache probe {unique_marker()}" + def test_exact_cache_returns_cached( + self, client: ComplexityRouterClient, resources: ResourceManager, scoped_key: str + ) -> None: + marker: Final = unique_marker() + model: Final = f"e2e-cache-{marker}" + prompt: Final = f"Reply with a short sentence about a blue lantern. Request marker: {marker}" + observation: Final = ProviderRequestObservation(marker) - first = chat_override(client.proxy, scoped_key, "gpt-5.5", prompt, cache=None) - assert first.status_code == 200, f"first call should succeed, got {first.status_code}: {first.body[:300]}" - assert "x-litellm-cache-key" not in first.headers, ( - "first (uncached) call must not report a cache-key header" - ) + with observed_provider_edge( + observation, + mode_raw=FIXTURE_MODE_RAW, + bundle_dir=FIXTURE_DIR, + bind_host=PROVIDER_EDGE_BIND_HOST, + advertise_host=PROVIDER_EDGE_ADVERTISE_HOST, + forward_timeout=REQUEST_TIMEOUT, + ) as edge: + model_id: Final = client.proxy.create_model( + model, + LiteLLMParamsBody( + model="openai/gpt-5.6", + api_key="os.environ/OPENAI_API_KEY", + api_base=f"{edge.api_base('openai')}/v1", + ), + ) + resources.defer(lambda: client.proxy.delete_model(model_id)) + body: Final = _CacheChatBody( + model=model, + messages=[ChatMessage(role="user", content=prompt)], + max_completion_tokens=512, + reasoning_effort="none", + cache=None, + ) + first: Final = client.proxy.transport.send( + "/chat/completions", headers=client.proxy.transport.bearer(scoped_key), json=body + ) + assert first.status_code == 200, f"first call should succeed, got {first.status_code}: {first.body[:300]}" + assert "x-litellm-cache-key" not in first.headers, "first call must be a cache miss" + answer: Final = ChatResponse.model_validate_json(first.body) + assert len(answer.choices) == 1 + choice: Final = answer.choices[0] + assert choice.message is not None and choice.message.role == "assistant" + assert choice.message.content is not None and choice.message.content.strip(), "first answer is empty" + assert choice.finish_reason == "stop" + assert answer.usage is not None + assert answer.usage.prompt_tokens is not None and answer.usage.prompt_tokens > 0 + assert answer.usage.completion_tokens is not None and answer.usage.completion_tokens > 0 + assert answer.usage.total_tokens == answer.usage.prompt_tokens + answer.usage.completion_tokens + assert observation.count == 1, "first miss must invoke the provider exactly once" - second = chat_override(client.proxy, scoped_key, "gpt-5.5", prompt, cache=None) - assert second.status_code == 200, f"second call should succeed, got {second.status_code}: {second.body[:300]}" - assert "x-litellm-cache-key" in second.headers, ( - "second identical call should hit the response cache and report a cache-key header " - "(requires the proxy's Redis response cache to be enabled)" - ) + second: Final = client.proxy.transport.send( + "/chat/completions", headers=client.proxy.transport.bearer(scoped_key), json=body + ) + assert second.status_code == 200, f"second call should succeed, got {second.status_code}: {second.body[:300]}" + assert second.headers.get("x-litellm-cache-key"), "identical request must hit the response cache" + assert _CachedAnswer.model_validate_json(second.body) == _CachedAnswer.model_validate_json(first.body), ( + "cache hit changed the answer, finish reason or usage" + ) + assert observation.count == 1, "two successful requests must invoke the provider exactly once" diff --git a/tests/e2e/test_provider_edge.py b/tests/e2e/test_provider_edge.py index 8ab389ee43c..18f72ac0e7a 100644 --- a/tests/e2e/test_provider_edge.py +++ b/tests/e2e/test_provider_edge.py @@ -34,10 +34,7 @@ from pathlib import Path from typing import Final import pytest -from pydantic import TypeAdapter - from e2e_http import RawResponse, StreamChunk, forward -from fixture_canonical import canonicalize from fixture_bundle import ( BundleRecorder, Interaction, @@ -49,6 +46,7 @@ from fixture_bundle import ( prepare_bundle, slug_for_test, ) +from fixture_canonical import canonicalize from fixture_mode import current_test_key from provider_edge import ( REPLAY_MISS_STATUS, @@ -56,15 +54,18 @@ from provider_edge import ( EdgeReply, EdgeStream, ProviderEdge, + ProviderRequestObservation, RecordEdge, ReplayEdge, ReplaySource, edge_request, handle_edge_request, + observed_provider_edge, provider_edge_api_base, replay_leftover_error, start_provider_edge, ) +from pydantic import TypeAdapter CHAT_PATH = "/openai/v1/chat/completions" UPLOAD_PATH = "/openai/v1/files" @@ -1290,3 +1291,62 @@ class TestApiBaseSeam: assert second.endswith("/anthropic") assert first.rsplit("/", 1)[0] == second.rsplit("/", 1)[0] assert (root / "manifest.json").is_file() + + +class TestProviderRequestObservation: + def test_live_counts_repeated_marker_calls_without_recording(self, tmp_path: Path) -> None: + observation: Final = ProviderRequestObservation("observed-lantern") + with fake_provider() as provider: + with observed_provider_edge( + observation, mode_raw="live", bundle_dir=tmp_path / "unused", + bind_host="127.0.0.1", advertise_host="127.0.0.1", + mounts={"openai": provider_url(provider)}, + ) as edge: + assert observation.count == 0 + unrelated: Final = call_edge(edge, "POST", CHAT_PATH, body=chat_body("other-lantern")) + assert unrelated.status_code == 200 + assert observation.count == 0 + first: Final = call_edge(edge, "POST", CHAT_PATH, body=chat_body("observed-lantern")) + assert first.status_code == 200 + assert json_object(first.body)["echo"] == chat_body("observed-lantern").decode() + assert observation.count == 1 + second: Final = call_edge(edge, "POST", CHAT_PATH, body=chat_body("observed-lantern")) + assert second.status_code == 200 + assert observation.count == 2 + assert len(provider.hits) == 3 + assert not (tmp_path / "unused").exists() + + def test_record_and_replay_count_each_matching_call(self, tmp_path: Path) -> None: + with fake_provider() as provider: + for mode, observation in ( + ("record", ProviderRequestObservation("observed-lantern")), + ("replay", ProviderRequestObservation("observed-lantern")), + ): + with observed_provider_edge( + observation, mode_raw=mode, bundle_dir=tmp_path / "bundle", + bind_host="127.0.0.1", advertise_host="127.0.0.1", + mounts={"openai": provider_url(provider)}, + ) as edge: + assert observation.count == 0 + for expected, response in ( + (index, call_edge(edge, "POST", CHAT_PATH, body=chat_body("observed-lantern"))) + for index in (1, 2) + ): + assert response.status_code == 200 + assert json_object(response.body)["hit"] == expected + assert observation.count == expected + assert len(provider.hits) == 2 + assert replay_leftover_error( + mode_raw="replay", bundle_dir=tmp_path / "bundle", test_key=current_test_key() + ) is None + + def test_failed_provider_attempt_is_counted(self, tmp_path: Path) -> None: + observation: Final = ProviderRequestObservation("observed-lantern") + with observed_provider_edge( + observation, mode_raw="live", bundle_dir=tmp_path / "unused", + bind_host="127.0.0.1", advertise_host="127.0.0.1", + mounts={"openai": "http://127.0.0.1:9"}, + ) as edge: + response: Final = call_edge(edge, "POST", CHAT_PATH, body=chat_body("observed-lantern")) + assert response.status_code == 502 + assert observation.count == 1