mirror of
https://github.com/BerriAI/litellm.git
synced 2026-08-28 05:25:59 +00:00
test(e2e): record and replay the non-streaming provider flows
Chat completions, embeddings, the non-streaming /v1/messages tests, and the OpenAI batch deployment now register through the provider edge, so E2E_FIXTURE_MODE=record captures their provider calls and replay serves them back offline. None of them was wired before, so record was a silent no-op over these suites and replay quietly went live instead of using the bundle Multipart uploads now key on their parsed parts: every ordinary form field, plus the field name, filename, content digest, and length of each file part. The boundary is envelope rather than content, so it stays out of the digest instead of changing the key on every run. A body that does not parse as its declared envelope still has the boundary normalized away before hashing, so the fallback is at least stable, and it records a name that says why Binary uploads hash byte for byte. Canonicalizing them first meant decoding with errors="replace", which collapsed every invalid byte to one U+FFFD and gave two different PDFs of the same length the same key Bundles stay out of the repo: they hold verbatim provider response bodies and expire seven days after recording. Publishing them for CI is LIT-5748, and streaming fidelity is LIT-5742
This commit is contained in:
mateo-berri
parent
7cb100af63
commit
d4162bd1ca
9 changed files with 475 additions and 65 deletions
|
|
@ -77,13 +77,24 @@ Mark live tests with `@pytest.mark.e2e` (on the class or the module). Pure cover
|
|||
|
||||
The seam is `provider_edge.py`: `start_provider_edge` boots an in-process HTTP server (one shared instance per pytest process, `e2e_config.provider_edge_base` is the accessor) that mounts each supported provider under a path prefix (`EDGE_MOUNTS`: `/openai` -> `https://api.openai.com`, `/anthropic` -> `https://api.anthropic.com`). A test participates by registering its deployment with `api_base=provider_edge_base("openai")` plus the provider's path suffix; `quota_management/spend_tracking/test_provider_edge_spend_e2e.py` is the reference. In live mode the accessor returns None and the deployment defaults to the real provider, so an edge-wired test runs in all three modes unchanged. Non-wired tests hit their providers live in every mode. The edge binds `E2E_PROVIDER_EDGE_BIND_HOST` (default 127.0.0.1) and advertises `E2E_PROVIDER_EDGE_ADVERTISE_HOST` in the api_base it hands out, for proxies running in containers
|
||||
|
||||
A bundle (default `tests/e2e/.fixtures`, override with `E2E_FIXTURE_DIR`) is a directory: `manifest.json` carries the record timestamp, harness git version, and format version, and each test gets a subdirectory holding one JSON file per provider call in call order (`0000-post-openai-v1-chat-completions.json`). Request headers are never stored (provider credentials never touch disk), non-JSON request bodies store a canonicalized sha256 digest instead of the bytes, and responses store status, filtered headers, and the verbatim body base64-encoded, which is part of why bundles are gitignored. `fixture_bundle.py` owns the format. Record serves the proxy the same filtered stored response replay will serve later, so the two modes are byte-identical from the proxy's side of the socket
|
||||
A bundle (default `tests/e2e/.fixtures`, override with `E2E_FIXTURE_DIR`) is a directory: `manifest.json` carries the record timestamp, harness git version, and format version, and each test gets a subdirectory holding one JSON file per provider call in call order (`0000-post-openai-v1-chat-completions.json`). Request headers are never stored (provider credentials never touch disk), non-JSON request bodies store a canonicalized sha256 digest instead of the bytes, `multipart/form-data` bodies store their ordinary fields plus a `field:filename` label, a digest of the file part's content, and that part's length, so the per-request random boundary and the envelope never reach the key, and responses store status, filtered headers, and the verbatim body base64-encoded, which is part of why bundles are gitignored. `fixture_bundle.py` owns the format. Record serves the proxy the same filtered stored response replay will serve later, so the two modes are byte-identical from the proxy's side of the socket
|
||||
|
||||
Replay matches calls per test by canonical key: `fixture_canonical.py` canonicalizes the recorded request (volatile headers and credential fields out, unique markers, generated ids, uuids, and timestamps replaced with fixed placeholders, object keys sorted) and the key is the method, edge path, and a content hash, so identity survives re-records and machine changes while any real content drift comes back as an HTTP 599 naming the computed key, the closest recorded key with its file, and a content diff, and never falls through to a live call. Matching is order-independent across distinct keys (concurrent calls may interleave) and FIFO within one key (a retry loop replays its responses in recorded order); a passed test must also consume its whole recording, or teardown fails it naming a leftover key. Either way the fix is always to re-record with `E2E_FIXTURE_MODE=record`. Every rewrite rule lives in `fixture_canonical.py`, so a new volatile header, credential field name, or generated-id shape is one edit there. Record starts fresh every time: it wipes the previous bundle (refusing to wipe a directory that is not a bundle) and never reads it. A replay bundle whose manifest is older than seven days hard-fails at collection time naming the bundle's age, so replay can never certify against fixtures that have drifted more than a week from the live providers
|
||||
|
||||
A replayed response carries the recorded provider response id, and `LiteLLM_SpendLogs.request_id` (the table's primary key) is that id, so a replay against a database that still holds the record run's rows silently dedupes its spend inserts and any spend assertion goes red with zero matching rows and nothing in the proxy log. Run both modes with `E2E_RESET_SPEND_LOGS=1` (plus `DATABASE_URL` in the runner env) so each session truncates the table after itself, or replay against a fresh database, which is the CI shape
|
||||
|
||||
Current limits: streaming chunk fidelity is LIT-5742 (a streamed response records as one buffered body), CI wiring is LIT-5748, Bedrock cannot be mounted (SigV4 signs the Host header, so a rewritten api_base fails signature verification), multipart uploads have per-run random boundaries (the digest changes every run, so they always miss), and deployments baked into the proxy's config file cannot be edge-wired (only `/model/new` registrations can carry the edge api_base)
|
||||
The same id reuse reaches the managed-object tables. A replayed `/v1/files` or `/v1/batches` response carries the recorded provider object id, and `LiteLLM_ManagedObjectTable.model_object_id` is unique, so a unified batch create replayed against a database that still holds the record run's row fails on a Prisma unique-constraint violation, which surfaces as a 500, makes the router retry, and exhausts the recording. Replay the batches suite against a fresh database, or truncate `LiteLLM_ManagedObjectTable` and `LiteLLM_ManagedFileTable` before the run
|
||||
|
||||
Edge-wired today: `quota_management/spend_tracking/test_provider_edge_spend_e2e.py` (the reference), `llm_translation/test_chat_completions_contract_e2e.py`, the OpenAI registrations in `llm_translation/test_embeddings_endpoint_e2e.py`, the Anthropic deployments in `llm_translation/test_messages_e2e.py` except the streaming test, and the OpenAI batch deployment behind `batches/` (`capabilities.openai_batch_params`). The mount base is not the same for both providers: OpenAI deployments register `f"{base}/v1"`, Anthropic deployments register `base` on its own, because litellm's Anthropic handler appends `/v1/messages` to `api_base` itself where the OpenAI handler appends only `/chat/completions`. Recording one suite locally is two runs against a proxy you already have up:
|
||||
|
||||
```bash
|
||||
E2E_FIXTURE_MODE=record E2E_FIXTURE_DIR=/tmp/e2e-fixtures E2E_RESET_SPEND_LOGS=1 uv run pytest tests/e2e/llm_translation/test_chat_completions_contract_e2e.py
|
||||
E2E_FIXTURE_MODE=replay E2E_FIXTURE_DIR=/tmp/e2e-fixtures E2E_RESET_SPEND_LOGS=1 uv run pytest tests/e2e/llm_translation/test_chat_completions_contract_e2e.py
|
||||
```
|
||||
|
||||
Point the proxy at bogus provider credentials for the replay run and it still has to pass: that is the whole proof that nothing left the process. Bundles are never committed. `tests/e2e/.fixtures` is gitignored because a bundle holds verbatim provider response bodies and hard-fails after seven days, and publishing one for CI is LIT-5748
|
||||
|
||||
Current limits: streaming chunk fidelity is LIT-5742 (a streamed response records as one buffered body), CI wiring is LIT-5748, Bedrock cannot be mounted (SigV4 signs the Host header, so a rewritten api_base fails signature verification), deployments baked into the proxy's config file cannot be edge-wired (only `/model/new` registrations can carry the edge api_base), and a file upload routed by `custom_llm_provider` through the proxy's `files_settings` block never passes a deployment at all, so the batches `model_param` and `provider_fallback` scenarios keep uploading live in every mode
|
||||
|
||||
## Typing
|
||||
|
||||
|
|
|
|||
|
|
@ -57,13 +57,15 @@ Some suites need extra services the bare proxy does not start. The `logging/` OT
|
|||
Record/replay scopes to the proxy's provider-bound traffic only. In `E2E_FIXTURE_MODE=record` the harness boots a local provider-edge server, edge-wired tests register their deployments with an `api_base` pointing at it, and every provider call the proxy makes is forwarded verbatim and written to a fixture bundle (default `tests/e2e/.fixtures`, override with `E2E_FIXTURE_DIR`). `E2E_FIXTURE_MODE=replay` runs the same tests against the same live proxy and database, but the edge answers the proxy's provider calls from the bundle instead of the provider, so the run makes zero provider calls and spends nothing while key auth, routing, cost calculation, and spend-log writes all still execute for real. Unset (or `live`) behaves exactly as before the knob existed. Both record and replay need the proxy up; only the provider is taken out of the loop
|
||||
|
||||
```bash
|
||||
E2E_FIXTURE_MODE=record uv run pytest tests/e2e/quota_management/spend_tracking/test_provider_edge_spend_e2e.py -v
|
||||
E2E_FIXTURE_MODE=replay uv run pytest tests/e2e/quota_management/spend_tracking/test_provider_edge_spend_e2e.py -v
|
||||
E2E_FIXTURE_MODE=record E2E_FIXTURE_DIR=/tmp/e2e-fixtures uv run pytest tests/e2e/quota_management/spend_tracking/test_provider_edge_spend_e2e.py -v
|
||||
E2E_FIXTURE_MODE=replay E2E_FIXTURE_DIR=/tmp/e2e-fixtures uv run pytest tests/e2e/quota_management/spend_tracking/test_provider_edge_spend_e2e.py -v
|
||||
```
|
||||
|
||||
Bundles stay local. `tests/e2e/.fixtures` is gitignored because a bundle holds verbatim provider response bodies and expires seven days after it was recorded, so record the suite you want before you replay it and never commit the result; publishing bundles for CI is LIT-5748
|
||||
|
||||
One sharp edge: a replayed response reuses the recorded provider response id, and that id is the primary key of `LiteLLM_SpendLogs`, so replaying against a database that still holds the record run's rows silently dedupes the spend writes and a spend assertion fails with zero rows. Run both commands above with `E2E_RESET_SPEND_LOGS=1` (and `DATABASE_URL` set in the pytest env) so each session truncates the spend log table after itself, or point replay at a fresh database
|
||||
|
||||
Replay answers any provider call that drifted from the recording with an HTTP 599 whose body names the computed and closest recorded keys, so the test fails loudly instead of silently going live, and a bundle older than seven days fails at collection time naming its age; either way the fix is to re-record. Only tests that register edge-wired deployments participate: everything else hits its provider live in every mode, so record exactly the suite you replay. If the proxy runs in a container, set `E2E_PROVIDER_EDGE_ADVERTISE_HOST` (e.g. `host.docker.internal`) so the api_base the proxy stores can reach the edge on the pytest host, and `E2E_PROVIDER_EDGE_BIND_HOST=0.0.0.0` so the edge accepts it. See `CLAUDE.md` in this directory for the bundle format, the edge design, and the current limits (streaming, Bedrock, multipart)
|
||||
Replay answers any provider call that drifted from the recording with an HTTP 599 whose body names the computed and closest recorded keys, so the test fails loudly instead of silently going live, and a bundle older than seven days fails at collection time naming its age; either way the fix is to re-record. Only tests that register edge-wired deployments participate: everything else hits its provider live in every mode, so record exactly the suite you replay. If the proxy runs in a container, set `E2E_PROVIDER_EDGE_ADVERTISE_HOST` (e.g. `host.docker.internal`) so the api_base the proxy stores can reach the edge on the pytest host, and `E2E_PROVIDER_EDGE_BIND_HOST=0.0.0.0` so the edge accepts it. The suites wired to the edge today are `quota_management/spend_tracking/test_provider_edge_spend_e2e.py`, `llm_translation/test_chat_completions_contract_e2e.py`, the OpenAI registrations in `llm_translation/test_embeddings_endpoint_e2e.py`, the non-streaming Anthropic tests in `llm_translation/test_messages_e2e.py`, and the OpenAI batch deployment behind `batches/`. See `CLAUDE.md` in this directory for the bundle format, the edge design, and the current limits (streaming, Bedrock)
|
||||
|
||||
Tests marked `@pytest.mark.e2e` hard-fail when no proxy answers `/health/liveliness`, so a run that goes red with `No live proxy` at setup means the proxy isn't up; they never skip for a missing proxy, so an absent proxy can't be mistaken for a pass
|
||||
|
||||
|
|
|
|||
|
|
@ -7,7 +7,7 @@ import os
|
|||
from dataclasses import dataclass
|
||||
from typing import Literal
|
||||
|
||||
from e2e_config import unique_marker
|
||||
from e2e_config import provider_edge_base, unique_marker
|
||||
from models import LiteLLMParamsBody
|
||||
|
||||
_BATCH_RUN = unique_marker()
|
||||
|
|
@ -17,6 +17,18 @@ def batch_model_name(base: str) -> str:
|
|||
return f"{base}-{_BATCH_RUN}"
|
||||
|
||||
|
||||
def openai_batch_params() -> LiteLLMParamsBody:
|
||||
"""The OpenAI batch deployment, wired through the record/replay edge when a fixture
|
||||
mode is active and straight at OpenAI otherwise (LIT-5974). Azure, Vertex, and
|
||||
Bedrock stay live: none of them has an edge mount."""
|
||||
base = provider_edge_base("openai")
|
||||
return LiteLLMParamsBody(
|
||||
model="openai/gpt-4o-mini",
|
||||
api_key="os.environ/OPENAI_API_KEY",
|
||||
api_base=None if base is None else f"{base}/v1",
|
||||
)
|
||||
|
||||
|
||||
def _env_ref(*names: str) -> str:
|
||||
for name in names:
|
||||
value = os.environ.get(name)
|
||||
|
|
@ -47,10 +59,7 @@ class Provider:
|
|||
def litellm_params(self) -> LiteLLMParamsBody:
|
||||
match self.name:
|
||||
case "openai":
|
||||
return LiteLLMParamsBody(
|
||||
model="openai/gpt-4o-mini",
|
||||
api_key="os.environ/OPENAI_API_KEY",
|
||||
)
|
||||
return openai_batch_params()
|
||||
case "azure":
|
||||
return LiteLLMParamsBody(
|
||||
model="azure/gpt-5.4-mini-batch",
|
||||
|
|
|
|||
|
|
@ -51,6 +51,7 @@ from capabilities import (
|
|||
decoded_model_from_id,
|
||||
is_managed_id,
|
||||
matches_id_shape,
|
||||
openai_batch_params,
|
||||
raw_id_matches_provider,
|
||||
)
|
||||
from e2e_http import (
|
||||
|
|
@ -506,13 +507,7 @@ class TestBatchFileContent:
|
|||
self, client: BatchClient, resources: ResourceManager
|
||||
) -> None:
|
||||
proxy_name = f"e2e-file-content-{unique_marker()}"
|
||||
model_id = client.create_model(
|
||||
proxy_name,
|
||||
LiteLLMParamsBody(
|
||||
model=f"openai/{OPENAI_FILE_CONTENT_BACKEND}",
|
||||
api_key="os.environ/OPENAI_API_KEY",
|
||||
),
|
||||
)
|
||||
model_id = client.create_model(proxy_name, openai_batch_params())
|
||||
resources.defer(lambda: client.delete_model(model_id))
|
||||
key = resources.key()
|
||||
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ Exercises the gateway against a live OpenAI deployment using customer request sh
|
|||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from e2e_config import unique_marker
|
||||
from e2e_config import provider_edge_base, unique_marker
|
||||
from e2e_http import StreamingResponse, assert_client_error, require_successful_call, unwrap
|
||||
from lifecycle import ResourceManager
|
||||
from models import ChatBody, ChatMessage, ChatResponse, LiteLLMParamsBody
|
||||
|
|
@ -38,10 +38,15 @@ class ChatErrorEnvelope(BaseModel):
|
|||
|
||||
|
||||
def _register_chat_model(proxy: ProxyClient, resources: ResourceManager) -> tuple[str, str]:
|
||||
base = provider_edge_base("openai")
|
||||
model = f"e2e-chat-sec-{unique_marker()}"
|
||||
model_id = proxy.create_model(
|
||||
model,
|
||||
LiteLLMParamsBody(model=OPENAI_BACKEND, api_key="os.environ/OPENAI_API_KEY"),
|
||||
LiteLLMParamsBody(
|
||||
model=OPENAI_BACKEND,
|
||||
api_key="os.environ/OPENAI_API_KEY",
|
||||
api_base=None if base is None else f"{base}/v1",
|
||||
),
|
||||
)
|
||||
resources.defer(lambda: proxy.delete_model(model_id))
|
||||
return model, resources.key()
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@ covered by tests/e2e/quota_management/spend_tracking/.
|
|||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from e2e_config import unique_marker
|
||||
from e2e_config import provider_edge_base, unique_marker
|
||||
from e2e_http import (
|
||||
assert_client_error,
|
||||
require_successful_call,
|
||||
|
|
@ -27,6 +27,18 @@ class _OptionalEmbeddingsBody(BaseModel):
|
|||
input: str | list[str] | None = None
|
||||
|
||||
|
||||
def _openai_embeddings_params() -> LiteLLMParamsBody:
|
||||
"""The OpenAI embeddings deployment, wired through the record/replay edge when a
|
||||
fixture mode is active and straight at OpenAI otherwise (LIT-5974). Bedrock and
|
||||
Vertex stay live: SigV4 signs the Host header, and neither has an edge mount."""
|
||||
base = provider_edge_base("openai")
|
||||
return LiteLLMParamsBody(
|
||||
model="openai/text-embedding-3-small",
|
||||
api_key="os.environ/OPENAI_API_KEY",
|
||||
api_base=None if base is None else f"{base}/v1",
|
||||
)
|
||||
|
||||
|
||||
class TestEmbeddingsEndpoint:
|
||||
@pytest.mark.covers("llm.embeddings.openai.basic.nonstream.works")
|
||||
def test_embeddings_returns_vector(
|
||||
|
|
@ -35,9 +47,7 @@ class TestEmbeddingsEndpoint:
|
|||
model = f"e2e-embeddings-{unique_marker()}"
|
||||
model_id = endpoints_client.create_model(
|
||||
model,
|
||||
LiteLLMParamsBody(
|
||||
model="openai/text-embedding-3-small", api_key="os.environ/OPENAI_API_KEY"
|
||||
),
|
||||
_openai_embeddings_params(),
|
||||
)
|
||||
resources.defer(lambda: endpoints_client.delete_model(model_id))
|
||||
key = resources.key()
|
||||
|
|
@ -106,9 +116,7 @@ class TestEmbeddingsEndpoint:
|
|||
model = f"e2e-embeddings-array-{unique_marker()}"
|
||||
model_id = endpoints_client.create_model(
|
||||
model,
|
||||
LiteLLMParamsBody(
|
||||
model="openai/text-embedding-3-small", api_key="os.environ/OPENAI_API_KEY"
|
||||
),
|
||||
_openai_embeddings_params(),
|
||||
)
|
||||
resources.defer(lambda: endpoints_client.delete_model(model_id))
|
||||
key = resources.key()
|
||||
|
|
@ -140,9 +148,7 @@ class TestEmbeddingsEndpoint:
|
|||
model = f"e2e-embeddings-missin-{unique_marker()}"
|
||||
model_id = endpoints_client.create_model(
|
||||
model,
|
||||
LiteLLMParamsBody(
|
||||
model="openai/text-embedding-3-small", api_key="os.environ/OPENAI_API_KEY"
|
||||
),
|
||||
_openai_embeddings_params(),
|
||||
)
|
||||
resources.defer(lambda: endpoints_client.delete_model(model_id))
|
||||
key = resources.key()
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@ litellm-regression-tests/tests/test_inference_endpoints.py.
|
|||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from e2e_config import unique_marker
|
||||
from e2e_config import provider_edge_base, unique_marker
|
||||
from e2e_http import assert_client_error, require_successful_call, unwrap
|
||||
from endpoints_client import EndpointsClient, MessagesResult
|
||||
from lifecycle import ResourceManager
|
||||
|
|
@ -50,16 +50,27 @@ def _approx_equal(actual: float, expected: float) -> bool:
|
|||
return abs(actual - expected) <= max(1e-9, abs(expected) * 1e-2)
|
||||
|
||||
|
||||
def _anthropic_params() -> LiteLLMParamsBody:
|
||||
"""The Anthropic deployment, wired through the record/replay edge when a fixture
|
||||
mode is active (LIT-5974). The mount base carries no ``/v1``: litellm's Anthropic
|
||||
handler appends ``/v1/messages`` to ``api_base`` itself, where the OpenAI handler
|
||||
appends only ``/chat/completions``."""
|
||||
base = provider_edge_base("anthropic")
|
||||
return LiteLLMParamsBody(
|
||||
model=ANTHROPIC_BACKEND, api_key="os.environ/ANTHROPIC_API_KEY", api_base=base
|
||||
)
|
||||
|
||||
|
||||
class TestAnthropicMessages:
|
||||
def _register(
|
||||
self, endpoints_client: EndpointsClient, resources: ResourceManager
|
||||
self,
|
||||
endpoints_client: EndpointsClient,
|
||||
resources: ResourceManager,
|
||||
params: LiteLLMParamsBody | None = None,
|
||||
) -> tuple[str, str]:
|
||||
model = f"e2e-messages-{unique_marker()}"
|
||||
model_id = endpoints_client.create_model(
|
||||
model,
|
||||
LiteLLMParamsBody(
|
||||
model=ANTHROPIC_BACKEND, api_key="os.environ/ANTHROPIC_API_KEY"
|
||||
),
|
||||
model, _anthropic_params() if params is None else params
|
||||
)
|
||||
resources.defer(lambda: endpoints_client.delete_model(model_id))
|
||||
return model, resources.key()
|
||||
|
|
@ -81,12 +92,7 @@ class TestAnthropicMessages:
|
|||
self, endpoints_client: EndpointsClient, resources: ResourceManager
|
||||
) -> None:
|
||||
model = f"e2e-messages-cost-{unique_marker()}"
|
||||
model_id = endpoints_client.create_model(
|
||||
model,
|
||||
LiteLLMParamsBody(
|
||||
model=ANTHROPIC_BACKEND, api_key="os.environ/ANTHROPIC_API_KEY"
|
||||
),
|
||||
)
|
||||
model_id = endpoints_client.create_model(model, _anthropic_params())
|
||||
resources.defer(lambda: endpoints_client.delete_model(model_id))
|
||||
key = resources.key()
|
||||
|
||||
|
|
@ -131,7 +137,13 @@ class TestAnthropicMessages:
|
|||
def test_messages_streams_completion(
|
||||
self, endpoints_client: EndpointsClient, resources: ResourceManager
|
||||
) -> None:
|
||||
model, key = self._register(endpoints_client, resources)
|
||||
"""Stays on a live Anthropic deployment in every mode: the edge buffers a
|
||||
streamed response into one body, so chunk fidelity waits on LIT-5742."""
|
||||
model, key = self._register(
|
||||
endpoints_client,
|
||||
resources,
|
||||
LiteLLMParamsBody(model=ANTHROPIC_BACKEND, api_key="os.environ/ANTHROPIC_API_KEY"),
|
||||
)
|
||||
|
||||
result = endpoints_client.proxy.messages_stream(
|
||||
key,
|
||||
|
|
|
|||
|
|
@ -20,10 +20,9 @@ headers must never touch disk. An unmatched replay call returns HTTP
|
|||
proxy relays as a provider error the failing test surfaces.
|
||||
|
||||
v1 limits: only the mounts in ``EDGE_MOUNTS`` (SigV4 providers like Bedrock
|
||||
sign the Host header, so a forwarding edge breaks their signatures), JSON and
|
||||
opaque single-part bodies (multipart boundaries are random per request),
|
||||
streaming fidelity is LIT-5742, and CI wiring is LIT-5748. Suites that do not
|
||||
wire the edge keep hitting providers live in every mode.
|
||||
sign the Host header, so a forwarding edge breaks their signatures), streaming
|
||||
fidelity is LIT-5742, and CI wiring is LIT-5748. Suites that do not wire the
|
||||
edge keep hitting providers live in every mode.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -32,6 +31,7 @@ import base64
|
|||
import difflib
|
||||
import functools
|
||||
import hashlib
|
||||
import re
|
||||
import threading
|
||||
from collections import deque
|
||||
from collections.abc import Mapping
|
||||
|
|
@ -103,26 +103,194 @@ _RESPONSE_DROPPED_HEADERS: Final[frozenset[str]] = _HOP_BY_HOP_HEADERS | {
|
|||
_JSON: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue)
|
||||
|
||||
|
||||
def _edge_request(method: str, path: str, query: str, body: bytes | None) -> RecordedRequest:
|
||||
"""The identity replay matches on: the edge path (mount included), the query
|
||||
as params, and the body as parsed JSON, or as a canonicalized content digest
|
||||
when it is not JSON so opaque uploads still match across runs."""
|
||||
params: Final = dict(parse_qsl(query, keep_blank_values=True))
|
||||
if not body:
|
||||
return RecordedRequest(method=method.lower(), path=path, headers={}, params=params)
|
||||
decoded: Final = body.decode("utf-8", errors="replace")
|
||||
_BOUNDARY_PATTERN: Final = re.compile(
|
||||
r'boundary=(?:"([^"]*)"|([^;,\s]+))', re.IGNORECASE
|
||||
)
|
||||
_DISPOSITION_NAME_PATTERN: Final = re.compile(r'(?:^|;)\s*name="([^"]*)"')
|
||||
_DISPOSITION_FILENAME_PATTERN: Final = re.compile(r'(?:^|;)\s*filename="([^"]*)"')
|
||||
_UNPARSED_MULTIPART: Final = "<unparsed-multipart>"
|
||||
_BOUNDARY_PLACEHOLDER: Final = b"--<boundary>"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _MultipartPart:
|
||||
field_name: str
|
||||
filename: str | None
|
||||
content: bytes
|
||||
|
||||
|
||||
def _header_value(headers: Mapping[str, str], name: str) -> str:
|
||||
wanted: Final = name.lower()
|
||||
return next((value for key, value in headers.items() if key.lower() == wanted), "")
|
||||
|
||||
|
||||
def _multipart_boundary(content_type: str) -> str | None:
|
||||
if "multipart/form-data" not in content_type.lower():
|
||||
return None
|
||||
match: Final = _BOUNDARY_PATTERN.search(content_type)
|
||||
return None if match is None else match.group(1) or match.group(2)
|
||||
|
||||
|
||||
def _parse_multipart_part(segment: bytes) -> _MultipartPart | None:
|
||||
head, separator, content = segment.partition(b"\r\n\r\n")
|
||||
if not separator:
|
||||
return None
|
||||
disposition: Final = "".join(
|
||||
value
|
||||
for line in head.decode("utf-8", errors="replace").split("\r\n")
|
||||
for name, _, value in [line.partition(":")]
|
||||
if name.strip().lower() == "content-disposition"
|
||||
)
|
||||
name_match: Final = _DISPOSITION_NAME_PATTERN.search(disposition)
|
||||
if name_match is None:
|
||||
return None
|
||||
filename_match: Final = _DISPOSITION_FILENAME_PATTERN.search(disposition)
|
||||
return _MultipartPart(
|
||||
field_name=name_match.group(1),
|
||||
filename=None if filename_match is None else filename_match.group(1),
|
||||
content=content,
|
||||
)
|
||||
|
||||
|
||||
def _multipart_parts(body: bytes, boundary: str) -> tuple[_MultipartPart, ...] | None:
|
||||
"""The wire body split back into its parts, or None when it does not parse as the
|
||||
declared envelope so the caller can fall back to the opaque content digest."""
|
||||
segments: Final = body.split(b"--" + boundary.encode())
|
||||
if len(segments) < 3 or not segments[-1].startswith(b"--"):
|
||||
return None
|
||||
parsed: Final = tuple(
|
||||
_parse_multipart_part(segment.removeprefix(b"\r\n").removesuffix(b"\r\n"))
|
||||
for segment in segments[1:-1]
|
||||
)
|
||||
if any(part is None for part in parsed):
|
||||
return None
|
||||
return tuple(part for part in parsed if part is not None)
|
||||
|
||||
|
||||
def _content_digest(content: bytes) -> str:
|
||||
"""Text is canonicalized before hashing so a per-run marker inside an uploaded JSONL
|
||||
does not move the key; anything that is not UTF-8 is hashed byte for byte, since a
|
||||
lossy decode collapses every binary payload of one length onto one digest."""
|
||||
try:
|
||||
parsed: Final[JsonValue] = _JSON.validate_json(decoded)
|
||||
except ValueError:
|
||||
return RecordedRequest(
|
||||
method=method.lower(),
|
||||
path=path,
|
||||
headers={},
|
||||
params=params,
|
||||
file_sha256=hashlib.sha256(canonical_string(decoded).encode()).hexdigest(),
|
||||
file_bytes=len(body),
|
||||
text: Final = content.decode("utf-8")
|
||||
except UnicodeDecodeError:
|
||||
return hashlib.sha256(content).hexdigest()
|
||||
return hashlib.sha256(canonical_string(text).encode()).hexdigest()
|
||||
|
||||
|
||||
def _form_fields(fields: tuple[_MultipartPart, ...]) -> dict[str, str]:
|
||||
"""The ordinary field parts, flattened into the mapping the bundle format stores. A
|
||||
name sent more than once takes an index instead of overwriting the earlier value, so
|
||||
nothing an upload said is dropped from its key."""
|
||||
form: dict[str, str] = {}
|
||||
for part in fields:
|
||||
name = part.field_name
|
||||
occurrence = 1
|
||||
while name in form:
|
||||
name = f"{part.field_name}[{occurrence}]"
|
||||
occurrence += 1
|
||||
form[name] = part.content.decode("utf-8", errors="replace")
|
||||
return form
|
||||
|
||||
|
||||
def _file_identity(files: tuple[_MultipartPart, ...]) -> tuple[str | None, str | None, int | None]:
|
||||
"""Name, content digest, and total length for the uploaded file parts. The name
|
||||
carries each part's field name as well as its filename, so two uploads sending the
|
||||
same bytes under different field names stay apart. A lone file keeps its own content
|
||||
digest; several fold into one digest over the per-part identities, which is ordered,
|
||||
so parts arriving in a different order key differently."""
|
||||
if not files:
|
||||
return None, None, None
|
||||
names: Final = ", ".join(f"{part.field_name}:{part.filename}" for part in files)
|
||||
total: Final = sum(len(part.content) for part in files)
|
||||
if len(files) == 1:
|
||||
return names, _content_digest(files[0].content), total
|
||||
folded: Final = _JSON.dump_json(
|
||||
[
|
||||
[part.field_name, part.filename, _content_digest(part.content), len(part.content)]
|
||||
for part in files
|
||||
]
|
||||
)
|
||||
return names, hashlib.sha256(folded).hexdigest(), total
|
||||
|
||||
|
||||
def _multipart_request(
|
||||
method: str, path: str, params: dict[str, str], parts: tuple[_MultipartPart, ...]
|
||||
) -> RecordedRequest:
|
||||
"""A multipart upload keyed by what it says rather than by its wire bytes: every
|
||||
ordinary field, plus the identity of the uploaded file. The random per-request
|
||||
boundary is envelope, never content, so it never reaches the digest."""
|
||||
form: Final = _form_fields(tuple(part for part in parts if part.filename is None))
|
||||
file_name, file_sha256, file_bytes = _file_identity(
|
||||
tuple(part for part in parts if part.filename is not None)
|
||||
)
|
||||
return RecordedRequest(
|
||||
method=method,
|
||||
path=path,
|
||||
headers={},
|
||||
params=params,
|
||||
form=form,
|
||||
file_name=file_name,
|
||||
file_sha256=file_sha256,
|
||||
file_bytes=file_bytes,
|
||||
)
|
||||
|
||||
|
||||
def _opaque_request(
|
||||
method: str,
|
||||
path: str,
|
||||
params: dict[str, str],
|
||||
body: bytes,
|
||||
digested: bytes,
|
||||
file_name: str | None = None,
|
||||
) -> RecordedRequest:
|
||||
"""A body kept out of the bundle and matched on its digest alone. ``digested`` is
|
||||
what the digest runs over, which is the body itself unless something in it has to be
|
||||
normalized away first."""
|
||||
return RecordedRequest(
|
||||
method=method,
|
||||
path=path,
|
||||
headers={},
|
||||
params=params,
|
||||
file_name=file_name,
|
||||
file_sha256=_content_digest(digested),
|
||||
file_bytes=len(body),
|
||||
)
|
||||
|
||||
|
||||
def _edge_request(
|
||||
method: str, path: str, query: str, body: bytes | None, content_type: str = ""
|
||||
) -> RecordedRequest:
|
||||
"""The identity replay matches on: the edge path (mount included), the query as
|
||||
params, and the body as parsed JSON, as parsed multipart fields and file identity
|
||||
when the content type declares an envelope, or as a content digest otherwise so
|
||||
opaque uploads still match across runs. A multipart body that does not parse still
|
||||
has its boundary normalized away, because that boundary is fresh every request and
|
||||
would otherwise guarantee a miss."""
|
||||
params: Final = dict(parse_qsl(query, keep_blank_values=True))
|
||||
lowered_method: Final = method.lower()
|
||||
if not body:
|
||||
return RecordedRequest(method=lowered_method, path=path, headers={}, params=params)
|
||||
boundary: Final = _multipart_boundary(content_type)
|
||||
if boundary is not None:
|
||||
parts = _multipart_parts(body, boundary)
|
||||
if parts is not None:
|
||||
return _multipart_request(lowered_method, path, params, parts)
|
||||
return _opaque_request(
|
||||
lowered_method,
|
||||
path,
|
||||
params,
|
||||
body,
|
||||
body.replace(b"--" + boundary.encode(), _BOUNDARY_PLACEHOLDER),
|
||||
_UNPARSED_MULTIPART,
|
||||
)
|
||||
return RecordedRequest(method=method.lower(), path=path, headers={}, params=params, body=parsed)
|
||||
try:
|
||||
parsed: Final[JsonValue] = _JSON.validate_json(body)
|
||||
except ValueError:
|
||||
return _opaque_request(lowered_method, path, params, body, body)
|
||||
return RecordedRequest(
|
||||
method=lowered_method, path=path, headers={}, params=params, body=parsed
|
||||
)
|
||||
|
||||
|
||||
def _build_pool(recorded: tuple[Interaction, ...]) -> dict[str, deque[Interaction]]:
|
||||
|
|
@ -351,7 +519,9 @@ def handle_edge_request(
|
|||
return _text_reply(
|
||||
404, f"unknown provider mount {mount!r}; known mounts: {', '.join(sorted(mounts))}"
|
||||
)
|
||||
request: Final = _edge_request(method, split.path, split.query, body)
|
||||
request: Final = _edge_request(
|
||||
method, split.path, split.query, body, _header_value(headers, "content-type")
|
||||
)
|
||||
match backend:
|
||||
case RecordEdge():
|
||||
return _handle_record(
|
||||
|
|
|
|||
|
|
@ -53,8 +53,10 @@ from provider_edge import (
|
|||
)
|
||||
|
||||
CHAT_PATH = "/openai/v1/chat/completions"
|
||||
UPLOAD_PATH = "/openai/v1/files"
|
||||
REPLAY_MOUNTS = {"openai": "https://replay.invalid"}
|
||||
JSON_OBJECT = TypeAdapter(dict[str, object])
|
||||
BATCH_JSONL = b'{"custom_id":"one"}\n{"custom_id":"two"}\n'
|
||||
|
||||
|
||||
def json_object(body: bytes) -> dict[str, object]:
|
||||
|
|
@ -164,6 +166,46 @@ def chat_body(prompt: str) -> bytes:
|
|||
return json.dumps({"model": "gpt", "messages": [{"role": "user", "content": prompt}]}).encode()
|
||||
|
||||
|
||||
def multipart_body(
|
||||
boundary: str,
|
||||
fields: tuple[tuple[str, str], ...] = (),
|
||||
files: tuple[tuple[str, str, bytes], ...] = (),
|
||||
) -> bytes:
|
||||
"""One multipart/form-data body on the wire, exactly as ``requests`` writes it, with
|
||||
the boundary under the caller's control instead of randomly generated."""
|
||||
parts = [
|
||||
f'--{boundary}\r\nContent-Disposition: form-data; name="{name}"\r\n\r\n'.encode()
|
||||
+ value.encode()
|
||||
for name, value in fields
|
||||
] + [
|
||||
(
|
||||
f'--{boundary}\r\nContent-Disposition: form-data; name="{name}"; '
|
||||
f'filename="{filename}"\r\nContent-Type: application/octet-stream\r\n\r\n'
|
||||
).encode()
|
||||
+ content
|
||||
for name, filename, content in files
|
||||
]
|
||||
return b"\r\n".join(parts) + f"\r\n--{boundary}--\r\n".encode()
|
||||
|
||||
|
||||
def upload_headers(boundary: str) -> dict[str, str]:
|
||||
return {
|
||||
"content-type": f"multipart/form-data; boundary={boundary}",
|
||||
"authorization": "Bearer sk-upload-secret",
|
||||
}
|
||||
|
||||
|
||||
def record_upload(root: Path, body: bytes, boundary: str) -> None:
|
||||
with fake_provider() as provider:
|
||||
with running_edge(record_backend(root), {"openai": provider_url(provider)}) as edge:
|
||||
call_edge(edge, "POST", UPLOAD_PATH, body=body, headers=upload_headers(boundary))
|
||||
|
||||
|
||||
def replay_upload(root: Path, body: bytes, boundary: str) -> RawResponse:
|
||||
with running_edge(ReplayEdge(source=replay_source(root)), REPLAY_MOUNTS) as edge:
|
||||
return call_edge(edge, "POST", UPLOAD_PATH, body=body, headers=upload_headers(boundary))
|
||||
|
||||
|
||||
class TestRecordMode:
|
||||
def test_forwards_to_the_provider_and_writes_one_interaction_file(self, tmp_path: Path) -> None:
|
||||
root = tmp_path / "bundle"
|
||||
|
|
@ -328,6 +370,164 @@ class TestReplayMode:
|
|||
assert replayed.status_code == 200
|
||||
|
||||
|
||||
class TestMultipartIdentity:
|
||||
"""LIT-5974: a multipart upload is keyed by its parsed fields and file identity.
|
||||
``requests`` picks a fresh random boundary per request, so hashing the wire body
|
||||
made every upload miss on replay; parsing the envelope keys the upload on what it
|
||||
actually says, which is stable across runs and still separates real drift."""
|
||||
|
||||
def test_a_fresh_boundary_replays_the_same_upload(self, tmp_path: Path) -> None:
|
||||
root = tmp_path / "bundle"
|
||||
recorded = multipart_body(
|
||||
"d0a1b2c3d4e5f60718293a4b5c6d7e8f",
|
||||
fields=(("purpose", "batch"),),
|
||||
files=(("file", "batch.jsonl", BATCH_JSONL),),
|
||||
)
|
||||
record_upload(root, recorded, "d0a1b2c3d4e5f60718293a4b5c6d7e8f")
|
||||
|
||||
rerun = multipart_body(
|
||||
"ffffeeeeddddccccbbbbaaaa99998888",
|
||||
fields=(("purpose", "batch"),),
|
||||
files=(("file", "batch.jsonl", BATCH_JSONL),),
|
||||
)
|
||||
assert rerun != recorded
|
||||
replayed = replay_upload(root, rerun, "ffffeeeeddddccccbbbbaaaa99998888")
|
||||
assert replayed.status_code == 200, replayed.body[:400]
|
||||
|
||||
def test_the_stored_request_carries_fields_and_file_identity_but_no_secrets(
|
||||
self, tmp_path: Path
|
||||
) -> None:
|
||||
root = tmp_path / "bundle"
|
||||
boundary = "0123456789abcdef0123456789abcdef"
|
||||
record_upload(
|
||||
root,
|
||||
multipart_body(
|
||||
boundary,
|
||||
fields=(("purpose", "batch"),),
|
||||
files=(("file", "batch.jsonl", BATCH_JSONL),),
|
||||
),
|
||||
boundary,
|
||||
)
|
||||
|
||||
raw = this_tests_files(root)[0].read_text(encoding="utf-8")
|
||||
interaction = Interaction.model_validate_json(raw)
|
||||
assert interaction.request.form == {"purpose": "batch"}
|
||||
assert interaction.request.file_name == "file:batch.jsonl"
|
||||
assert interaction.request.file_bytes == len(BATCH_JSONL)
|
||||
stored = interaction.request.model_dump_json()
|
||||
assert boundary not in stored
|
||||
assert "sk-upload-secret" not in stored
|
||||
assert "custom_id" not in stored
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("fields", "files"),
|
||||
[
|
||||
pytest.param(
|
||||
(("purpose", "batch"),),
|
||||
(("file", "batch.jsonl", b'{"custom_id":"three"}\n'),),
|
||||
id="file-content",
|
||||
),
|
||||
pytest.param(
|
||||
(("purpose", "batch"),),
|
||||
(("file", "other.jsonl", BATCH_JSONL),),
|
||||
id="file-name",
|
||||
),
|
||||
pytest.param(
|
||||
(("purpose", "fine-tune"),),
|
||||
(("file", "batch.jsonl", BATCH_JSONL),),
|
||||
id="form-field",
|
||||
),
|
||||
pytest.param(
|
||||
(("purpose", "batch"), ("purpose", "batch")),
|
||||
(("file", "batch.jsonl", BATCH_JSONL),),
|
||||
id="repeated-form-field",
|
||||
),
|
||||
pytest.param(
|
||||
(("purpose", "batch"),),
|
||||
(
|
||||
("file", "batch.jsonl", BATCH_JSONL),
|
||||
("mask", "mask.jsonl", BATCH_JSONL),
|
||||
),
|
||||
id="extra-file-part",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_a_structurally_different_upload_misses(
|
||||
self,
|
||||
tmp_path: Path,
|
||||
fields: tuple[tuple[str, str], ...],
|
||||
files: tuple[tuple[str, str, bytes], ...],
|
||||
) -> None:
|
||||
root = tmp_path / "bundle"
|
||||
record_upload(
|
||||
root,
|
||||
multipart_body(
|
||||
"aaaaaaaabbbbbbbbccccccccdddddddd",
|
||||
fields=(("purpose", "batch"),),
|
||||
files=(("file", "batch.jsonl", BATCH_JSONL),),
|
||||
),
|
||||
"aaaaaaaabbbbbbbbccccccccdddddddd",
|
||||
)
|
||||
|
||||
drifted = replay_upload(
|
||||
root,
|
||||
multipart_body("11112222333344445555666677778888", fields=fields, files=files),
|
||||
"11112222333344445555666677778888",
|
||||
)
|
||||
assert drifted.status_code == REPLAY_MISS_STATUS
|
||||
|
||||
def test_several_file_parts_separate_when_their_contents_swap(self, tmp_path: Path) -> None:
|
||||
root = tmp_path / "bundle"
|
||||
image, mask = b"image-bytes", b"mask-bytes"
|
||||
record_upload(
|
||||
root,
|
||||
multipart_body(
|
||||
"1a1a1a1a2b2b2b2b3c3c3c3c4d4d4d4d",
|
||||
fields=(("prompt", "a cat"),),
|
||||
files=(("image", "a.png", image), ("mask", "b.png", mask)),
|
||||
),
|
||||
"1a1a1a1a2b2b2b2b3c3c3c3c4d4d4d4d",
|
||||
)
|
||||
|
||||
swapped = replay_upload(
|
||||
root,
|
||||
multipart_body(
|
||||
"5e5e5e5e6f6f6f6f7070707081818181",
|
||||
fields=(("prompt", "a cat"),),
|
||||
files=(("image", "a.png", mask), ("mask", "b.png", image)),
|
||||
),
|
||||
"5e5e5e5e6f6f6f6f7070707081818181",
|
||||
)
|
||||
assert swapped.status_code == REPLAY_MISS_STATUS
|
||||
|
||||
same = replay_upload(
|
||||
root,
|
||||
multipart_body(
|
||||
"9292929203030303a4a4a4a4b5b5b5b5",
|
||||
fields=(("prompt", "a cat"),),
|
||||
files=(("image", "a.png", image), ("mask", "b.png", mask)),
|
||||
),
|
||||
"9292929203030303a4a4a4a4b5b5b5b5",
|
||||
)
|
||||
assert same.status_code == 200, same.body[:400]
|
||||
|
||||
def test_a_body_that_does_not_match_its_declared_boundary_stays_opaque(
|
||||
self, tmp_path: Path
|
||||
) -> None:
|
||||
root = tmp_path / "bundle"
|
||||
opaque = b"custom_id one\ncustom_id two\n"
|
||||
absent = "boundary-that-is-absent-from-the-body"
|
||||
record_upload(root, opaque, absent)
|
||||
|
||||
raw = this_tests_files(root)[0].read_text(encoding="utf-8")
|
||||
interaction = Interaction.model_validate_json(raw)
|
||||
assert interaction.request.form is None
|
||||
assert interaction.request.file_name == "<unparsed-multipart>"
|
||||
assert interaction.request.file_bytes == len(opaque)
|
||||
assert "custom_id" not in interaction.request.model_dump_json()
|
||||
assert replay_upload(root, opaque, absent).status_code == 200
|
||||
|
||||
|
||||
class TestReplayLeftover:
|
||||
def test_partially_consumed_recording_names_the_leftover(self, tmp_path: Path) -> None:
|
||||
root = tmp_path / "bundle"
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue