mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-11 22:51:28 +00:00
f5df60f106
Adversarial review of the new multipart keying turned up collisions where two different provider requests computed the same replay key, which is the dangerous failure for a replay harness: the second request silently gets the first one's response instead of missing loudly. - a part counts as an upload when it has a filename or declares its own content type, and the declared content type joins the identity, so two uploads of the same bytes under the same field no longer collapse - the uploaded parts contribute a JSON list of [field, filename, type] triples instead of a "field:filename" string, so a separator inside a filename can no longer impersonate a field boundary - repeated field names get a "name[n]" suffix with a literal "[" doubled first, so a repeated field and a literally indexed one stay distinct - a field value that is not UTF-8 is stored as a base64 sha256 digest; base64 rather than hex because the canonicalizer rewrites 64-character hex runs to <sha256> and folded every binary value onto one key - a field whose name reads as a credential is stored as <secret>. This stays key-preserving because the key is recomputed from the stored request rather than saved beside it, so the live request carrying the real value still matches its redacted fixture - the uploaded byte length leaves the key. The canonicalizer absorbs timestamp and id drift inside a file, and that drift moves the count, so keeping it there made re-records miss Also stops a lookalike parameter such as "xboundary=" from being read as the multipart boundary, and gives the OpenAI batch backend model a single constant instead of three copies of the literal. BUNDLE_FORMAT_VERSION goes to 3 because all of this moves recorded keys. A bundle recorded under the old rules now fails naming both versions instead of missing on every call.
878 lines
36 KiB
Python
878 lines
36 KiB
Python
"""Harness coverage for the provider-edge record/replay server (LIT-5745).
|
|
|
|
No proxy and no ``e2e`` marker. A stdlib http.server stands in for the
|
|
provider (dependency injection via the mounts mapping, no monkeypatching):
|
|
record mode must forward each edge call to it verbatim, persist one
|
|
interaction file, and serve the proxy the same filtered response replay will
|
|
serve later; replay mode must serve byte-identical responses from the bundle
|
|
alone, with the fake provider's hit log proving nothing leaves the process,
|
|
and answer any drifted call with HTTP ``REPLAY_MISS_STATUS`` naming the
|
|
computed and closest recorded canonical keys (LIT-5741; the pure canonicalizer
|
|
is pinned in test_fixture_canonical.py). Requests are made through
|
|
``e2e_http.forward`` so the whole HTTP surface of the edge is exercised; the
|
|
pure ``handle_edge_request`` core is pinned socket-free alongside.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import threading
|
|
from collections.abc import Generator, Mapping
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from contextlib import contextmanager
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from typing import Final
|
|
|
|
import pytest
|
|
from pydantic import TypeAdapter
|
|
|
|
from e2e_http import RawResponse, forward
|
|
from fixture_canonical import canonicalize
|
|
from fixture_bundle import (
|
|
BundleRecorder,
|
|
Interaction,
|
|
LoadedBundle,
|
|
RecordedHttpResponse,
|
|
RecordedRequest,
|
|
load_bundle,
|
|
prepare_bundle,
|
|
slug_for_test,
|
|
)
|
|
from fixture_mode import current_test_key
|
|
from provider_edge import (
|
|
REPLAY_MISS_STATUS,
|
|
EdgeBackend,
|
|
ProviderEdge,
|
|
RecordEdge,
|
|
ReplayEdge,
|
|
ReplaySource,
|
|
edge_request,
|
|
handle_edge_request,
|
|
provider_edge_api_base,
|
|
replay_leftover_error,
|
|
start_provider_edge,
|
|
)
|
|
|
|
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]:
|
|
return JSON_OBJECT.validate_json(body)
|
|
|
|
|
|
class _FakeProvider(ThreadingHTTPServer):
|
|
daemon_threads = True
|
|
|
|
def __init__(self, bind: tuple[str, int]) -> None:
|
|
super().__init__(bind, _FakeProviderHandler)
|
|
self.hits: list[str] = []
|
|
|
|
|
|
class _FakeProviderHandler(BaseHTTPRequestHandler):
|
|
protocol_version = "HTTP/1.1"
|
|
|
|
def do_POST(self) -> None:
|
|
self._respond()
|
|
|
|
def do_GET(self) -> None:
|
|
self._respond()
|
|
|
|
def _respond(self) -> None:
|
|
provider = self.server
|
|
assert isinstance(provider, _FakeProvider)
|
|
length = int(self.headers.get("content-length") or "0")
|
|
body = self.rfile.read(length) if length else b""
|
|
provider.hits.append(f"{self.command} {self.path}")
|
|
payload = json.dumps(
|
|
{"echo": body.decode("utf-8"), "path": self.path, "hit": len(provider.hits)}
|
|
).encode()
|
|
self.send_response(200)
|
|
self.send_header("content-type", "application/json")
|
|
self.send_header("content-length", str(len(payload)))
|
|
self.send_header("x-upstream", "fake")
|
|
self.send_header("set-cookie", "session=fake-cookie")
|
|
self.end_headers()
|
|
self.wfile.write(payload)
|
|
|
|
def log_message(self, format: str, *args: object) -> None:
|
|
"""Silence the per-request stderr line BaseHTTPRequestHandler emits."""
|
|
|
|
|
|
@contextmanager
|
|
def fake_provider() -> Generator[_FakeProvider]:
|
|
server = _FakeProvider(("127.0.0.1", 0))
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
yield server
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
|
|
|
|
def provider_url(server: _FakeProvider) -> str:
|
|
return f"http://127.0.0.1:{server.server_address[1]}"
|
|
|
|
|
|
@contextmanager
|
|
def running_edge(backend: EdgeBackend, mounts: Mapping[str, str]) -> Generator[ProviderEdge]:
|
|
running = start_provider_edge(backend, mounts=mounts, bind_host="127.0.0.1")
|
|
try:
|
|
yield running.edge
|
|
finally:
|
|
running.shutdown()
|
|
|
|
|
|
def record_backend(root: Path) -> RecordEdge:
|
|
recorder = prepare_bundle(root)
|
|
assert isinstance(recorder, BundleRecorder)
|
|
return RecordEdge(recorder=recorder, lock=threading.Lock())
|
|
|
|
|
|
def replay_source(root: Path) -> ReplaySource:
|
|
loaded = load_bundle(root)
|
|
assert isinstance(loaded, LoadedBundle)
|
|
return ReplaySource(bundle=loaded)
|
|
|
|
|
|
def call_edge(
|
|
edge: ProviderEdge,
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
body: bytes | None = None,
|
|
headers: dict[str, str] | None = None,
|
|
) -> RawResponse:
|
|
outcome = forward(
|
|
method,
|
|
f"http://{edge.advertise_host}:{edge.port}{path}",
|
|
headers=headers or {},
|
|
body=body,
|
|
timeout=10.0,
|
|
)
|
|
assert isinstance(outcome, RawResponse)
|
|
return outcome
|
|
|
|
|
|
def this_tests_files(root: Path) -> list[Path]:
|
|
slug_dir = root / slug_for_test(current_test_key())
|
|
return sorted(slug_dir.glob("*.json")) if slug_dir.is_dir() else []
|
|
|
|
|
|
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"
|
|
with fake_provider() as provider:
|
|
with running_edge(record_backend(root), {"openai": provider_url(provider)}) as edge:
|
|
reply = call_edge(edge, "POST", CHAT_PATH, body=chat_body("hi"))
|
|
assert provider.hits == ["POST /v1/chat/completions"]
|
|
assert reply.status_code == 200
|
|
served = json_object(reply.body)
|
|
assert served["echo"] == chat_body("hi").decode()
|
|
files = this_tests_files(root)
|
|
assert [file.name for file in files] == ["0000-post-openai-v1-chat-completions.json"]
|
|
interaction = Interaction.model_validate_json(files[0].read_text(encoding="utf-8"))
|
|
assert interaction.request.method == "post"
|
|
assert interaction.request.path == CHAT_PATH
|
|
assert interaction.request.body == json_object(chat_body("hi"))
|
|
assert interaction.response.status_code == 200
|
|
|
|
def test_never_stores_headers_so_credentials_never_touch_disk(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "bundle"
|
|
with fake_provider() as provider:
|
|
with running_edge(record_backend(root), {"openai": provider_url(provider)}) as edge:
|
|
call_edge(
|
|
edge,
|
|
"POST",
|
|
CHAT_PATH,
|
|
body=chat_body("hi"),
|
|
headers={"authorization": "Bearer sk-live-provider-secret-abc123"},
|
|
)
|
|
raw = this_tests_files(root)[0].read_text(encoding="utf-8")
|
|
assert "sk-live-provider-secret-abc123" not in raw
|
|
interaction = Interaction.model_validate_json(raw)
|
|
assert interaction.request.headers == {}
|
|
|
|
def test_strips_volatile_response_headers_and_serves_the_filtered_copy(self, tmp_path: Path) -> None:
|
|
"""What record serves the proxy must equal what replay will serve later
|
|
(record/replay parity), so the filtered stored copy is served in both."""
|
|
root = tmp_path / "bundle"
|
|
with fake_provider() as provider:
|
|
with running_edge(record_backend(root), {"openai": provider_url(provider)}) as edge:
|
|
reply = call_edge(edge, "POST", CHAT_PATH, body=chat_body("hi"))
|
|
assert reply.headers.get("x-upstream") == "fake"
|
|
assert "set-cookie" not in reply.headers
|
|
interaction = Interaction.model_validate_json(
|
|
this_tests_files(root)[0].read_text(encoding="utf-8")
|
|
)
|
|
assert interaction.response.headers.get("x-upstream") == "fake"
|
|
assert "set-cookie" not in interaction.response.headers
|
|
assert "content-length" not in interaction.response.headers
|
|
|
|
def test_unreachable_provider_records_and_serves_a_502(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "bundle"
|
|
with running_edge(record_backend(root), {"openai": "http://127.0.0.1:9"}) as edge:
|
|
reply = call_edge(edge, "POST", CHAT_PATH, body=chat_body("hi"))
|
|
assert reply.status_code == 502
|
|
assert b"could not reach the provider" in reply.body
|
|
interaction = Interaction.model_validate_json(
|
|
this_tests_files(root)[0].read_text(encoding="utf-8")
|
|
)
|
|
assert interaction.response.status_code == 502
|
|
|
|
|
|
class TestReplayMode:
|
|
def test_serves_recorded_bytes_with_zero_provider_hits(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "bundle"
|
|
with fake_provider() as provider:
|
|
with running_edge(record_backend(root), {"openai": provider_url(provider)}) as edge:
|
|
recorded = call_edge(edge, "POST", CHAT_PATH, body=chat_body("hi"))
|
|
hits_after_record = list(provider.hits)
|
|
with running_edge(
|
|
ReplayEdge(source=replay_source(root)), {"openai": provider_url(provider)}
|
|
) as edge:
|
|
replayed = call_edge(edge, "POST", CHAT_PATH, body=chat_body("hi"))
|
|
assert provider.hits == hits_after_record
|
|
assert replayed.status_code == recorded.status_code
|
|
assert replayed.body == recorded.body
|
|
assert replayed.headers.get("x-upstream") == "fake"
|
|
|
|
def test_request_identity_ignores_auth_headers(self, tmp_path: Path) -> None:
|
|
"""The proxy sends different bearer tokens across runs (fresh virtual
|
|
keys, rotated provider keys), so headers are no part of the match."""
|
|
root = tmp_path / "bundle"
|
|
with fake_provider() as provider:
|
|
with running_edge(record_backend(root), {"openai": provider_url(provider)}) as edge:
|
|
call_edge(
|
|
edge, "POST", CHAT_PATH, body=chat_body("hi"),
|
|
headers={"authorization": "Bearer sk-first-run"},
|
|
)
|
|
with running_edge(ReplayEdge(source=replay_source(root)), REPLAY_MOUNTS) as edge:
|
|
replayed = call_edge(
|
|
edge, "POST", CHAT_PATH, body=chat_body("hi"),
|
|
headers={"authorization": "Bearer sk-second-run"},
|
|
)
|
|
assert replayed.status_code == 200
|
|
|
|
def test_content_drift_returns_the_miss_status_naming_both_keys(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "bundle"
|
|
with fake_provider() as provider:
|
|
with running_edge(record_backend(root), {"openai": provider_url(provider)}) as edge:
|
|
call_edge(edge, "POST", CHAT_PATH, body=chat_body("x"))
|
|
with running_edge(ReplayEdge(source=replay_source(root)), REPLAY_MOUNTS) as edge:
|
|
missed = call_edge(edge, "POST", CHAT_PATH, body=chat_body("y"))
|
|
assert missed.status_code == REPLAY_MISS_STATUS
|
|
message = missed.body.decode()
|
|
assert f"no recorded interaction matches key post {CHAT_PATH} #" in message
|
|
assert f"closest recorded key is post {CHAT_PATH} #" in message
|
|
assert '"content": "x"' in message
|
|
assert '"content": "y"' in message
|
|
assert "re-record with E2E_FIXTURE_MODE=record" in message
|
|
|
|
def test_query_params_are_part_of_the_identity(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "bundle"
|
|
with fake_provider() as provider:
|
|
with running_edge(record_backend(root), {"openai": provider_url(provider)}) as edge:
|
|
call_edge(edge, "GET", "/openai/v1/models?purpose=batch")
|
|
assert provider.hits == ["GET /v1/models?purpose=batch"]
|
|
with running_edge(ReplayEdge(source=replay_source(root)), REPLAY_MOUNTS) as edge:
|
|
missed = call_edge(edge, "GET", "/openai/v1/models?purpose=other")
|
|
matched = call_edge(edge, "GET", "/openai/v1/models?purpose=batch")
|
|
assert missed.status_code == REPLAY_MISS_STATUS
|
|
assert matched.status_code == 200
|
|
|
|
def test_identical_requests_replay_their_responses_in_recorded_order(self, tmp_path: Path) -> None:
|
|
"""A poll or retry loop repeats the same request and the proxy asserts
|
|
on the progression, so duplicates under one key stay FIFO."""
|
|
root = tmp_path / "bundle"
|
|
with fake_provider() as provider:
|
|
with running_edge(record_backend(root), {"openai": provider_url(provider)}) as edge:
|
|
call_edge(edge, "POST", CHAT_PATH, body=chat_body("hi"))
|
|
call_edge(edge, "POST", CHAT_PATH, body=chat_body("hi"))
|
|
with running_edge(ReplayEdge(source=replay_source(root)), REPLAY_MOUNTS) as edge:
|
|
first = json_object(call_edge(edge, "POST", CHAT_PATH, body=chat_body("hi")).body)
|
|
second = json_object(call_edge(edge, "POST", CHAT_PATH, body=chat_body("hi")).body)
|
|
assert first["hit"] == 1
|
|
assert second["hit"] == 2
|
|
|
|
def test_exhausted_key_returns_the_miss_status(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "bundle"
|
|
with fake_provider() as provider:
|
|
with running_edge(record_backend(root), {"openai": provider_url(provider)}) as edge:
|
|
call_edge(edge, "POST", CHAT_PATH, body=chat_body("hi"))
|
|
with running_edge(ReplayEdge(source=replay_source(root)), REPLAY_MOUNTS) as edge:
|
|
call_edge(edge, "POST", CHAT_PATH, body=chat_body("hi"))
|
|
exhausted = call_edge(edge, "POST", CHAT_PATH, body=chat_body("hi"))
|
|
assert exhausted.status_code == REPLAY_MISS_STATUS
|
|
assert b"already consumed" in exhausted.body
|
|
|
|
def test_non_json_bodies_match_by_canonical_digest_without_storing_them(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "bundle"
|
|
opaque = b"custom_id one\ncustom_id two\n"
|
|
with fake_provider() as provider:
|
|
with running_edge(record_backend(root), {"openai": provider_url(provider)}) as edge:
|
|
call_edge(edge, "POST", "/openai/v1/files", body=opaque)
|
|
raw = this_tests_files(root)[0].read_text(encoding="utf-8")
|
|
interaction = Interaction.model_validate_json(raw)
|
|
assert interaction.request.body is None
|
|
assert interaction.request.file_sha256 is not None
|
|
assert interaction.request.file_bytes == len(opaque)
|
|
assert "custom_id" not in interaction.request.model_dump_json()
|
|
with running_edge(ReplayEdge(source=replay_source(root)), REPLAY_MOUNTS) as edge:
|
|
replayed = call_edge(edge, "POST", "/openai/v1/files", body=opaque)
|
|
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 == json.dumps(
|
|
[["file", "batch.jsonl", "application/octet-stream"]], separators=(",", ":")
|
|
)
|
|
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
|
|
|
|
|
|
def raw_multipart(boundary: str, *parts: tuple[str, bytes]) -> bytes:
|
|
"""A body assembled from literal part headers, so a test can send the shapes a
|
|
well-formed helper cannot: a file part with no filename, a declared per-part content
|
|
type, a repeated or bracketed field name, or a non-UTF-8 value."""
|
|
return (
|
|
b"".join(
|
|
f"--{boundary}\r\n{head}\r\n\r\n".encode() + content + b"\r\n"
|
|
for head, content in parts
|
|
)
|
|
+ f"--{boundary}--\r\n".encode()
|
|
)
|
|
|
|
|
|
def upload_key(body: bytes, boundary: str) -> str:
|
|
content_type: Final = f"multipart/form-data; boundary={boundary}"
|
|
return canonicalize(edge_request("POST", UPLOAD_PATH, "", body, content_type)).key
|
|
|
|
|
|
DISPOSITION = 'Content-Disposition: form-data; name="{name}"'
|
|
FILE_DISPOSITION = DISPOSITION + '; filename="{filename}"'
|
|
|
|
|
|
class TestMultipartIdentityEdges:
|
|
"""The identity a multipart upload keys on, pinned against the ways two materially
|
|
different uploads could otherwise collapse onto one key. A collision here is the
|
|
dangerous failure: replay would answer one request with another's response."""
|
|
|
|
def test_a_declared_part_content_type_separates_otherwise_identical_uploads(self) -> None:
|
|
boundary = "0123456789abcdef0123456789abcdef"
|
|
as_json = raw_multipart(
|
|
boundary,
|
|
(FILE_DISPOSITION.format(name="file", filename="a") + "\r\nContent-Type: application/json", b"xy"),
|
|
)
|
|
as_csv = raw_multipart(
|
|
boundary,
|
|
(FILE_DISPOSITION.format(name="file", filename="a") + "\r\nContent-Type: text/csv", b"xy"),
|
|
)
|
|
|
|
assert upload_key(as_json, boundary) != upload_key(as_csv, boundary)
|
|
|
|
def test_a_file_part_without_a_filename_is_not_mistaken_for_a_plain_field(self) -> None:
|
|
boundary = "0123456789abcdef0123456789abcdef"
|
|
upload = raw_multipart(
|
|
boundary,
|
|
(DISPOSITION.format(name="file") + "\r\nContent-Type: application/octet-stream", b"CONTENT"),
|
|
)
|
|
plain_field = raw_multipart(boundary, (DISPOSITION.format(name="file"), b"CONTENT"))
|
|
|
|
request = edge_request(
|
|
"POST", UPLOAD_PATH, "", upload, f"multipart/form-data; boundary={boundary}"
|
|
)
|
|
|
|
assert upload_key(upload, boundary) != upload_key(plain_field, boundary)
|
|
assert request.form == {}
|
|
assert b"CONTENT".decode() not in request.model_dump_json()
|
|
|
|
def test_a_filename_carrying_a_per_run_marker_keys_the_same_next_run(self) -> None:
|
|
boundary = "0123456789abcdef0123456789abcdef"
|
|
|
|
def upload(marker: str) -> str:
|
|
body = raw_multipart(
|
|
boundary,
|
|
(FILE_DISPOSITION.format(name="one", filename=f"{marker}.jsonl"), b"first"),
|
|
(FILE_DISPOSITION.format(name="two", filename="steady.jsonl"), b"second"),
|
|
)
|
|
return upload_key(body, boundary)
|
|
|
|
assert upload("a1b2c3d4e5f6") == upload("0f9e8d7c6b5a")
|
|
|
|
def test_a_separator_inside_a_filename_cannot_forge_a_different_split(self) -> None:
|
|
boundary = "0123456789abcdef0123456789abcdef"
|
|
colon_in_filename = raw_multipart(
|
|
boundary, (FILE_DISPOSITION.format(name="file", filename="a:b.jsonl"), b"same")
|
|
)
|
|
colon_in_field = raw_multipart(
|
|
boundary, (FILE_DISPOSITION.format(name="file:a", filename="b.jsonl"), b"same")
|
|
)
|
|
|
|
assert upload_key(colon_in_filename, boundary) != upload_key(colon_in_field, boundary)
|
|
|
|
def test_a_repeated_field_cannot_collide_with_a_literal_indexed_name(self) -> None:
|
|
boundary = "0123456789abcdef0123456789abcdef"
|
|
repeated = raw_multipart(
|
|
boundary,
|
|
(DISPOSITION.format(name="purpose"), b"x"),
|
|
(DISPOSITION.format(name="purpose"), b"y"),
|
|
)
|
|
literal_index = raw_multipart(
|
|
boundary,
|
|
(DISPOSITION.format(name="purpose"), b"x"),
|
|
(DISPOSITION.format(name="purpose[1]"), b"y"),
|
|
)
|
|
|
|
assert upload_key(repeated, boundary) != upload_key(literal_index, boundary)
|
|
|
|
def test_two_binary_field_values_of_one_length_stay_apart(self) -> None:
|
|
boundary = "0123456789abcdef0123456789abcdef"
|
|
first = raw_multipart(boundary, (DISPOSITION.format(name="blob"), b"\xff\xfe\xfd"))
|
|
second = raw_multipart(boundary, (DISPOSITION.format(name="blob"), b"\xf0\xf1\xf2"))
|
|
|
|
assert upload_key(first, boundary) != upload_key(second, boundary)
|
|
|
|
def test_a_secret_named_field_never_reaches_the_stored_request(self) -> None:
|
|
boundary = "0123456789abcdef0123456789abcdef"
|
|
body = raw_multipart(
|
|
boundary,
|
|
(DISPOSITION.format(name="openai_api_key"), b"sk-live-DEADBEEF-0123456789abcd"),
|
|
(DISPOSITION.format(name="purpose"), b"batch"),
|
|
)
|
|
|
|
request = edge_request(
|
|
"POST", UPLOAD_PATH, "", body, f"multipart/form-data; boundary={boundary}"
|
|
)
|
|
|
|
assert "sk-live-DEADBEEF-0123456789abcd" not in request.model_dump_json()
|
|
assert request.form == {"openai_api_key": "<secret>", "purpose": "batch"}
|
|
|
|
def test_a_redacted_field_still_matches_the_live_request_that_carried_the_secret(
|
|
self,
|
|
) -> None:
|
|
boundary = "0123456789abcdef0123456789abcdef"
|
|
|
|
def upload(secret: str) -> str:
|
|
body = raw_multipart(
|
|
boundary,
|
|
(DISPOSITION.format(name="openai_api_key"), secret.encode()),
|
|
(DISPOSITION.format(name="purpose"), b"batch"),
|
|
)
|
|
return upload_key(body, boundary)
|
|
|
|
assert upload("sk-live-DEADBEEF-0123456789abcd") == upload("<secret>")
|
|
|
|
def test_a_length_change_the_canonicalizer_absorbs_does_not_move_the_key(self) -> None:
|
|
boundary = "0123456789abcdef0123456789abcdef"
|
|
|
|
def upload(created: str) -> str:
|
|
body = raw_multipart(
|
|
boundary,
|
|
(
|
|
FILE_DISPOSITION.format(name="file", filename="batch.jsonl"),
|
|
b'{"created_at":"' + created.encode() + b'"}',
|
|
),
|
|
)
|
|
return upload_key(body, boundary)
|
|
|
|
assert upload("2026-08-21T02:08:19Z") == upload("2026-08-21T02:08:19.123456Z")
|
|
|
|
@pytest.mark.parametrize(
|
|
"content_type",
|
|
[
|
|
pytest.param("multipart/form-data; myboundary=zzz; boundary={boundary}", id="lookalike-parameter"),
|
|
pytest.param("multipart/form-data; BOUNDARY={boundary}", id="uppercase-parameter"),
|
|
],
|
|
)
|
|
def test_the_boundary_parameter_is_read_the_way_the_client_meant_it(
|
|
self, content_type: str
|
|
) -> None:
|
|
boundary = "0123456789abcdef0123456789abcdef"
|
|
body = raw_multipart(
|
|
boundary, (FILE_DISPOSITION.format(name="file", filename="batch.jsonl"), BATCH_JSONL)
|
|
)
|
|
|
|
request = edge_request(
|
|
"POST", UPLOAD_PATH, "", body, content_type.format(boundary=boundary)
|
|
)
|
|
|
|
assert request.form == {}
|
|
assert request.file_name is not None
|
|
assert "batch.jsonl" in request.file_name
|
|
|
|
def test_an_empty_declared_boundary_falls_back_instead_of_splitting_on_dashes(self) -> None:
|
|
body = b'--\r\nContent-Disposition: form-data; name="a"\r\n\r\nvalue\r\n----\r\n'
|
|
|
|
request = edge_request(
|
|
"POST", UPLOAD_PATH, "", body, 'multipart/form-data; boundary=""'
|
|
)
|
|
|
|
assert request.form is None
|
|
assert request.file_sha256 is not None
|
|
|
|
|
|
class TestReplayLeftover:
|
|
def test_partially_consumed_recording_names_the_leftover(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "bundle"
|
|
with fake_provider() as provider:
|
|
with running_edge(record_backend(root), {"openai": provider_url(provider)}) as edge:
|
|
call_edge(edge, "POST", CHAT_PATH, body=chat_body("hi"))
|
|
call_edge(edge, "GET", "/openai/v1/models")
|
|
source = replay_source(root)
|
|
with running_edge(ReplayEdge(source=source), REPLAY_MOUNTS) as edge:
|
|
call_edge(edge, "POST", CHAT_PATH, body=chat_body("hi"))
|
|
error = source.leftover_error(current_test_key())
|
|
assert error is not None
|
|
assert "1 of 2 recorded interactions never consumed" in error
|
|
assert "e.g. get /openai/v1/models #" in error
|
|
assert "re-record with E2E_FIXTURE_MODE=record" in error
|
|
|
|
def test_fully_consumed_recording_leaves_nothing(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "bundle"
|
|
with fake_provider() as provider:
|
|
with running_edge(record_backend(root), {"openai": provider_url(provider)}) as edge:
|
|
call_edge(edge, "POST", CHAT_PATH, body=chat_body("hi"))
|
|
source = replay_source(root)
|
|
with running_edge(ReplayEdge(source=source), REPLAY_MOUNTS) as edge:
|
|
call_edge(edge, "POST", CHAT_PATH, body=chat_body("hi"))
|
|
assert source.leftover_error(current_test_key()) is None
|
|
|
|
def test_test_without_recordings_has_no_leftover(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "bundle"
|
|
assert isinstance(prepare_bundle(root), BundleRecorder)
|
|
assert replay_source(root).leftover_error("suite.py::test_never_recorded") is None
|
|
|
|
def test_inert_outside_replay_mode(self, tmp_path: Path) -> None:
|
|
missing = tmp_path / "missing"
|
|
assert replay_leftover_error(mode_raw="", bundle_dir=missing, test_key="k") is None
|
|
assert replay_leftover_error(mode_raw="record", bundle_dir=missing, test_key="k") is None
|
|
|
|
|
|
class TestConcurrentReplay:
|
|
def test_parallel_identical_calls_serve_each_recording_exactly_once(self, tmp_path: Path) -> None:
|
|
"""The edge server handles requests on concurrent threads and a burst
|
|
of parallel identical calls consumes one shared pool: no response
|
|
duplicated, none forgotten, nothing left over at teardown."""
|
|
root = tmp_path / "bundle"
|
|
recorder = prepare_bundle(root)
|
|
assert isinstance(recorder, BundleRecorder)
|
|
for ordinal in range(32):
|
|
recorder.record(
|
|
test_key=current_test_key(),
|
|
request=RecordedRequest(method="post", path=CHAT_PATH, headers={}, body={"n": "same"}),
|
|
response=RecordedHttpResponse(
|
|
status_code=200,
|
|
headers={"content-type": "application/json"},
|
|
body_b64=base64.b64encode(json.dumps({"value": f"v{ordinal:02d}"}).encode()).decode(),
|
|
),
|
|
)
|
|
source = replay_source(root)
|
|
body = json.dumps({"n": "same"}).encode()
|
|
barrier = threading.Barrier(8)
|
|
with running_edge(ReplayEdge(source=source), REPLAY_MOUNTS) as edge:
|
|
|
|
def consume(_: int) -> tuple[str, ...]:
|
|
barrier.wait()
|
|
return tuple(
|
|
str(json_object(call_edge(edge, "POST", CHAT_PATH, body=body).body)["value"])
|
|
for _call in range(4)
|
|
)
|
|
|
|
with ThreadPoolExecutor(max_workers=8) as executor:
|
|
served = sorted(value for values in executor.map(consume, range(8)) for value in values)
|
|
assert served == [f"v{ordinal:02d}" for ordinal in range(32)]
|
|
assert source.leftover_error(current_test_key()) is None
|
|
|
|
|
|
class TestHandleEdgeRequestPure:
|
|
def test_unknown_mount_404s_naming_the_known_mounts(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "bundle"
|
|
assert isinstance(prepare_bundle(root), BundleRecorder)
|
|
reply = handle_edge_request(
|
|
ReplayEdge(source=replay_source(root)),
|
|
{"openai": "https://api.openai.com", "anthropic": "https://api.anthropic.com"},
|
|
"POST",
|
|
"/bedrock/model/invoke",
|
|
{},
|
|
b"{}",
|
|
timeout=1.0,
|
|
)
|
|
assert reply.status_code == 404
|
|
assert b"unknown provider mount 'bedrock'" in reply.body
|
|
assert b"anthropic, openai" in reply.body
|
|
|
|
def test_replay_serves_a_directly_recorded_interaction(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "bundle"
|
|
recorder = prepare_bundle(root)
|
|
assert isinstance(recorder, BundleRecorder)
|
|
recorder.record(
|
|
test_key=current_test_key(),
|
|
request=RecordedRequest(method="post", path=CHAT_PATH, headers={}, body={"prompt": "x"}),
|
|
response=RecordedHttpResponse(
|
|
status_code=201, headers={"x-upstream": "fake"}, body_b64=base64.b64encode(b"ok").decode()
|
|
),
|
|
)
|
|
reply = handle_edge_request(
|
|
ReplayEdge(source=replay_source(root)),
|
|
{"openai": "https://api.openai.com"},
|
|
"POST",
|
|
CHAT_PATH,
|
|
{"authorization": "Bearer sk-anything"},
|
|
json.dumps({"prompt": "x"}).encode(),
|
|
timeout=1.0,
|
|
)
|
|
assert reply.status_code == 201
|
|
assert reply.body == b"ok"
|
|
assert reply.headers == {"x-upstream": "fake"}
|
|
|
|
|
|
class TestApiBaseSeam:
|
|
def test_live_mode_returns_none(self, tmp_path: Path) -> None:
|
|
for mode_raw in ("live", ""):
|
|
assert (
|
|
provider_edge_api_base(
|
|
"openai",
|
|
mode_raw=mode_raw,
|
|
bundle_dir=tmp_path / "bundle",
|
|
bind_host="127.0.0.1",
|
|
advertise_host="127.0.0.1",
|
|
)
|
|
is None
|
|
)
|
|
|
|
def test_invalid_mode_raises_naming_the_value(self, tmp_path: Path) -> None:
|
|
with pytest.raises(ValueError, match="cached"):
|
|
provider_edge_api_base(
|
|
"openai",
|
|
mode_raw="cached",
|
|
bundle_dir=tmp_path / "bundle",
|
|
bind_host="127.0.0.1",
|
|
advertise_host="127.0.0.1",
|
|
)
|
|
|
|
def test_unknown_mount_raises_naming_the_known_mounts(self, tmp_path: Path) -> None:
|
|
with pytest.raises(ValueError, match="unknown provider mount 'bedrock'"):
|
|
provider_edge_api_base(
|
|
"bedrock",
|
|
mode_raw="record",
|
|
bundle_dir=tmp_path / "bundle",
|
|
bind_host="127.0.0.1",
|
|
advertise_host="127.0.0.1",
|
|
)
|
|
|
|
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"
|
|
)
|
|
second = provider_edge_api_base(
|
|
"anthropic", mode_raw="record", bundle_dir=root, bind_host="127.0.0.1", advertise_host="127.0.0.1"
|
|
)
|
|
assert first is not None and second is not None
|
|
assert first.endswith("/openai")
|
|
assert second.endswith("/anthropic")
|
|
assert first.rsplit("/", 1)[0] == second.rsplit("/", 1)[0]
|
|
assert (root / "manifest.json").is_file()
|