diff --git a/tests/code_coverage_tests/test_provider_cache.py b/tests/code_coverage_tests/test_provider_cache.py index e57d33a0406..dda3372c15f 100644 --- a/tests/code_coverage_tests/test_provider_cache.py +++ b/tests/code_coverage_tests/test_provider_cache.py @@ -26,6 +26,7 @@ from e2e_http import NetworkError, PreparedForward, RawResponse, StreamChunk, St from models import LiteLLMParamsBody, ModelMode from botocore.credentials import Credentials from botocore.eventstream import EventStreamBuffer +from fixture_bundle import slug_for_test from provider_cache import ( SIGNATURE_HEADERS, CacheEdge, @@ -35,7 +36,9 @@ from provider_cache import ( ResponseStore, cacheable_endpoint, request_identity, + scoped_edge_base, slotted_key, + split_test_segment, successful_response, ) from provider_cache_redis import PUBLISH, RedisCommands, RedisResponseStore, configured_cache, redis_store @@ -47,7 +50,13 @@ from provider_cache_routing import ( route_cache_model, ) from fixture_mode import SESSION_TEST_KEY -from provider_edge import EDGE_MOUNTS, configured_cache_backend, resolve_mount, start_provider_edge +from provider_edge import ( + EDGE_MOUNTS, + configured_cache_backend, + provider_edge_api_base, + resolve_mount, + start_provider_edge, +) from provider_edge_bedrock import bedrock_signer from redis.exceptions import ConnectionError as RedisConnectionError @@ -57,6 +66,7 @@ SUCCESS: Final = b'{"id":"provider-fixed-id","choices":[{"message":{"content":"h HEADERS: Final = {"content-type": "application/json", "authorization": "Bearer synthetic-account-one"} TEST_KEY: Final = "tests/e2e/synthetic_suite.py::TestCase::test_case" OTHER_TEST_KEY: Final = "tests/e2e/synthetic_suite.py::TestCase::test_other_case" +TEST_SLUG: Final = slug_for_test(TEST_KEY) def marked(marker: str) -> bytes: @@ -173,11 +183,11 @@ def store(redis_url: str) -> RedisResponseStore: return redis_store(redis_url, "test-" + uuid.uuid4().hex) -def cache_edge(store: ResponseStore, test_key: str = TEST_KEY) -> CacheEdge: +def cache_edge(store: ResponseStore) -> CacheEdge: """A cache edge standing in for one pytest process. A fresh instance over the same store is the next build running the same test: the recordings survive, the per-test FIFO slot counters start over.""" - return CacheEdge(store, SECRET, test_key=lambda: test_key) + return CacheEdge(store, SECRET) def slot_key( @@ -186,12 +196,13 @@ def slot_key( ) -> str: prepared: Final = prepare_forward("POST", url, headers, body) assert isinstance(prepared, PreparedForward) - return slotted_key(SECRET, request_identity(SECRET, test_key, "POST", url, prepared.headers, body), slot) + identity: Final = request_identity(SECRET, slug_for_test(test_key), "POST", url, prepared.headers, body) + return slotted_key(SECRET, identity, slot) -def bedrock_cache_edge(store: ResponseStore, test_key: str = TEST_KEY) -> CacheEdge: +def bedrock_cache_edge(store: ResponseStore) -> CacheEdge: return CacheEdge( - store, SECRET, test_key=lambda: test_key, + store, SECRET, policies={BEDROCK_MOUNT: MountPolicy( sign=bedrock_signer("us-east-1", lambda: STATIC_CREDENTIALS), unkeyed_headers=SIGNATURE_HEADERS, )}, @@ -199,11 +210,14 @@ def bedrock_cache_edge(store: ResponseStore, test_key: str = TEST_KEY) -> CacheE @contextmanager -def edge(cache: CacheEdge, provider: Provider) -> Generator[str, None, None]: +def edge(cache: CacheEdge, provider: Provider, test_key: str | None = TEST_KEY) -> Generator[str, None, None]: + """The URL a deployment registered by ``test_key`` would carry, or the bare + mount URL for None, which is what a registration made outside any test gets.""" upstream: Final = f"http://127.0.0.1:{provider.server_port}" running: Final = start_provider_edge(cache, mounts={"openai": upstream}) + base: Final = running.edge.api_base("openai") try: - yield running.edge.api_base("openai") + "/v1/chat/completions" + yield f"{base if test_key is None else scoped_edge_base(base, test_key)}/v1/chat/completions" finally: running.shutdown() @@ -213,7 +227,7 @@ def bedrock_edge(cache: CacheEdge, provider: Provider, action: str = "converse") upstream: Final = f"http://127.0.0.1:{provider.server_port}" running: Final = start_provider_edge(cache, mounts={BEDROCK_MOUNT: upstream}) try: - yield f"{running.edge.api_base(BEDROCK_MOUNT)}/model/{BEDROCK_MODEL}/{action}" + yield f"{scoped_edge_base(running.edge.api_base(BEDROCK_MOUNT), TEST_KEY)}/model/{BEDROCK_MODEL}/{action}" finally: running.shutdown() @@ -288,7 +302,7 @@ def test_expiry_does_not_slide(store: RedisResponseStore, provider: Provider) -> url: Final = f"http://127.0.0.1:{provider.server_port}/v1/chat/completions" def drain() -> None: - head = cache_edge(short).forward("openai", "POST", url, dict(HEADERS), BODY, 5) + head = cache_edge(short).forward("openai", "POST", url, dict(HEADERS), BODY, 5, test_key=TEST_SLUG) assert isinstance(head, StreamHead) assert b"".join(step.data for step in head.steps if isinstance(step, StreamChunk)) == SUCCESS @@ -312,7 +326,7 @@ def test_concurrent_builds_publish_one_recording_atomically( edges: Final = tuple(cache_edge(store) for _ in range(5)) def drain(cache: CacheEdge) -> bytes: - head = cache.forward("openai", "POST", url, dict(HEADERS), BODY, 5) + head = cache.forward("openai", "POST", url, dict(HEADERS), BODY, 5, test_key=TEST_SLUG) assert isinstance(head, StreamHead) return b"".join(step.data for step in head.steps if isinstance(step, StreamChunk)) @@ -401,7 +415,7 @@ def test_corrupt_entry_is_replaced_by_same_successful_request(store: RedisRespon assert store.publish(key, lease, payload) caches: Final = tuple(cache_edge(store) for _ in range(2)) for cache in caches: - head = cache.forward("openai", "POST", upstream, dict(HEADERS), BODY, 5) + head = cache.forward("openai", "POST", upstream, dict(HEADERS), BODY, 5, test_key=TEST_SLUG) assert isinstance(head, StreamHead) assert b"".join(step.data for step in head.steps if isinstance(step, StreamChunk)) == SUCCESS assert len(provider.hits) == 1 @@ -471,28 +485,101 @@ def test_another_test_never_reuses_this_tests_recording( with edge(cache_edge(store), provider) as url: call(url) assert len(provider.hits) == 1 - with edge(cache_edge(store, OTHER_TEST_KEY), provider) as url: + with edge(cache_edge(store), provider, OTHER_TEST_KEY) as url: call(url) assert len(provider.hits) == 2 - with edge(cache_edge(store, OTHER_TEST_KEY), provider) as url: + with edge(cache_edge(store), provider, OTHER_TEST_KEY) as url: call(url) assert len(provider.hits) == 2 -def test_calls_outside_any_test_are_never_cached( - store: RedisResponseStore, provider: Provider, +def test_a_request_without_a_test_segment_is_never_cached( + store: RedisResponseStore, provider: Provider, monkeypatch: pytest.MonkeyPatch, ) -> None: - url: Final = f"http://127.0.0.1:{provider.server_port}/v1/chat/completions" - cache: Final = CacheEdge(store, SECRET, test_key=lambda: SESSION_TEST_KEY) - for _ in range(2): - head = cache.forward("openai", "POST", url, dict(HEADERS), BODY, 5) - assert isinstance(head, StreamHead) - assert b"".join(step.data for step in head.steps if isinstance(step, StreamChunk)) == SUCCESS + """The bare mount URL is what a deployment registered outside any test would + carry. The serving process is inside a test here, and that must not count: + the edge never names the test from its own process state.""" + monkeypatch.setenv("PYTEST_CURRENT_TEST", f"{TEST_KEY} (call)") + cache: Final = cache_edge(store) + with edge(cache, provider, test_key=None) as url: + assert call(url).body == SUCCESS + assert call(url).body == SUCCESS assert len(provider.hits) == 2 assert dict(cache.counters.counts) == { "bypass": 2, "mount:openai:bypass": 2, "upstream_attempts": 2, "mount:openai:upstream_attempts": 2, } + with edge(cache_edge(store), provider) as url: + assert call(url).body == SUCCESS + assert len(provider.hits) == 3 + + +def test_attribution_comes_from_the_deployment_path_not_the_serving_process( + store: RedisResponseStore, provider: Provider, monkeypatch: pytest.MonkeyPatch, +) -> None: + """Under xdist the process serving a call is unrelated to the test that made + it: the proxy is a separate pod, and the compat matrix's shared aliases had + every worker's edge answering every other worker's cells. The recording must + land under the test whose deployment the request came through, whatever + ``PYTEST_CURRENT_TEST`` says in the edge's own process.""" + monkeypatch.setenv("PYTEST_CURRENT_TEST", f"{OTHER_TEST_KEY} (call)") + monkeypatch.setenv("E2E_PROVIDER_CACHE_METRICS_DIR", "unused-but-enables-the-probe") + first: Final = cache_edge(store) + with edge(first, provider) as url: + assert call(url).body == SUCCESS + assert len(provider.hits) == 1 + assert dict(first.probe.rows[0])["test_key"] == TEST_SLUG + monkeypatch.setenv("PYTEST_CURRENT_TEST", f"{TEST_KEY} (call)") + with edge(cache_edge(store), provider, OTHER_TEST_KEY) as url: + assert call(url).body == SUCCESS + assert len(provider.hits) == 2 + with edge(cache_edge(store), provider) as url: + assert call(url).body == SUCCESS + assert len(provider.hits) == 2 + + +@pytest.mark.parametrize("upstream_path,expected", [ + (f"t/{TEST_SLUG}/v1/chat/completions", (TEST_SLUG, "v1/chat/completions")), + (f"t/{TEST_SLUG}/model/{BEDROCK_MODEL}/converse-stream", (TEST_SLUG, f"model/{BEDROCK_MODEL}/converse-stream")), + ("v1/chat/completions", (None, "v1/chat/completions")), + (f"model/{BEDROCK_MODEL}/invoke", (None, f"model/{BEDROCK_MODEL}/invoke")), + ("t//v1/chat/completions", (None, "v1/chat/completions")), + ("t", (None, "")), +]) +def test_the_test_segment_is_read_off_the_path_and_never_reaches_the_provider( + upstream_path: str, expected: tuple[str | None, str], +) -> None: + assert split_test_segment(upstream_path) == expected + assert split_test_segment(scoped_edge_base("", TEST_KEY).lstrip("/") + "/v1/chat/completions") == ( + TEST_SLUG, "v1/chat/completions", + ) + + +def test_the_cache_edge_base_is_scoped_to_the_registering_test( + redis_url: str, monkeypatch: pytest.MonkeyPatch, tmp_path, +) -> 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() + + def base_for(test_key: str) -> str | None: + return provider_edge_api_base( + "openai", mode_raw="live", bundle_dir=tmp_path, bind_host="127.0.0.1", advertise_host="127.0.0.1", + test_key=test_key, + ) + + try: + scoped: Final = base_for(TEST_KEY) + assert scoped is not None and scoped.endswith(f"/openai/t/{TEST_SLUG}") + assert base_for(OTHER_TEST_KEY) != scoped + assert base_for(SESSION_TEST_KEY) is None + monkeypatch.setenv("E2E_PROVIDER_CACHE", "0") + configured_cache.cache_clear() + assert base_for(TEST_KEY) is None + finally: + configured_cache.cache_clear() def test_counters_attribute_every_outcome_to_its_mount( @@ -505,8 +592,8 @@ def test_counters_attribute_every_outcome_to_its_mount( cache: Final = cache_edge(store) running: Final = start_provider_edge(cache, mounts={"openai": upstream, "anthropic": upstream}) try: - call(running.edge.api_base("openai") + "/v1/chat/completions") - call(running.edge.api_base("anthropic") + "/v1/messages") + call(scoped_edge_base(running.edge.api_base("openai"), TEST_KEY) + "/v1/chat/completions") + call(scoped_edge_base(running.edge.api_base("anthropic"), TEST_KEY) + "/v1/messages") finally: running.shutdown() counts: Final = dict(cache.counters.counts) @@ -531,7 +618,7 @@ def test_a_rejection_says_whether_the_body_was_cut_short_or_simply_unfinished( provider.response = b'data: {"choices":[{"index":0,"delta":{"content":"hi"},"finish_reason":"stop"}]}\n\ndata: [DONE]\n\n' running: Final = start_provider_edge(cut_short, mounts={"openai": upstream}) try: - forward("POST", running.edge.api_base("openai") + "/v1/chat/completions", + forward("POST", scoped_edge_base(running.edge.api_base("openai"), TEST_KEY) + "/v1/chat/completions", headers=HEADERS, body=MARKED, timeout=5) finally: running.shutdown() @@ -542,7 +629,7 @@ def test_a_rejection_says_whether_the_body_was_cut_short_or_simply_unfinished( provider.response = b'{"choices":[{"index":0,"message":{"content":"hi"}}]}' second: Final = start_provider_edge(unfinished, mounts={"openai": upstream}) try: - call(second.edge.api_base("openai") + "/v1/chat/completions", MARKED) + call(scoped_edge_base(second.edge.api_base("openai"), TEST_KEY) + "/v1/chat/completions", MARKED) finally: second.shutdown() @@ -551,7 +638,7 @@ def test_a_rejection_says_whether_the_body_was_cut_short_or_simply_unfinished( provider.response = b'{"message":"Too many requests"}' third: Final = start_provider_edge(refused, mounts={"openai": upstream}) try: - call(third.edge.api_base("openai") + "/v1/chat/completions", MARKED) + call(scoped_edge_base(third.edge.api_base("openai"), TEST_KEY) + "/v1/chat/completions", MARKED) finally: third.shutdown() @@ -586,7 +673,7 @@ def openai_edge(cache: CacheEdge, provider: Provider, path: str) -> Generator[st 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") + path + yield scoped_edge_base(running.edge.api_base("openai"), TEST_KEY) + path finally: running.shutdown() @@ -787,7 +874,7 @@ class TestBedrockSigning: def signing_edge() -> CacheEdge: return CacheEdge( - store, SECRET, test_key=lambda: TEST_KEY, + store, SECRET, policies={BEDROCK_MOUNT: MountPolicy(sign=varying, unkeyed_headers=SIGNATURE_HEADERS)}, ) @@ -1065,7 +1152,8 @@ def test_connection_failure_releases_capture_lease(store: RedisResponseStore) -> unavailable.bind(("127.0.0.1", 0)) url: Final = f"http://127.0.0.1:{unavailable.getsockname()[1]}/v1/chat/completions" cache: Final = cache_edge(store) - assert isinstance(cache.forward("openai", "POST", url, dict(HEADERS), BODY, 0.2), NetworkError) + head: Final = cache.forward("openai", "POST", url, dict(HEADERS), BODY, 0.2, test_key=TEST_SLUG) + assert isinstance(head, NetworkError) key: Final = slot_key(url) lease: Final = store.lookup(key) assert isinstance(lease, CaptureLease) @@ -1076,7 +1164,7 @@ def test_connection_failure_releases_capture_lease(store: RedisResponseStore) -> 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 = cache_edge(store) - head: Final = cache.forward("openai", "POST", url, dict(HEADERS), BODY, 5) + head: Final = cache.forward("openai", "POST", url, dict(HEADERS), BODY, 5, test_key=TEST_SLUG) assert isinstance(head, StreamHead) head.steps.close() key: Final = slot_key(url) @@ -1094,7 +1182,7 @@ def test_effective_account_change_cannot_reuse_cache( 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("openai", "POST", url, dict(HEADERS), BODY, 5) + head = cache.forward("openai", "POST", url, dict(HEADERS), BODY, 5, test_key=TEST_SLUG) assert isinstance(head, StreamHead) assert b"".join(step.data for step in head.steps if isinstance(step, StreamChunk)) == SUCCESS assert len(provider.hits) == 2 diff --git a/tests/e2e/PROVIDER_CACHE.md b/tests/e2e/PROVIDER_CACHE.md index d37b37eeba7..d8da807c643 100644 --- a/tests/e2e/PROVIDER_CACHE.md +++ b/tests/e2e/PROVIDER_CACHE.md @@ -1,6 +1,6 @@ # 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, as do Anthropic-on-Bedrock registrations that carry no AWS identity of their own. Existing custom API bases, named credentials, mocked models and realtime WebSocket deployments keep their existing routing. Other provider protocols remain live +`E2E_PROVIDER_CACHE=1` enables automatic response reuse in the live E2E mode. Standard OpenAI and Anthropic model registrations made from inside a test use the provider edge, as do Anthropic-on-Bedrock registrations that carry no AWS identity of their own. A registration made outside any test, or with `provider_live=True`, keeps its real provider path. 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`, `/v1/messages`, `/v1/embeddings` and `/v1/responses` on the OpenAI and Anthropic mounts, SSE streams included, and for `/model/{id}/converse` and `/model/{id}/invoke` on a Bedrock mount. Unsupported endpoints pass through. Each endpoint family has its own completeness rule, so a truncated embedding or a Responses run that never reached `response.completed` is not stored @@ -10,13 +10,13 @@ Two details of that rule are worth knowing before changing it. A ConverseStream ## Request identity -A recording belongs to one test. The key is a keyed digest over the test's node id, the method, the URL, the effective outbound headers (including authentication and HTTP-library defaults), body presence and the body bytes, with one normalization: a 12-hex-digit run, the shape `unique_marker()` mints, is replaced by a placeholder in both the URL and a UTF-8 body. Nothing else is normalized away. No prompts, JSON values or credentials are rewritten, and the rule is the one `fixture_canonical.py` already applies for record/replay, so there is a single definition of what a marker is +A recording belongs to one test, and the test is named by the deployment rather than by the process. A deployment registered from inside a test gets the cache edge's mount URL with a test segment appended, `{edge}/{mount}/t/{slug}`, where the slug is `slug_for_test` of the registering test's node id, and the edge reads that segment off every request before forwarding. The key is a keyed digest over that slug, the method, the upstream URL, the effective outbound headers (including authentication and HTTP-library defaults), body presence and the body bytes, with one normalization: a 12-hex-digit run, the shape `unique_marker()` mints, is replaced by a placeholder in both the URL and a UTF-8 body. Nothing else is normalized away. No prompts, JSON values or credentials are rewritten, and the rule is the one `fixture_canonical.py` already applies for record/replay, so there is a single definition of what a marker is Requests that differ only by their markers therefore share a canonical identity, which is what makes the cache reusable across builds: every e2e test salts its prompt afresh, so an exact-byte key would miss on every call. Within one test, calls that share a canonical identity are still recorded and replayed separately, by a FIFO slot index appended to the key. That matters because a replayed response carries the recorded provider response id, `LiteLLM_SpendLogs.request_id` is that id, and one shared recording answering two calls would collapse two spend rows into one -Two different tests never share a recording, and a provider call made outside any test (fixtures, session setup) is never cached, because the identity has no test node id to bind to +Two different tests never share a recording. A request that reaches the edge without a test segment is forwarded live and never cached, and the edge never names the test from its own process's `PYTEST_CURRENT_TEST`. It used to, and that was wrong whenever the calling test and the serving process differed: the proxy is a separate pod, and under xdist the Claude Code compat matrix registered its shared aliases from every worker, each pointing at that worker's edge, so the router spread one worker's calls across all of them and each call was keyed on whatever test the serving worker was in. Builds 234 and 235 of the e2e pipeline, same commit, credited the same Bedrock request to unrelated tests 92% of the time, which is why that mount never converged -A client that varies its own request between runs defeats that identity without breaking any rule, and the Claude Code compat cells did. The CLI sends a device id and a session id in `metadata.user_id`, and its system prompt names both its memory directory and its working directory, adding the branch and recent commits when that directory is a git repository. Driven with a fresh HOME and the checkout as its working directory, every cell sent different bytes every build. The fix belongs in the driver rather than here: `claude_code/cli_driver.py` pins the config directory, the working directory and both identifiers, which is why the cache needs no rule for any of it. Normalizing them instead would have hidden a real defect class, since a rule cannot tell a client's own churn from a value a test means to assert on +The Claude Code compat cells are not cached. Their aliases are registered once per worker session and shared by every cell, so no call to them belongs to one test, and the matrix exists to prove the real CLI against real providers; `claude_code/conftest.py` registers them with `provider_live=True`. The driver still pins the CLI's config directory, working directory, device id and session id (`_driver_unit_tests/test_request_determinism.py` holds that), so a CLI-driven deployment registered by one test would send stable bytes. Normalizing those values in the key instead would hide a real defect class, since a rule cannot tell a client's own churn from a value a test means to assert on Provider `Set-Cookie` headers are dropped before validation and never recorded: the edge already withholds them from the proxy, and OpenAI responses always carry Cloudflare bot-management cookies @@ -62,4 +62,4 @@ Provider remaining-quota headers describe the captured response. Metrics derived ## Qualification -`tests/code_coverage_tests/test_provider_cache.py` exercises local HTTP providers and disposable real Redis, including the marker-canonical key, the FIFO slot index, per-test isolation, SigV4 re-signing against a local upstream, and each endpoint's completeness rule. 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 +`tests/code_coverage_tests/test_provider_cache.py` exercises local HTTP providers and disposable real Redis, including the marker-canonical key, the FIFO slot index, per-test isolation, attribution from the deployment's test segment whatever the serving process is running, SigV4 re-signing against a local upstream, and each endpoint's completeness rule. 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/claude_code/conftest.py b/tests/e2e/claude_code/conftest.py index 6e3dce0377e..bf226161267 100644 --- a/tests/e2e/claude_code/conftest.py +++ b/tests/e2e/claude_code/conftest.py @@ -600,10 +600,13 @@ def _build_control_plane_client(proxy_config: ProxyConfig): def _register_deployment(proxy, deployment: CompatDeployment) -> str: """Register one deployment and return its proxy-assigned model_id - once it is servable on the data plane.""" + once it is servable on the data plane. The aliases are shared by every + cell and, under xdist, by every worker, so no call to them belongs to + one test and none is cached: the matrix exists to reach real providers.""" return proxy.create_model( deployment.model_name, deployment.litellm_params, + provider_live=True, ) diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py index ac4cfb71407..e17a589af47 100644 --- a/tests/e2e/conftest.py +++ b/tests/e2e/conftest.py @@ -124,8 +124,7 @@ def pytest_configure(config: pytest.Config) -> None: ) config.addinivalue_line( "markers", - "cli_determinism: drives the real claude CLI for several seconds, which widens the window in which " - "another test's in-flight upstream call is attributed to it; deselected unless E2E_CLI_DETERMINISM is set", + "cli_determinism: drives the real claude CLI for several seconds; deselected unless E2E_CLI_DETERMINISM is set", ) config.addinivalue_line( "markers", diff --git a/tests/e2e/e2e_config.py b/tests/e2e/e2e_config.py index 82ddb09f7f5..e228d6c018f 100644 --- a/tests/e2e/e2e_config.py +++ b/tests/e2e/e2e_config.py @@ -13,7 +13,7 @@ from pathlib import Path from typing import Final from dotenv import load_dotenv -from fixture_mode import deterministic_marker, parse_fixture_mode +from fixture_mode import current_test_key, deterministic_marker, parse_fixture_mode from provider_edge import provider_edge_api_base # Local runs keep provider / DataDog keys in tests/e2e/.env (see CONTRIBUTING.md). @@ -200,13 +200,15 @@ def datadog_mcp_url(*, toolsets: str = "core") -> str: def provider_edge_base(mount: str) -> str | None: """The api_base an edge-wired deployment should register with, using this process's fixture-mode and edge-host configuration: None in live mode, the - shared edge server's mount URL in record and replay.""" + shared edge server's mount URL in record and replay, and with the shared + cache on, the cache edge's mount URL scoped to the running test.""" return provider_edge_api_base( mount, mode_raw=FIXTURE_MODE_RAW, bundle_dir=FIXTURE_DIR, bind_host=PROVIDER_EDGE_BIND_HOST, advertise_host=PROVIDER_EDGE_ADVERTISE_HOST, + test_key=current_test_key(), forward_timeout=REQUEST_TIMEOUT, ) diff --git a/tests/e2e/provider_cache.py b/tests/e2e/provider_cache.py index 22967869c78..55e9f9322c7 100644 --- a/tests/e2e/provider_cache.py +++ b/tests/e2e/provider_cache.py @@ -27,8 +27,8 @@ from e2e_http import ( prepare_forward, primed_steps, ) +from fixture_bundle import slug_for_test from fixture_canonical import MARKER_PATTERN, MARKER_PLACEHOLDER -from fixture_mode import SESSION_TEST_KEY, current_test_key from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter, ValidationError LIFETIME_SECONDS: Final = 86_400 @@ -58,6 +58,19 @@ EVENT_TYPE_HEADER: Final = ":event-type" EVENTSTREAM_HEADERS: Final[TypeAdapter[dict[str, str]]] = TypeAdapter(dict[str, str]) OPENAI_JSON_PATHS: Final = frozenset({"/v1/chat/completions", "/v1/messages", "/v1/embeddings", "/v1/responses"}) JSON_VALUE: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue) +TEST_SEGMENT: Final = "t" + + +def scoped_edge_base(base: str, test_key: str) -> str: + return f"{base}/{TEST_SEGMENT}/{slug_for_test(test_key)}" + + +def split_test_segment(upstream_path: str) -> tuple[str | None, str]: + head, _, rest = upstream_path.partition("/") + if head != TEST_SEGMENT: + return None, upstream_path + slug, _, remainder = rest.partition("/") + return slug or None, remainder @dataclass(frozen=True, slots=True) @@ -517,7 +530,6 @@ class CacheEdge: wait_seconds: float = 2.0 clock: Callable[[], float] = time.monotonic sleep: Callable[[float], None] = time.sleep - test_key: Callable[[], str] = current_test_key def lookup(self, key: str) -> CacheLookup: deadline: Final = self.clock() + self.wait_seconds @@ -555,9 +567,9 @@ class CacheEdge: def forward( self, mount: str, method: str, url: str, headers: dict[str, str], body: bytes | None, timeout: float, + *, test_key: str | None, ) -> StreamHead | NetworkError: - test_key: Final = self.test_key() - if test_key == SESSION_TEST_KEY or not cacheable_endpoint(mount, method, url, body): + if test_key is None or not cacheable_endpoint(mount, method, url, body): self.count(mount, "bypass") self.count(mount, "upstream_attempts") return forward_stream( diff --git a/tests/e2e/provider_edge.py b/tests/e2e/provider_edge.py index 2606b26fe99..224cf77159e 100644 --- a/tests/e2e/provider_edge.py +++ b/tests/e2e/provider_edge.py @@ -88,13 +88,21 @@ from fixture_canonical import ( ) from fixture_mode import ( FIXTURE_MODES, + SESSION_TEST_KEY, InvalidFixtureMode, ReplayMiss, current_test_key, parse_fixture_mode, ) from fixture_profile import IneligibleRequest, MatchProfile, match_profile, strict_identity -from provider_cache import SIGNATURE_HEADERS, CacheEdge, MountPolicy, is_bedrock +from provider_cache import ( + SIGNATURE_HEADERS, + CacheEdge, + MountPolicy, + is_bedrock, + scoped_edge_base, + split_test_segment, +) from provider_cache_routing import LIVE_PROVIDER_REQUIRED from pydantic import JsonValue, TypeAdapter @@ -778,14 +786,14 @@ def _handle_record( def _handle_live( method: str, url: str, headers: Mapping[str, str], body: bytes | None, timeout: float, - cache: CacheEdge | None = None, mount: str = "", + cache: CacheEdge | None = None, mount: str = "", test_key: str | 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) - if cache is None else cache.forward(mount, method, url, forwarded, body, timeout) + if cache is None else cache.forward(mount, method, url, forwarded, body, timeout, test_key=test_key) ) match head: case NetworkError(message=message): @@ -826,7 +834,7 @@ def handle_edge_request( return _text_reply(404, f"unknown provider mount {unknown!r}; known mounts: {', '.join(sorted(mounts))}") mount: Final = resolved.mount upstream_base: Final = resolved.upstream_base - upstream_path: Final = resolved.upstream_path + test_key, upstream_path = split_test_segment(resolved.upstream_path) profile: Final = ( backend.recorder.profile if isinstance(backend, RecordEdge) @@ -858,7 +866,7 @@ def handle_edge_request( case CacheEdge(): return _handle_live( method, _upstream_url(upstream_base, upstream_path, split.query), headers, body, timeout, - backend, mount, + backend, mount, test_key, ) case LiveEdge(): return _handle_live( @@ -1093,19 +1101,25 @@ def provider_edge_api_base( bundle_dir: Path, bind_host: str, advertise_host: str, + test_key: str, forward_timeout: float = 60.0, ) -> str | None: """The api_base a suite gives an edge-wired deployment: None in live mode (the deployment keeps its real provider api_base) and the process-wide edge - server's mount URL in record and replay, booting the server on first use.""" + server's mount URL in record and replay, booting the server on first use. + With the shared cache configured, live mode answers with the cache edge's + mount URL scoped to ``test_key``, the test registering the deployment, and + None outside any test, since a call nobody can attribute is never cached.""" 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": - if configured_cache_backend() is not None: - return _shared_cache_edge(bind_host, advertise_host, forward_timeout).api_base(mount) - return None + if configured_cache_backend() is None or test_key == SESSION_TEST_KEY: + return None + return scoped_edge_base( + _shared_cache_edge(bind_host, advertise_host, forward_timeout).api_base(mount), test_key + ) case "record" | "replay": if is_bedrock(mount): return None diff --git a/tests/e2e/proxy_client.py b/tests/e2e/proxy_client.py index f8ed8843461..44d9df5e5c5 100644 --- a/tests/e2e/proxy_client.py +++ b/tests/e2e/proxy_client.py @@ -613,6 +613,8 @@ class ProxyClient: model_name: str, litellm_params: LiteLLMParamsBody, mode: ModelMode | None = None, + *, + provider_live: bool = False, ) -> str: """Register a deployment under `model_name` and return its proxy-assigned model_id, once the model is actually servable on the data plane.""" @@ -621,15 +623,20 @@ class ProxyClient: model_name=model_name, litellm_params=litellm_params, model_info=ModelInfoBody(mode=mode), - ) + ), + provider_live=provider_live, ) - def register_model(self, body: ModelNewBody, listed_for: str | None = None) -> str: + def register_model( + self, body: ModelNewBody, listed_for: str | None = None, *, provider_live: bool = False + ) -> str: """`create_model` for deployments that carry more than a mode: access groups, team scoping, a pinned id. `listed_for` is the virtual key whose /v1/models view must list the deployment before it counts as servable, because a team-scoped deployment is listed to its own team and to nobody else, master - key included; leave it unset for a proxy-wide model. + key included; leave it unset for a proxy-wide model. `provider_live` keeps + the deployment on its real provider path whatever the cache setting, for a + deployment shared across tests or workers, which no one test could own. /model/new is a control-plane route; the data plane (which serves /chat, /ocr, ...) only picks the new model up on its next DB reload, so a call @@ -650,7 +657,8 @@ class ProxyClient: headers=self.management_headers(), 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, + enabled=os.environ.get("E2E_PROVIDER_CACHE", "0") == "1" and not provider_live, + mode=body.model_info.mode, )}), response_type=ModelNewResponse, ) diff --git a/tests/e2e/pytest.ini b/tests/e2e/pytest.ini index f6d23a3ec12..f9e5995079b 100644 --- a/tests/e2e/pytest.ini +++ b/tests/e2e/pytest.ini @@ -10,5 +10,5 @@ markers = weekly: real-provider anomaly load test that spends real money; deselected unless E2E_WEEKLY_ANOMALY is set managed_files: needs a proxy running with require_managed_files enabled; deselected unless E2E_MANAGED_FILES_STACK is set prompt_caching_stack: needs a proxy running with router_settings.optional_pre_call_checks including prompt_caching; deselected unless E2E_PROMPT_CACHING_STACK is set - cli_determinism: drives the real claude CLI for several seconds, which widens the window in which another test's in-flight upstream call is attributed to it; deselected unless E2E_CLI_DETERMINISM is set + cli_determinism: drives the real claude CLI for several seconds; deselected unless E2E_CLI_DETERMINISM is set redis_chaos: load test that pauses the proxy's Redis outright mid-run; needs a proxy booted from gateway/redis_chaos_ci_config.yml on the same host, and is deselected unless E2E_REDIS_CHAOS is set diff --git a/tests/e2e/test_provider_edge.py b/tests/e2e/test_provider_edge.py index c8d70697182..d776c338ef7 100644 --- a/tests/e2e/test_provider_edge.py +++ b/tests/e2e/test_provider_edge.py @@ -1264,6 +1264,7 @@ class TestApiBaseSeam: bundle_dir=tmp_path / "bundle", bind_host="127.0.0.1", advertise_host="127.0.0.1", + test_key="tests/e2e/synthetic_suite.py::test_case", ) is None ) @@ -1276,6 +1277,7 @@ class TestApiBaseSeam: bundle_dir=tmp_path / "bundle", bind_host="127.0.0.1", advertise_host="127.0.0.1", + test_key="tests/e2e/synthetic_suite.py::test_case", ) def test_unknown_mount_raises_naming_the_known_mounts(self, tmp_path: Path) -> None: @@ -1286,6 +1288,7 @@ class TestApiBaseSeam: bundle_dir=tmp_path / "bundle", bind_host="127.0.0.1", advertise_host="127.0.0.1", + test_key="tests/e2e/synthetic_suite.py::test_case", ) @pytest.mark.parametrize("mode_raw", ["record", "replay"]) @@ -1301,15 +1304,18 @@ class TestApiBaseSeam: bundle_dir=tmp_path / "bundle", bind_host="127.0.0.1", advertise_host="127.0.0.1", + test_key="tests/e2e/synthetic_suite.py::test_case", ) is None def test_record_mode_boots_one_shared_edge_and_prepares_the_bundle(self, tmp_path: Path) -> None: root = tmp_path / "bundle" first = provider_edge_api_base( - "openai", mode_raw="record", bundle_dir=root, bind_host="127.0.0.1", advertise_host="127.0.0.1" + "openai", mode_raw="record", bundle_dir=root, bind_host="127.0.0.1", advertise_host="127.0.0.1", + test_key="tests/e2e/synthetic_suite.py::test_case", ) second = provider_edge_api_base( - "anthropic", mode_raw="record", bundle_dir=root, bind_host="127.0.0.1", advertise_host="127.0.0.1" + "anthropic", mode_raw="record", bundle_dir=root, bind_host="127.0.0.1", advertise_host="127.0.0.1", + test_key="tests/e2e/synthetic_suite.py::test_case", ) assert first is not None and second is not None assert first.endswith("/openai") diff --git a/tests/e2e/test_proxy_client.py b/tests/e2e/test_proxy_client.py index 0c4aed5bd65..fc13e701fc9 100644 --- a/tests/e2e/test_proxy_client.py +++ b/tests/e2e/test_proxy_client.py @@ -26,6 +26,8 @@ from typing import Final, cast import pytest from e2e_config import parse_replica_urls from e2e_http import NoBody, Result, Success, without_retries +from fixture_bundle import slug_for_test +from fixture_mode import current_test_key from idp import Keycloak from lifecycle import ResourceManager from management.jwt_actors import ActorFactory @@ -581,3 +583,35 @@ def test_partial_updates_preserve_explicit_null_at_the_http_boundary(operation: ) assert json.loads(bodies.get_nowait()) == expected assert bodies.empty() + + +@pytest.mark.parametrize("provider_live", (False, True)) +def test_registration_binds_the_deployment_to_this_test_unless_it_is_provider_live( + provider_live: bool, monkeypatch: pytest.MonkeyPatch, +) -> None: + """With the shared cache on, a deployment registered from inside a test carries + this test's segment in its api_base, which is how the edge knows whose recording + a call belongs to. `provider_live` is the opt-out for a deployment no single test + owns: it goes to the proxy exactly as written.""" + from provider_cache_redis import configured_cache + + monkeypatch.setenv("E2E_PROVIDER_CACHE", "1") + monkeypatch.setenv("E2E_PROVIDER_CACHE_REDIS_URL", "redis://127.0.0.1:1/0") + monkeypatch.setenv("E2E_PROVIDER_CACHE_HMAC_KEY", "synthetic-cache-hmac-key-for-tests") + monkeypatch.setenv("E2E_PROVIDER_CACHE_NAMESPACE", "registration-seam") + configured_cache.cache_clear() + bodies: Final[SimpleQueue[bytes]] = SimpleQueue() + try: + with caller_boundary(status=401, bodies=bodies) as (bootstrap, _), without_retries(): + with pytest.raises(AssertionError): + bootstrap.proxy.create_model( + "owned", LiteLLMParamsBody(model="openai/synthetic"), provider_live=provider_live + ) + finally: + configured_cache.cache_clear() + params: Final = json.loads(bodies.get_nowait())["litellm_params"] + assert bodies.empty() + if provider_live: + assert params.get("api_base") is None + return + assert params["api_base"].endswith(f"/openai/t/{slug_for_test(current_test_key())}/v1")