scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

Merge pull request #37565 from BerriAI/litellm_lit_5745_provider_edge_replay

Browse source 
feat(e2e): move record/replay to the provider edge (LIT-5745)
This commit is contained in:
Mateo Wang 2026-08-20 13:50:04 -07:00 • committed by GitHub
commit fc3b160fb5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
16 changed files with 1461 additions and 1581 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
tests/e2e/CLAUDE.md
View file

@ -73,13 +73,17 @@ Mark live tests with `@pytest.mark.e2e` (on the class or the module). Pure cover
## Record and replay fixtures
`E2E_FIXTURE_MODE` selects the transport every client is built on: `live` (the default, and what an unset variable means: nothing changes), `record` (run against the live proxy and write every interaction to a fixture bundle), or `replay` (serve every interaction back from the bundle with no HTTP at all, so a replay run needs no proxy and cannot bill a provider). The seam is `select_transport` in `fixture_transport.py`, applied inside `build_proxy_client`; both transports fulfil the same `Transport` protocol, so no test or client changes shape in any mode
`E2E_FIXTURE_MODE` scopes the proxy's provider-bound traffic: `live` (the default, and what an unset variable means: nothing changes), `record` (the proxy's provider calls are forwarded to the real provider through a local edge server and written to a fixture bundle), or `replay` (the edge answers those calls from the bundle, so the run makes zero provider calls and spends nothing). Test-to-proxy traffic always goes over the wire in every mode: record and replay both need the live proxy and database, because the point is that key auth, routing, cost calculation, and spend-log writes execute for real while only the provider is swapped out. Breaking any of those in the proxy turns a replay run red
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 transport call in call order (`0000-post-chat-completions.json`). Auth header values and credential request fields (`api_key`, `*_secret_key`, `static_headers`, and the like; the list is `fixture_canonical.py`'s) are redacted on write, and file uploads store a sha256 digest instead of the bytes; response bodies are stored verbatim (a /key/generate response keeps the ephemeral virtual key it minted), which is part of why bundles are gitignored. `fixture_bundle.py` owns the format
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
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, path, and a content hash, so identity survives re-records and machine changes while any real content drift is a `ReplayMiss` that names 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 poll 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 proxy
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
Deliberately not here yet: streaming chunk fidelity (LIT-5742) and scoping record/replay to provider-bound traffic (LIT-5745)
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)
## Typing

10
tests/e2e/CONTRIBUTING.md
View file

@ -54,14 +54,16 @@ Some suites need extra services the bare proxy does not start. The `logging/` OT
### Record and replay
`E2E_FIXTURE_MODE=record` runs a suite against the live proxy as usual while writing every request/response pair to a fixture bundle (default `tests/e2e/.fixtures`, override with `E2E_FIXTURE_DIR`); `E2E_FIXTURE_MODE=replay` then runs the same suite entirely from that bundle, with no proxy traffic and no provider spend; the proxy liveness gate is skipped, so replay runs with no proxy up at all. Unset (or `live`) behaves exactly as before the knob existed
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/llm_translation/ -v
E2E_FIXTURE_MODE=replay uv run pytest tests/e2e/llm_translation/ -v
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
```
Replay fails hard (`ReplayMiss`) when the tests drift from the recording, and a bundle older than seven days fails at collection time naming its age; either way the fix is to re-record. See `CLAUDE.md` in this directory for the bundle format and the transport seam
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)
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

14
tests/e2e/conftest.py
View file

@ -23,12 +23,8 @@ import requests
from e2e_config import CONTROL_PLANE_BASE_URL, FIXTURE_DIR, FIXTURE_MODE_RAW, PROXY_BASE_URL
from e2e_db import RESET_OPT_IN_ENV, reset_spend_logs, run_spend_log_cleanup
from fixture_transport import (
fixture_mode_collection_error,
fixture_report_lines,
parse_fixture_mode,
replay_leftover_error,
)
from fixture_mode import fixture_mode_collection_error, fixture_report_lines
from provider_edge import replay_leftover_error
from junit_properties import attach_result_properties
from lifecycle import ProxyClientProvider, ResourceManager
from proxy_client import ProxyClient, build_proxy_client
@ -114,12 +110,10 @@ def _proxy_fail_reason() -> str | None:
def pytest_runtest_setup(item: pytest.Item) -> None:
"""Hard-fail `e2e`-marked tests unless a proxy answers its liveness probe.
Unmarked tests (unit coverage of the harness) don't touch the proxy, so they
run even when none is up. Never skip for a missing proxy. Replay mode serves
every call from the fixture bundle, so it needs no live proxy either."""
run even when none is up. Never skip for a missing proxy. Replay mode needs
the proxy too: only provider-bound traffic replays from the bundle."""
if item.get_closest_marker("e2e") is None:
return
if parse_fixture_mode(FIXTURE_MODE_RAW) == "replay":
return
reason = _proxy_fail_reason()
if reason is not None:
pytest.fail(reason)

32
tests/e2e/e2e_config.py
View file

@ -13,7 +13,8 @@ from pathlib import Path
from dotenv import load_dotenv
from fixture_transport import deterministic_marker, parse_fixture_mode
from fixture_mode import 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).
# Compose injects them into the proxy container, but pytest on the host does not
@ -92,15 +93,24 @@ PROPAGATION_TIMEOUT = float(os.environ.get("E2E_PROPAGATION_TIMEOUT", "15"))
EXPECT_RUST = os.environ.get("E2E_EXPECT_RUST", "").strip().lower() in ("1", "true", "yes")
# Record/replay fixture selection (see fixture_transport.py). The raw mode value
# is parsed and validated there; "live" (the default, also for empty values)
# means the harness behaves exactly as before this knob existed.
# Record/replay fixture selection (see fixture_mode.py and provider_edge.py).
# The raw mode value is parsed and validated there; "live" (the default, also
# for empty values) means the harness behaves exactly as before this knob
# existed.
FIXTURE_MODE_RAW = os.environ.get("E2E_FIXTURE_MODE", "live")
FIXTURE_DIR = Path(
os.environ.get("E2E_FIXTURE_DIR", "").strip()
or str(Path(__file__).resolve().parent / ".fixtures")
)
# Where the provider-edge server binds, and the host name edge api_base URLs
# advertise to the proxy. They differ when the proxy runs in a container and
# reaches the pytest host via a gateway name like host.docker.internal.
PROVIDER_EDGE_BIND_HOST = os.environ.get("E2E_PROVIDER_EDGE_BIND_HOST", "").strip() or "127.0.0.1"
PROVIDER_EDGE_ADVERTISE_HOST = (
os.environ.get("E2E_PROVIDER_EDGE_ADVERTISE_HOST", "").strip() or PROVIDER_EDGE_BIND_HOST
)
# Deliberately modest concurrency. The suite shares its proxy with every other
# suite in the run, and 750 users at spawn rate 50 saturated the request path hard
# enough to distort latency-sensitive neighbours (and to spend real provider money
@ -157,6 +167,20 @@ def datadog_mcp_url(*, toolsets: str = "core") -> str:
return f"{base}?toolsets={toolsets}" if toolsets else base
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."""
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,
forward_timeout=REQUEST_TIMEOUT,
)
def unique_marker() -> str:
"""A short unique token per call/run, so concurrent runs and the shared
response cache never collide on prompts, tags, or customer ids. In record

34
tests/e2e/e2e_http.py
View file

@ -647,3 +647,37 @@ def download(
content_type=_hdr(resp, "content-type"),
body=resp.text,
)
class RawResponse(BaseModel):
"""A verbatim upstream HTTP response for the provider edge (provider_edge.py):
status, lowercased headers, raw bytes. No Result classification because the
edge relays provider errors to the proxy untouched."""
status_code: int
headers: dict[str, str]
body: bytes
def forward(
method: str,
url: str,
*,
headers: dict[str, str],
body: bytes | None,
timeout: float = 60.0,
) -> RawResponse | NetworkError:
"""Relay one provider-bound request verbatim for the provider edge's record
mode. No retries, no redirects, no schema: the proxy owns retry policy and
the recorded bundle must hold exactly what the provider returned."""
try:
resp = requests.request(
method, url, headers=headers, data=body, timeout=timeout, allow_redirects=False
)
except requests.RequestException as exc:
return NetworkError(message=str(exc))
return RawResponse(
status_code=resp.status_code,
headers={name.lower(): value for name, value in resp.headers.items()},
body=resp.content,
)

133
tests/e2e/fixture_bundle.py
View file

@ -1,17 +1,18 @@
"""On-disk fixture bundle format for record/replay e2e runs (LIT-5729).
"""On-disk fixture bundle format for record/replay e2e runs (LIT-5729/LIT-5745).
A bundle is a directory: one ``manifest.json`` (record timestamp + harness
version + format version) plus one subdirectory per test, holding one JSON file
per transport interaction in call order. Bundles older than
per provider-bound interaction in call order. Bundles older than
``MAX_BUNDLE_AGE`` hard-fail replay at collection time (see conftest), so a
green replay run can never certify against fixtures that have drifted more than
a week from the live proxy.
a week from the live providers.
This module owns the format only. The transports that produce and consume it
live in fixture_transport.py and the canonical match keys they compute live in
fixture_canonical.py (LIT-5741); streaming chunk fidelity and provider-scoping
are follow-ups (LIT-5742/5745). Every interaction file stores the full redacted
request because replay matches on its canonicalized content.
This module owns the format only. The provider-edge server that produces and
consumes it lives in provider_edge.py (LIT-5745) and the canonical match keys
it computes live in fixture_canonical.py (LIT-5741); streaming chunk fidelity
is a follow-up (LIT-5742). Every interaction file stores the full redacted
request because replay matches on its canonicalized content, and the response
as the raw HTTP status, filtered headers, and base64 body the provider sent.
"""
from __future__ import annotations
@ -23,29 +24,14 @@ import subprocess
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Annotated, Final, Literal
from typing import Final
from pydantic import BaseModel, Field, JsonValue, TypeAdapter
from pydantic import BaseModel, JsonValue
from e2e_http import (
BinaryStream,
NetworkError,
ProbeResult,
RateLimitedError,
Result,
StreamingResponse,
Success,
UnauthorizedError,
UnknownApiError,
ValidationError,
)
BUNDLE_FORMAT_VERSION: Final = 1
BUNDLE_FORMAT_VERSION: Final = 2
MAX_BUNDLE_AGE: Final = timedelta(days=7)
MANIFEST_FILENAME: Final = "manifest.json"
_JSON: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue)
class Manifest(BaseModel):
format_version: int
@ -54,13 +40,14 @@ class Manifest(BaseModel):
class RecordedRequest(BaseModel):
"""The request as the transport saw it, auth header values and credential
body/form fields redacted.
"""The provider-bound request as the edge saw it, headers empty (SDK
telemetry headers vary run to run and auth material never touches disk).
Replay matches on the canonical content key fixture_canonical.py computes
over ``method`` (the transport verb, not the HTTP verb), ``path``, and the
canonicalized headers, params, body, form, and file identity. File uploads
store a content digest instead of the bytes."""
over ``method``, ``path`` (the edge path including the provider mount,
query string excluded), and the canonicalized headers, params, body, form,
and file identity. Non-JSON bodies store a canonicalized content digest
instead of the bytes."""
method: str
path: str
@ -73,85 +60,19 @@ class RecordedRequest(BaseModel):
file_bytes: int | None = None
class RecordedResult(BaseModel):
"""A ``Result[R]`` flattened for disk. ``data`` holds the success payload as
raw JSON; replay re-validates it against the ``response_type`` the caller
passes, exactly like a live response body."""
class RecordedHttpResponse(BaseModel):
"""The provider's raw HTTP response: status, headers minus hop-by-hop and
volatile entries (see provider_edge.py), and the body as base64 so binary
payloads survive JSON."""
shape: Literal["result"] = "result"
kind: Literal["success", "network", "unauthorized", "rate_limited", "validation", "unknown"]
status_code: int | None = None
data: JsonValue | None = None
message: str | None = None
body: str | None = None
retry_after_seconds: int | None = None
class RecordedStreaming(BaseModel):
shape: Literal["streaming"] = "streaming"
payload: StreamingResponse
class RecordedBinary(BaseModel):
shape: Literal["binary"] = "binary"
payload: BinaryStream
class RecordedProbe(BaseModel):
shape: Literal["probe"] = "probe"
payload: ProbeResult
type RecordedResponse = RecordedResult | RecordedStreaming | RecordedBinary | RecordedProbe
status_code: int
headers: dict[str, str]
body_b64: str
class Interaction(BaseModel):
request: RecordedRequest
response: Annotated[
RecordedResult | RecordedStreaming | RecordedBinary | RecordedProbe,
Field(discriminator="shape"),
]
def to_json_value(model: BaseModel) -> JsonValue:
return _JSON.validate_json(model.model_dump_json(by_alias=True))
def from_result[R: BaseModel](result: Result[R]) -> RecordedResult:
match result:
case Success(status_code=status_code, data=data):
return RecordedResult(kind="success", status_code=status_code, data=to_json_value(data))
case NetworkError(message=message):
return RecordedResult(kind="network", message=message)
case UnauthorizedError():
return RecordedResult(kind="unauthorized")
case RateLimitedError(retry_after_seconds=retry_after_seconds, body=body):
return RecordedResult(kind="rate_limited", retry_after_seconds=retry_after_seconds, body=body)
case ValidationError(message=message):
return RecordedResult(kind="validation", message=message)
case UnknownApiError(status_code=status_code, body=body):
return RecordedResult(kind="unknown", status_code=status_code, body=body)
def to_result[R: BaseModel](recorded: RecordedResult, response_type: type[R]) -> Result[R]:
match recorded.kind:
case "success":
return Success(
status_code=recorded.status_code or 200,
data=response_type.model_validate(recorded.data),
)
case "network":
return NetworkError(message=recorded.message or "")
case "unauthorized":
return UnauthorizedError()
case "rate_limited":
return RateLimitedError(
retry_after_seconds=recorded.retry_after_seconds, body=recorded.body or ""
)
case "validation":
return ValidationError(message=recorded.message or "")
case "unknown":
return UnknownApiError(status_code=recorded.status_code or 0, body=recorded.body or "")
response: RecordedHttpResponse
def slugify(raw: str, *, limit: int = 60) -> str:
@ -198,7 +119,7 @@ class BundleRecorder:
root: Path
_ordinals: dict[str, int] = field(default_factory=dict)
def record(self, *, test_key: str, request: RecordedRequest, response: RecordedResponse) -> None:
def record(self, *, test_key: str, request: RecordedRequest, response: RecordedHttpResponse) -> None:
slug = slug_for_test(test_key)
ordinal = self._ordinals.get(slug, 0)
self._ordinals[slug] = ordinal + 1

132
tests/e2e/fixture_mode.py Normal file
View file

@ -0,0 +1,132 @@
"""Fixture-mode selection and per-test determinism for record/replay e2e runs.
``E2E_FIXTURE_MODE`` is live (the default; nothing changes), record, or replay.
This module owns everything mode-shaped that is independent of the provider
edge itself: parsing the raw env value, the collection-time gate that aborts a
run whose mode can never work (unknown value, or replay against a missing or
stale bundle), the pytest report-header lines, the running test's node id, and
the deterministic per-test marker that lets a replay run regenerate exactly
the requests the record run sent. The provider-edge server that records and
serves provider traffic lives in provider_edge.py (LIT-5745).
"""
from __future__ import annotations
import hashlib
import os
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Final, Literal, assert_never
from fixture_bundle import (
FreshBundle,
StaleBundle,
UnreadableBundle,
check_freshness,
format_age,
)
type FixtureMode = Literal["live", "record", "replay"]
FIXTURE_MODES: Final[tuple[FixtureMode, ...]] = ("live", "record", "replay")
SESSION_TEST_KEY: Final = "session"
@dataclass(frozen=True, slots=True)
class InvalidFixtureMode:
value: str
def parse_fixture_mode(raw: str) -> FixtureMode | InvalidFixtureMode:
normalized = raw.strip().lower() or "live"
match normalized:
case "live" | "record" | "replay":
return normalized
case _:
return InvalidFixtureMode(value=raw)
def current_test_key() -> str:
"""The pytest node id of the running test, from the PYTEST_CURRENT_TEST env
var pytest maintains (``<nodeid> (setup|call|teardown)``); ``session`` for
calls outside any test (e.g. session-finish cleanup)."""
raw = os.environ.get("PYTEST_CURRENT_TEST", "")
if not raw:
return SESSION_TEST_KEY
return raw.rsplit(" (", 1)[0]
class ReplayMiss(AssertionError):
"""Replay had no recorded interaction for a provider call the proxy made.
The suite drifted from the bundle (or the bundle from the suite): re-record."""
_marker_ordinals: Final[dict[str, int]] = {}
def deterministic_marker() -> str:
"""Stable stand-in for uuid-based unique markers in record and replay modes:
the Nth marker of a test is a pure function of the test's node id and N, so a
replay run regenerates exactly the model names, prompts, and tags the record
run sent and every recorded provider interaction still matches its key."""
test_key = current_test_key()
ordinal = _marker_ordinals.get(test_key, 0)
_marker_ordinals[test_key] = ordinal + 1
return hashlib.sha1(f"{test_key}#{ordinal}".encode()).hexdigest()[:12]
def fixture_mode_collection_error(mode_raw: str, bundle_dir: Path, *, now: datetime) -> str | None:
"""Session-abort reason for a fixture-mode setup that can never work, or None.
Called at collection time (conftest pytest_sessionstart) so a stale or missing
bundle fails the whole run up front, naming the bundle age, instead of failing
every test individually."""
mode = parse_fixture_mode(mode_raw)
match mode:
case InvalidFixtureMode(value=value):
return f"E2E_FIXTURE_MODE={value!r} is not one of {', '.join(FIXTURE_MODES)}"
case "live" | "record":
return None
case "replay":
freshness = check_freshness(bundle_dir, now=now)
match freshness:
case FreshBundle():
return None
case StaleBundle(recorded_at=recorded_at, age=age, limit=limit):
return (
f"fixture bundle at {bundle_dir} is stale: recorded {recorded_at.isoformat()}, "
f"age {format_age(age)} exceeds the {limit.days}-day limit; "
"re-record with E2E_FIXTURE_MODE=record"
)
case UnreadableBundle(reason=reason):
return f"E2E_FIXTURE_MODE=replay cannot use bundle at {bundle_dir}: {reason}"
case _:
assert_never(freshness)
case _:
assert_never(mode)
def fixture_report_lines(mode_raw: str, bundle_dir: Path, *, now: datetime) -> list[str]:
"""pytest report-header lines; empty in live mode so an unset
E2E_FIXTURE_MODE keeps today's output byte-identical."""
mode = parse_fixture_mode(mode_raw)
match mode:
case InvalidFixtureMode() | "live":
return []
case "record":
return [f"e2e fixture mode: record -> {bundle_dir}"]
case "replay":
freshness = check_freshness(bundle_dir, now=now)
match freshness:
case FreshBundle(manifest=manifest):
return [
f"e2e fixture mode: replay <- {bundle_dir} "
f"(recorded {manifest.recorded_at.isoformat()}, harness {manifest.harness_version})"
]
case StaleBundle() | UnreadableBundle():
return [f"e2e fixture mode: replay <- {bundle_dir}"]
case _:
assert_never(freshness)
case _:
assert_never(mode)

724
tests/e2e/fixture_transport.py
View file

@ -1,724 +0,0 @@
"""Record/replay transports behind the same ``Transport`` protocol (LIT-5729).
``RecordingTransport`` decorates the live transport: every call passes through
unchanged and its request/response pair is appended to the fixture bundle.
``ReplayTransport`` implements the protocol from a recorded bundle alone: no
HTTP, no proxy, no provider spend. Because both fulfil ``Transport``, no test
or client changes shape; ``build_proxy_client`` picks the transport from
``E2E_FIXTURE_MODE`` (live | record | replay, default live).
Replay matches each call by test node id and canonical content key
(fixture_canonical.py, LIT-5741): volatile headers, credential fields, unique
markers, generated ids, and timestamps are canonicalized out before hashing, so
matching is order-independent across distinct keys, FIFO within a key, and a
miss fails hard (``ReplayMiss``) printing the computed key and the closest
recorded key without ever falling through to a live call. Streaming chunk
fidelity is LIT-5742; scoping record/replay to provider-bound traffic is
LIT-5745.
"""
from __future__ import annotations
import difflib
import functools
import hashlib
import os
from collections import deque
from dataclasses import dataclass, field
from datetime import datetime
from itertools import islice
from pathlib import Path
from typing import Final, Literal, assert_never
from pydantic import BaseModel, JsonValue
from e2e_http import AuthHeaders, BinaryStream, ProbeResult, Result, StreamingResponse
from fixture_bundle import (
BundleRecorder,
FreshBundle,
Interaction,
LoadedBundle,
RecordedBinary,
RecordedProbe,
RecordedRequest,
RecordedResponse,
RecordedResult,
RecordedStreaming,
StaleBundle,
UnreadableBundle,
UnsafeBundleDir,
check_freshness,
format_age,
from_result,
interaction_filename,
load_bundle,
prepare_bundle,
slug_for_test,
to_json_value,
to_result,
)
from fixture_canonical import CanonicalRequest, canonicalize, is_secret_field
from transport import Transport
type FixtureMode = Literal["live", "record", "replay"]
FIXTURE_MODES: Final[tuple[FixtureMode, ...]] = ("live", "record", "replay")
SESSION_TEST_KEY: Final = "session"
REDACTED_HEADER_NAMES: Final[frozenset[str]] = frozenset({"authorization", "x-litellm-api-key"})
REDACTED_VALUE: Final = "<redacted>"
@dataclass(frozen=True, slots=True)
class InvalidFixtureMode:
value: str
def parse_fixture_mode(raw: str) -> FixtureMode | InvalidFixtureMode:
normalized = raw.strip().lower() or "live"
match normalized:
case "live" | "record" | "replay":
return normalized
case _:
return InvalidFixtureMode(value=raw)
def current_test_key() -> str:
"""The pytest node id of the running test, from the PYTEST_CURRENT_TEST env
var pytest maintains (``<nodeid> (setup|call|teardown)``); ``session`` for
calls outside any test (e.g. session-finish cleanup)."""
raw = os.environ.get("PYTEST_CURRENT_TEST", "")
if not raw:
return SESSION_TEST_KEY
return raw.rsplit(" (", 1)[0]
class ReplayMiss(AssertionError):
"""Replay had no recorded interaction for a call the suite made. The test
drifted from the bundle (or the bundle from the suite): re-record."""
_marker_ordinals: Final[dict[str, int]] = {}
def deterministic_marker() -> str:
"""Stable stand-in for uuid-based unique markers in record and replay modes:
the Nth marker of a test is a pure function of the test's node id and N, so a
replay run regenerates exactly the model names, prompts, and tags the record
run sent and every recorded poll response still satisfies its predicate."""
test_key = current_test_key()
ordinal = _marker_ordinals.get(test_key, 0)
_marker_ordinals[test_key] = ordinal + 1
return hashlib.sha1(f"{test_key}#{ordinal}".encode()).hexdigest()[:12]
def _dump_flat(model: BaseModel | None) -> dict[str, str]:
if model is None:
return {}
dumped: dict[str, object] = model.model_dump(by_alias=True, exclude_none=True)
return {key: str(value) for key, value in dumped.items()}
def _redact(headers: dict[str, str]) -> dict[str, str]:
return {
name: REDACTED_VALUE if name.lower() in REDACTED_HEADER_NAMES else value
for name, value in headers.items()
}
def _redact_secret_fields(value: JsonValue) -> JsonValue:
match value:
case dict():
return {
key: REDACTED_VALUE
if is_secret_field(key) and item is not None
else _redact_secret_fields(item)
for key, item in value.items()
}
case list():
return [_redact_secret_fields(item) for item in value]
case _:
return value
def _redact_flat(fields: dict[str, str]) -> dict[str, str]:
return {
key: REDACTED_VALUE if is_secret_field(key) else value for key, value in fields.items()
}
def recorded_request(
method: str,
path: str,
*,
headers: BaseModel,
body: BaseModel | None = None,
params: BaseModel | None = None,
form: BaseModel | None = None,
file_name: str | None = None,
file_content: bytes | None = None,
) -> RecordedRequest:
return RecordedRequest(
method=method,
path=path,
headers=_redact(_dump_flat(headers)),
params=_redact_flat(_dump_flat(params)),
body=None if body is None else _redact_secret_fields(to_json_value(body)),
form=None if form is None else _redact_flat(_dump_flat(form)),
file_name=file_name,
file_sha256=None if file_content is None else hashlib.sha256(file_content).hexdigest(),
file_bytes=None if file_content is None else len(file_content),
)
@dataclass(frozen=True, slots=True)
class RecordingTransport:
"""Decorator over the live transport: forwards every call and appends the
interaction to the bundle, so a green live run leaves behind exactly the
traffic replay needs."""
inner: Transport
recorder: BundleRecorder
def _record(self, request: RecordedRequest, response: RecordedResponse) -> None:
self.recorder.record(test_key=current_test_key(), request=request, response=response)
def bearer(self, key: str) -> AuthHeaders:
return self.inner.bearer(key)
@property
def master(self) -> AuthHeaders:
return self.inner.master
def post[R: BaseModel](
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
) -> Result[R]:
result = self.inner.post(path, headers=headers, json=json, response_type=response_type)
self._record(recorded_request("post", path, headers=headers, body=json), from_result(result))
return result
def get[R: BaseModel](
self,
path: str,
*,
headers: BaseModel,
params: BaseModel,
response_type: type[R],
timeout: float | None = None,
) -> Result[R]:
result = self.inner.get(
path, headers=headers, params=params, response_type=response_type, timeout=timeout
)
self._record(recorded_request("get", path, headers=headers, params=params), from_result(result))
return result
def delete[R: BaseModel](
self,
path: str,
*,
headers: BaseModel,
json: BaseModel,
response_type: type[R],
params: BaseModel | None = None,
) -> Result[R]:
result = self.inner.delete(
path, headers=headers, json=json, response_type=response_type, params=params
)
self._record(
recorded_request("delete", path, headers=headers, body=json, params=params),
from_result(result),
)
return result
def patch[R: BaseModel](
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
) -> Result[R]:
result = self.inner.patch(path, headers=headers, json=json, response_type=response_type)
self._record(recorded_request("patch", path, headers=headers, body=json), from_result(result))
return result
def put[R: BaseModel](
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
) -> Result[R]:
result = self.inner.put(path, headers=headers, json=json, response_type=response_type)
self._record(recorded_request("put", path, headers=headers, body=json), from_result(result))
return result
def stream(self, path: str, *, headers: BaseModel, json: BaseModel) -> StreamingResponse:
response = self.inner.stream(path, headers=headers, json=json)
self._record(
recorded_request("stream", path, headers=headers, body=json),
RecordedStreaming(payload=response),
)
return response
def stream_binary(
self, path: str, *, headers: BaseModel, json: BaseModel, chunk_size: int = 8192
) -> BinaryStream:
response = self.inner.stream_binary(path, headers=headers, json=json, chunk_size=chunk_size)
self._record(
recorded_request("stream_binary", path, headers=headers, body=json),
RecordedBinary(payload=response),
)
return response
def send(
self,
path: str,
*,
headers: BaseModel,
json: BaseModel,
params: BaseModel | None = None,
stream: bool = False,
) -> StreamingResponse:
response = self.inner.send(path, headers=headers, json=json, params=params, stream=stream)
self._record(
recorded_request("send", path, headers=headers, body=json, params=params),
RecordedStreaming(payload=response),
)
return response
def probe(self, path: str, *, params: BaseModel) -> ProbeResult:
response = self.inner.probe(path, params=params)
self._record(
recorded_request("probe", path, headers=self.master, params=params),
RecordedProbe(payload=response),
)
return response
def upload[R: BaseModel](
self,
path: str,
*,
headers: BaseModel,
form: BaseModel,
filename: str,
content: bytes,
file_content_type: str = "application/jsonl",
file_field: str = "file",
params: BaseModel | None = None,
response_type: type[R],
) -> Result[R]:
result = self.inner.upload(
path,
headers=headers,
form=form,
filename=filename,
content=content,
file_content_type=file_content_type,
file_field=file_field,
params=params,
response_type=response_type,
)
self._record(
recorded_request(
"upload",
path,
headers=headers,
params=params,
form=form,
file_name=filename,
file_content=content,
),
from_result(result),
)
return result
def download(self, path: str, *, headers: BaseModel) -> StreamingResponse:
response = self.inner.download(path, headers=headers)
self._record(
recorded_request("download", path, headers=headers),
RecordedStreaming(payload=response),
)
return response
def _build_pool(recorded: tuple[Interaction, ...]) -> dict[str, deque[Interaction]]:
keys: Final = tuple(canonicalize(interaction.request).key for interaction in recorded)
return {
key: deque(
interaction
for candidate_key, interaction in zip(keys, recorded, strict=True)
if candidate_key == key
)
for key in dict.fromkeys(keys)
}
def _closest_recorded(
canonical: CanonicalRequest, recorded: tuple[Interaction, ...]
) -> tuple[CanonicalRequest, str]:
candidates: Final = tuple(canonicalize(interaction.request) for interaction in recorded)
ratios: Final = tuple(
difflib.SequenceMatcher(
None, f"{canonical.method} {canonical.path}\n{canonical.content}",
f"{candidate.method} {candidate.path}\n{candidate.content}",
).ratio()
for candidate in candidates
)
best: Final = max(range(len(candidates)), key=lambda index: ratios[index])
return candidates[best], interaction_filename(best, recorded[best].request)
def _miss_message(test_key: str, slug: str, canonical: CanonicalRequest, bundle: LoadedBundle) -> str:
recorded: Final = bundle.interactions.get(slug, ())
if not recorded:
return (
f"replay miss for {test_key}: computed key {canonical.key} but nothing is recorded "
f"under {slug}; re-record with E2E_FIXTURE_MODE=record"
)
closest, closest_file = _closest_recorded(canonical, recorded)
diff: Final = "\n".join(
islice(
difflib.unified_diff(
closest.pretty_content().splitlines(),
canonical.pretty_content().splitlines(),
fromfile=f"closest recorded ({closest_file})",
tofile="test made",
lineterm="",
),
60,
)
)
return (
f"replay miss for {test_key}: no recorded interaction matches key {canonical.key}; "
f"closest recorded key is {closest.key} ({closest_file})\n{diff}\n"
"re-record with E2E_FIXTURE_MODE=record"
)
@dataclass(slots=True)
class ReplaySource:
"""One shared pool per test over a loaded bundle, so every client built in
the session consumes the same recorded interactions. Every pool is built
once at construction and per-key consumption is a single atomic deque pop,
so concurrent replay calls never race. Calls match by canonical content
key: order-independent across distinct keys (concurrent tests interleave
calls nondeterministically), FIFO within one key (a poll loop replays its
recorded responses in recorded order)."""
bundle: LoadedBundle
_pools: dict[str, dict[str, deque[Interaction]]] = field(init=False)
def __post_init__(self) -> None:
self._pools = {
slug: _build_pool(recorded) for slug, recorded in self.bundle.interactions.items()
}
def _pool(self, slug: str) -> dict[str, deque[Interaction]]:
return self._pools.get(slug, {})
def next_interaction(self, request: RecordedRequest) -> Interaction:
test_key: Final = current_test_key()
slug: Final = slug_for_test(test_key)
pool: Final = self._pool(slug)
canonical: Final = canonicalize(request)
queue: Final = pool.get(canonical.key)
if queue is None:
raise ReplayMiss(_miss_message(test_key, slug, canonical, self.bundle))
try:
return queue.popleft()
except IndexError:
raise ReplayMiss(
f"replay exhausted for {test_key}: every recorded interaction for key "
f"{canonical.key} is already consumed; re-record with E2E_FIXTURE_MODE=record"
) from None
def leftover_error(self, test_key: str) -> str | None:
"""Non-None when the test consumed fewer interactions than were recorded,
meaning a passing replay proved less than the bundle claims."""
slug: Final = slug_for_test(test_key)
recorded: Final = self.bundle.interactions.get(slug, ())
if not recorded:
return None
leftover: Final = tuple(
interaction for queue in self._pool(slug).values() for interaction in queue
)
if not leftover:
return None
return (
f"replay incomplete for {test_key}: {len(leftover)} of {len(recorded)} recorded "
f"interactions never consumed, e.g. {canonicalize(leftover[0].request).key}; "
"re-record with E2E_FIXTURE_MODE=record"
)
def _expect_result(interaction: Interaction) -> RecordedResult:
match interaction.response:
case RecordedResult() as recorded:
return recorded
case RecordedStreaming() | RecordedBinary() | RecordedProbe():
raise ReplayMiss(
f"recorded {interaction.request.method} {interaction.request.path} is not a typed result"
)
def _expect_streaming(interaction: Interaction) -> StreamingResponse:
match interaction.response:
case RecordedStreaming(payload=payload):
return payload
case RecordedResult() | RecordedBinary() | RecordedProbe():
raise ReplayMiss(
f"recorded {interaction.request.method} {interaction.request.path} is not a streaming response"
)
@dataclass(frozen=True, slots=True)
class ReplayTransport:
"""A ``Transport`` served entirely from a recorded bundle: never opens a
connection, so a replay run cannot bill a provider."""
source: ReplaySource
master_key: str
def bearer(self, key: str) -> AuthHeaders:
return AuthHeaders(authorization=f"Bearer {key}")
@property
def master(self) -> AuthHeaders:
return self.bearer(self.master_key)
def post[R: BaseModel](
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
) -> Result[R]:
return to_result(
_expect_result(
self.source.next_interaction(recorded_request("post", path, headers=headers, body=json))
),
response_type,
)
def get[R: BaseModel](
self,
path: str,
*,
headers: BaseModel,
params: BaseModel,
response_type: type[R],
timeout: float | None = None,
) -> Result[R]:
return to_result(
_expect_result(
self.source.next_interaction(recorded_request("get", path, headers=headers, params=params))
),
response_type,
)
def delete[R: BaseModel](
self,
path: str,
*,
headers: BaseModel,
json: BaseModel,
response_type: type[R],
params: BaseModel | None = None,
) -> Result[R]:
return to_result(
_expect_result(
self.source.next_interaction(
recorded_request("delete", path, headers=headers, body=json, params=params)
)
),
response_type,
)
def patch[R: BaseModel](
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
) -> Result[R]:
return to_result(
_expect_result(
self.source.next_interaction(recorded_request("patch", path, headers=headers, body=json))
),
response_type,
)
def put[R: BaseModel](
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
) -> Result[R]:
return to_result(
_expect_result(
self.source.next_interaction(recorded_request("put", path, headers=headers, body=json))
),
response_type,
)
def stream(self, path: str, *, headers: BaseModel, json: BaseModel) -> StreamingResponse:
return _expect_streaming(
self.source.next_interaction(recorded_request("stream", path, headers=headers, body=json))
)
def stream_binary(
self, path: str, *, headers: BaseModel, json: BaseModel, chunk_size: int = 8192
) -> BinaryStream:
interaction = self.source.next_interaction(
recorded_request("stream_binary", path, headers=headers, body=json)
)
match interaction.response:
case RecordedBinary(payload=payload):
return payload
case RecordedResult() | RecordedStreaming() | RecordedProbe():
raise ReplayMiss(
f"recorded stream_binary {interaction.request.path} is not a binary stream"
)
def send(
self,
path: str,
*,
headers: BaseModel,
json: BaseModel,
params: BaseModel | None = None,
stream: bool = False,
) -> StreamingResponse:
return _expect_streaming(
self.source.next_interaction(
recorded_request("send", path, headers=headers, body=json, params=params)
)
)
def probe(self, path: str, *, params: BaseModel) -> ProbeResult:
interaction = self.source.next_interaction(
recorded_request("probe", path, headers=self.master, params=params)
)
match interaction.response:
case RecordedProbe(payload=payload):
return payload
case RecordedResult() | RecordedStreaming() | RecordedBinary():
raise ReplayMiss(f"recorded probe {interaction.request.path} is not a probe result")
def upload[R: BaseModel](
self,
path: str,
*,
headers: BaseModel,
form: BaseModel,
filename: str,
content: bytes,
file_content_type: str = "application/jsonl",
file_field: str = "file",
params: BaseModel | None = None,
response_type: type[R],
) -> Result[R]:
return to_result(
_expect_result(
self.source.next_interaction(
recorded_request(
"upload",
path,
headers=headers,
params=params,
form=form,
file_name=filename,
file_content=content,
)
)
),
response_type,
)
def download(self, path: str, *, headers: BaseModel) -> StreamingResponse:
return _expect_streaming(
self.source.next_interaction(recorded_request("download", path, headers=headers))
)
@functools.lru_cache(maxsize=8)
def _shared_recorder(root: Path) -> BundleRecorder:
prepared = prepare_bundle(root)
if isinstance(prepared, UnsafeBundleDir):
raise ValueError(f"E2E_FIXTURE_DIR {prepared.path} {prepared.reason}")
return prepared
@functools.lru_cache(maxsize=8)
def _shared_replay_source(root: Path) -> ReplaySource:
loaded = load_bundle(root)
if isinstance(loaded, UnreadableBundle):
raise ValueError(f"cannot replay from {root}: {loaded.reason}")
return ReplaySource(bundle=loaded)
def replay_leftover_error(*, mode_raw: str, bundle_dir: Path, test_key: str) -> str | None:
"""Teardown-time completeness check: in replay mode a passed test with
unconsumed recorded interactions must fail instead of passing against a
recording it no longer matches. Inert in every other mode."""
if parse_fixture_mode(mode_raw) != "replay":
return None
return _shared_replay_source(bundle_dir).leftover_error(test_key)
def select_transport(
live: Transport, *, mode_raw: str, bundle_dir: Path, master_key: str
) -> Transport:
"""The one seam every client build goes through: wraps (record), replaces
(replay), or passes through (live) the transport per E2E_FIXTURE_MODE. The
recorder and replay cursors are process-wide singletons per bundle dir, so
every client in a session shares one bundle and one recorded sequence."""
mode = parse_fixture_mode(mode_raw)
match mode:
case InvalidFixtureMode(value=value):
raise ValueError(f"E2E_FIXTURE_MODE={value!r} is not one of {', '.join(FIXTURE_MODES)}")
case "live":
return live
case "record":
return RecordingTransport(inner=live, recorder=_shared_recorder(bundle_dir))
case "replay":
return ReplayTransport(source=_shared_replay_source(bundle_dir), master_key=master_key)
case _:
assert_never(mode)
def fixture_mode_collection_error(mode_raw: str, bundle_dir: Path, *, now: datetime) -> str | None:
"""Session-abort reason for a fixture-mode setup that can never work, or None.
Called at collection time (conftest pytest_sessionstart) so a stale or missing
bundle fails the whole run up front, naming the bundle age, instead of failing
every test individually."""
mode = parse_fixture_mode(mode_raw)
match mode:
case InvalidFixtureMode(value=value):
return f"E2E_FIXTURE_MODE={value!r} is not one of {', '.join(FIXTURE_MODES)}"
case "live" | "record":
return None
case "replay":
freshness = check_freshness(bundle_dir, now=now)
match freshness:
case FreshBundle():
return None
case StaleBundle(recorded_at=recorded_at, age=age, limit=limit):
return (
f"fixture bundle at {bundle_dir} is stale: recorded {recorded_at.isoformat()}, "
f"age {format_age(age)} exceeds the {limit.days}-day limit; "
"re-record with E2E_FIXTURE_MODE=record"
)
case UnreadableBundle(reason=reason):
return f"E2E_FIXTURE_MODE=replay cannot use bundle at {bundle_dir}: {reason}"
case _:
assert_never(freshness)
case _:
assert_never(mode)
def fixture_report_lines(mode_raw: str, bundle_dir: Path, *, now: datetime) -> list[str]:
"""pytest report-header lines; empty in live mode so an unset
E2E_FIXTURE_MODE keeps today's output byte-identical."""
mode = parse_fixture_mode(mode_raw)
match mode:
case InvalidFixtureMode() | "live":
return []
case "record":
return [f"e2e fixture mode: record -> {bundle_dir}"]
case "replay":
freshness = check_freshness(bundle_dir, now=now)
match freshness:
case FreshBundle(manifest=manifest):
return [
f"e2e fixture mode: replay <- {bundle_dir} "
f"(recorded {manifest.recorded_at.isoformat()}, harness {manifest.harness_version})"
]
case StaleBundle() | UnreadableBundle():
return [f"e2e fixture mode: replay <- {bundle_dir}"]
case _:
assert_never(freshness)
case _:
assert_never(mode)

11
tests/e2e/llm_translation/test_audio_transcriptions_e2e.py
View file

@ -3,7 +3,10 @@
Registers an OpenAI speech-to-text deployment at runtime and uploads a spoken
weather question (the realtime suite's 24kHz WAV fixture) as multipart, asserting
the returned transcript is non-empty and mentions the word it was asked about.
Also pins missing file/model negatives.
Also pins missing file/model negatives. A model-less request comes back as one of
two 400s depending on whether any wildcard deployment happens to be registered on
the shared proxy, so the assertion accepts either phrasing and holds both to naming
the model as the problem.
"""
from __future__ import annotations
@ -25,6 +28,8 @@ WEATHER_WAV = (
Path(__file__).resolve().parent / "realtime" / "fixtures" / "weather_question_24k.wav"
)
MISSING_MODEL_PHRASES: Final = ("model=none", "invalid model", "model is required")
class _OptionalTranscriptionForm(BaseModel):
model: str | None = None
@ -105,8 +110,8 @@ class TestAudioTranscriptions:
match result:
case UnknownApiError(status_code=400, body=body):
lowered: Final = body.lower()
assert "model" in lowered and ("required" in lowered or "invalid model" in lowered), (
f"missing model error must identify the required model: {body[:300]}"
assert any(phrase in lowered for phrase in MISSING_MODEL_PHRASES), (
f"missing model error must name the model as the problem: {body[:300]}"
)
case other:
pytest.fail(f"missing model expected a model-specific 400, got {other!r}")

546
tests/e2e/provider_edge.py Normal file
View file

@ -0,0 +1,546 @@
"""Provider-edge record/replay server for e2e runs (LIT-5745).
Record and replay scope to provider-bound traffic only: the proxy boots for
real, tests hit it for real, and only the hop from the proxy to the provider
is recorded or served from a bundle. Suites opt in per deployment by pointing
``litellm_params.api_base`` at ``provider_edge_api_base(mount)``, which is an
in-process HTTP server mounting each supported provider under a path prefix
(``http://127.0.0.1:<port>/openai`` forwards to ``https://api.openai.com``).
In record mode the edge relays each request verbatim, stores the interaction,
and serves the proxy the same filtered response replay will serve later; in
replay mode it serves straight from the bundle and never opens a provider
connection, so a green replay run with a fake provider key proves the entire
proxy pipeline (auth, routing, spend logging) without provider spend.
Request identity reuses fixture_canonical.py: interactions match by canonical
content key, order-independent across keys and FIFO within one. Edge requests
store no headers at all: SDK telemetry headers vary run to run and credential
headers must never touch disk. An unmatched replay call returns HTTP
``REPLAY_MISS_STATUS`` naming the closest recorded interaction, which the
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.
"""
from __future__ import annotations
import base64
import difflib
import functools
import hashlib
import threading
from collections import deque
from collections.abc import Mapping
from dataclasses import dataclass, field
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from itertools import islice
from pathlib import Path
from types import MappingProxyType
from typing import Final, Literal, assert_never
from urllib.parse import parse_qsl, urlsplit
from pydantic import JsonValue, TypeAdapter
from e2e_http import NetworkError, RawResponse, forward
from fixture_bundle import (
BundleRecorder,
Interaction,
LoadedBundle,
RecordedHttpResponse,
RecordedRequest,
UnreadableBundle,
UnsafeBundleDir,
interaction_filename,
load_bundle,
prepare_bundle,
slug_for_test,
)
from fixture_canonical import CanonicalRequest, canonical_string, canonicalize
from fixture_mode import (
FIXTURE_MODES,
InvalidFixtureMode,
ReplayMiss,
current_test_key,
parse_fixture_mode,
)
EDGE_MOUNTS: Final[Mapping[str, str]] = MappingProxyType(
{
"openai": "https://api.openai.com",
"anthropic": "https://api.anthropic.com",
}
)
REPLAY_MISS_STATUS: Final = 599
_HOP_BY_HOP_HEADERS: Final[frozenset[str]] = frozenset(
{
"connection",
"keep-alive",
"proxy-authenticate",
"proxy-authorization",
"te",
"trailers",
"transfer-encoding",
"upgrade",
}
)
_REQUEST_DROPPED_HEADERS: Final[frozenset[str]] = _HOP_BY_HOP_HEADERS | {
"host",
"content-length",
"accept-encoding",
}
_RESPONSE_DROPPED_HEADERS: Final[frozenset[str]] = _HOP_BY_HOP_HEADERS | {
"content-encoding",
"content-length",
"set-cookie",
}
_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")
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),
)
return RecordedRequest(method=method.lower(), path=path, headers={}, params=params, body=parsed)
def _build_pool(recorded: tuple[Interaction, ...]) -> dict[str, deque[Interaction]]:
keys: Final = tuple(canonicalize(interaction.request).key for interaction in recorded)
return {
key: deque(
interaction
for candidate_key, interaction in zip(keys, recorded, strict=True)
if candidate_key == key
)
for key in dict.fromkeys(keys)
}
def _closest_recorded(
canonical: CanonicalRequest, recorded: tuple[Interaction, ...]
) -> tuple[CanonicalRequest, str]:
candidates: Final = tuple(canonicalize(interaction.request) for interaction in recorded)
ratios: Final = tuple(
difflib.SequenceMatcher(
None, f"{canonical.method} {canonical.path}\n{canonical.content}",
f"{candidate.method} {candidate.path}\n{candidate.content}",
).ratio()
for candidate in candidates
)
best: Final = max(range(len(candidates)), key=lambda index: ratios[index])
return candidates[best], interaction_filename(best, recorded[best].request)
def _miss_message(test_key: str, slug: str, canonical: CanonicalRequest, bundle: LoadedBundle) -> str:
recorded: Final = bundle.interactions.get(slug, ())
if not recorded:
return (
f"replay miss for {test_key}: computed key {canonical.key} but nothing is recorded "
f"under {slug}; re-record with E2E_FIXTURE_MODE=record"
)
closest, closest_file = _closest_recorded(canonical, recorded)
diff: Final = "\n".join(
islice(
difflib.unified_diff(
closest.pretty_content().splitlines(),
canonical.pretty_content().splitlines(),
fromfile=f"closest recorded ({closest_file})",
tofile="test made",
lineterm="",
),
60,
)
)
return (
f"replay miss for {test_key}: no recorded interaction matches key {canonical.key}; "
f"closest recorded key is {closest.key} ({closest_file})\n{diff}\n"
"re-record with E2E_FIXTURE_MODE=record"
)
@dataclass(slots=True)
class ReplaySource:
"""One shared pool per test over a loaded bundle, so every provider call the
proxy makes in the session consumes from the same recorded interactions.
Every pool is built once at construction and per-key consumption is a single
atomic deque pop, so concurrent replay calls never race. Calls match by
canonical content key: order-independent across distinct keys (concurrent
tests interleave calls nondeterministically), FIFO within one key (a retry
or poll loop replays its recorded responses in recorded order)."""
bundle: LoadedBundle
_pools: dict[str, dict[str, deque[Interaction]]] = field(init=False)
def __post_init__(self) -> None:
self._pools = {
slug: _build_pool(recorded) for slug, recorded in self.bundle.interactions.items()
}
def _pool(self, slug: str) -> dict[str, deque[Interaction]]:
return self._pools.get(slug, {})
def next_interaction(self, request: RecordedRequest) -> Interaction:
test_key: Final = current_test_key()
slug: Final = slug_for_test(test_key)
pool: Final = self._pool(slug)
canonical: Final = canonicalize(request)
queue: Final = pool.get(canonical.key)
if queue is None:
raise ReplayMiss(_miss_message(test_key, slug, canonical, self.bundle))
try:
return queue.popleft()
except IndexError:
raise ReplayMiss(
f"replay exhausted for {test_key}: every recorded interaction for key "
f"{canonical.key} is already consumed; re-record with E2E_FIXTURE_MODE=record"
) from None
def leftover_error(self, test_key: str) -> str | None:
"""Non-None when the test consumed fewer interactions than were recorded,
meaning a passing replay proved less than the bundle claims."""
slug: Final = slug_for_test(test_key)
recorded: Final = self.bundle.interactions.get(slug, ())
if not recorded:
return None
leftover: Final = tuple(
interaction for queue in self._pool(slug).values() for interaction in queue
)
if not leftover:
return None
return (
f"replay incomplete for {test_key}: {len(leftover)} of {len(recorded)} recorded "
f"interactions never consumed, e.g. {canonicalize(leftover[0].request).key}; "
"re-record with E2E_FIXTURE_MODE=record"
)
@dataclass(frozen=True, slots=True)
class RecordEdge:
"""Record backend: forward to the provider, persist, serve the filtered copy.
The lock serializes recorder writes because the edge server handles requests
on concurrent threads."""
recorder: BundleRecorder
lock: threading.Lock
@dataclass(frozen=True, slots=True)
class ReplayEdge:
source: ReplaySource
type EdgeBackend = RecordEdge | ReplayEdge
@dataclass(frozen=True, slots=True)
class EdgeReply:
status_code: int
headers: dict[str, str]
body: bytes
def _text_reply(status_code: int, message: str) -> EdgeReply:
return EdgeReply(
status_code=status_code,
headers={"content-type": "text/plain; charset=utf-8"},
body=message.encode(),
)
def _reply_from_recorded(response: RecordedHttpResponse) -> EdgeReply:
return EdgeReply(
status_code=response.status_code,
headers=dict(response.headers),
body=base64.b64decode(response.body_b64),
)
def _recorded_response(outcome: RawResponse | NetworkError) -> RecordedHttpResponse:
match outcome:
case RawResponse(status_code=status_code, headers=headers, body=body):
return RecordedHttpResponse(
status_code=status_code,
headers={
name: value
for name, value in headers.items()
if name not in _RESPONSE_DROPPED_HEADERS
},
body_b64=base64.b64encode(body).decode("ascii"),
)
case NetworkError(message=message):
return RecordedHttpResponse(
status_code=502,
headers={"content-type": "text/plain; charset=utf-8"},
body_b64=base64.b64encode(
f"provider edge could not reach the provider: {message}".encode()
).decode("ascii"),
)
def _upstream_url(upstream_base: str, upstream_path: str, query: str) -> str:
url: Final = f"{upstream_base}/{upstream_path}"
return f"{url}?{query}" if query else url
def _handle_record(
backend: RecordEdge,
request: RecordedRequest,
*,
method: str,
url: str,
headers: Mapping[str, str],
body: bytes | None,
timeout: float,
) -> EdgeReply:
forwarded: Final = {
name: value for name, value in headers.items() if name.lower() not in _REQUEST_DROPPED_HEADERS
}
outcome: Final = forward(method, url, headers=forwarded, body=body, timeout=timeout)
response: Final = _recorded_response(outcome)
with backend.lock:
backend.recorder.record(test_key=current_test_key(), request=request, response=response)
return _reply_from_recorded(response)
def _handle_replay(source: ReplaySource, request: RecordedRequest) -> EdgeReply:
try:
interaction: Final = source.next_interaction(request)
except ReplayMiss as miss:
return _text_reply(REPLAY_MISS_STATUS, str(miss))
return _reply_from_recorded(interaction.response)
def handle_edge_request(
backend: EdgeBackend,
mounts: Mapping[str, str],
method: str,
raw_path: str,
headers: Mapping[str, str],
body: bytes | None,
*,
timeout: float,
) -> EdgeReply:
"""The edge's pure core, one HTTP exchange in and out: resolve the mount
prefix, then record (forward + persist) or replay (serve from the bundle).
Socket-free so unit tests exercise every branch without a server."""
split: Final = urlsplit(raw_path)
mount, _, upstream_path = split.path.lstrip("/").partition("/")
upstream_base: Final = mounts.get(mount)
if upstream_base is None:
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)
match backend:
case RecordEdge():
return _handle_record(
backend,
request,
method=method,
url=_upstream_url(upstream_base, upstream_path, split.query),
headers=headers,
body=body,
timeout=timeout,
)
case ReplayEdge(source=source):
return _handle_replay(source, request)
case _:
assert_never(backend)
class _EdgeHandler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
def do_GET(self) -> None:
self._handle()
def do_POST(self) -> None:
self._handle()
def do_PUT(self) -> None:
self._handle()
def do_PATCH(self) -> None:
self._handle()
def do_DELETE(self) -> None:
self._handle()
def _handle(self) -> None:
edge_server: Final = self.server
assert isinstance(edge_server, _EdgeHTTPServer)
length: Final = int(self.headers.get("content-length") or "0")
body: Final = self.rfile.read(length) if length else None
reply: Final = handle_edge_request(
edge_server.backend,
edge_server.mounts,
self.command,
self.path,
{name.lower(): value for name, value in self.headers.items()},
body,
timeout=edge_server.forward_timeout,
)
self.send_response(reply.status_code)
for name, value in reply.headers.items():
self.send_header(name, value)
self.send_header("content-length", str(len(reply.body)))
self.end_headers()
self.wfile.write(reply.body)
def log_message(self, format: str, *args: object) -> None:
"""Silence the per-request stderr line BaseHTTPRequestHandler emits."""
class _EdgeHTTPServer(ThreadingHTTPServer):
daemon_threads = True
def __init__(
self,
bind: tuple[str, int],
*,
backend: EdgeBackend,
mounts: Mapping[str, str],
forward_timeout: float,
) -> None:
super().__init__(bind, _EdgeHandler)
self.backend: Final = backend
self.mounts: Final = mounts
self.forward_timeout: Final = forward_timeout
@dataclass(frozen=True, slots=True)
class ProviderEdge:
port: int
advertise_host: str
def api_base(self, mount: str) -> str:
return f"http://{self.advertise_host}:{self.port}/{mount}"
@dataclass(frozen=True, slots=True)
class RunningEdge:
edge: ProviderEdge
server: _EdgeHTTPServer
def shutdown(self) -> None:
self.server.shutdown()
self.server.server_close()
def start_provider_edge(
backend: EdgeBackend,
*,
mounts: Mapping[str, str] = EDGE_MOUNTS,
bind_host: str = "127.0.0.1",
advertise_host: str | None = None,
forward_timeout: float = 60.0,
) -> RunningEdge:
"""Boot an edge server on an OS-assigned port in a daemon thread.
``advertise_host`` is what api_base URLs name (it differs from the bind
host when the proxy runs in a container and reaches the host machine via
a gateway address like host.docker.internal)."""
server: Final = _EdgeHTTPServer(
(bind_host, 0), backend=backend, mounts=mounts, forward_timeout=forward_timeout
)
thread: Final = threading.Thread(target=server.serve_forever, name="e2e-provider-edge", daemon=True)
thread.start()
return RunningEdge(
edge=ProviderEdge(port=server.server_address[1], advertise_host=advertise_host or bind_host),
server=server,
)
@functools.lru_cache(maxsize=8)
def _shared_recorder(root: Path) -> BundleRecorder:
prepared = prepare_bundle(root)
if isinstance(prepared, UnsafeBundleDir):
raise ValueError(f"E2E_FIXTURE_DIR {prepared.path} {prepared.reason}")
return prepared
@functools.lru_cache(maxsize=8)
def _shared_replay_source(root: Path) -> ReplaySource:
loaded = load_bundle(root)
if isinstance(loaded, UnreadableBundle):
raise ValueError(f"cannot replay from {root}: {loaded.reason}")
return ReplaySource(bundle=loaded)
@functools.lru_cache(maxsize=8)
def _shared_edge(
mode: Literal["record", "replay"],
bundle_dir: Path,
bind_host: str,
advertise_host: str,
forward_timeout: float,
) -> ProviderEdge:
backend: Final[EdgeBackend] = (
RecordEdge(recorder=_shared_recorder(bundle_dir), lock=threading.Lock())
if mode == "record"
else ReplayEdge(source=_shared_replay_source(bundle_dir))
)
return start_provider_edge(
backend,
mounts=EDGE_MOUNTS,
bind_host=bind_host,
advertise_host=advertise_host,
forward_timeout=forward_timeout,
).edge
def replay_leftover_error(*, mode_raw: str, bundle_dir: Path, test_key: str) -> str | None:
"""Teardown-time completeness check: in replay mode a passed test with
unconsumed recorded interactions must fail instead of passing against a
recording it no longer matches. Inert in every other mode."""
if parse_fixture_mode(mode_raw) != "replay":
return None
return _shared_replay_source(bundle_dir).leftover_error(test_key)
def provider_edge_api_base(
mount: str,
*,
mode_raw: str,
bundle_dir: Path,
bind_host: str,
advertise_host: 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."""
mode: Final = parse_fixture_mode(mode_raw)
match mode:
case InvalidFixtureMode(value=value):
raise ValueError(f"E2E_FIXTURE_MODE={value!r} is not one of {', '.join(FIXTURE_MODES)}")
case "live":
return None
case "record" | "replay":
if mount not in EDGE_MOUNTS:
raise ValueError(
f"unknown provider mount {mount!r}; known mounts: {', '.join(sorted(EDGE_MOUNTS))}"
)
return _shared_edge(mode, bundle_dir, bind_host, advertise_host, forward_timeout).api_base(mount)
case _:
assert_never(mode)

16
tests/e2e/proxy_client.py
View file

@ -65,8 +65,6 @@ from models import (
)
from e2e_config import (
CONTROL_PLANE_BASE_URL,
FIXTURE_DIR,
FIXTURE_MODE_RAW,
MASTER_KEY,
POLL_INTERVAL,
POLL_TIMEOUT,
@ -74,7 +72,6 @@ from e2e_config import (
REQUEST_TIMEOUT,
settle_propagation,
)
from fixture_transport import select_transport
from transport import HttpTransport, SplitTransport, Transport
RowsPredicate = Callable[[list[SpendLogRow]], bool]
@ -547,9 +544,9 @@ def build_proxy_client(
pass all three together, since a caller that overrides only the data plane
would leave management calls pointed at the env default.
E2E_FIXTURE_MODE wraps (record) or replaces (replay) the transport here, so
every client built from this seam records or replays without changing shape;
unset it stays the plain SplitTransport (see fixture_transport.py)."""
Test-to-proxy traffic always goes over the wire, in every E2E_FIXTURE_MODE:
record and replay scope to the proxy's provider-bound calls via the
provider edge (see provider_edge.py), never to this transport."""
split = SplitTransport(
data=HttpTransport(
base_url=base_url,
@ -563,12 +560,7 @@ def build_proxy_client(
),
)
return ProxyClient(
transport=select_transport(
split,
mode_raw=FIXTURE_MODE_RAW,
bundle_dir=FIXTURE_DIR,
master_key=master_key,
),
transport=split,
poll_timeout=POLL_TIMEOUT,
poll_interval=POLL_INTERVAL,
)

50
tests/e2e/quota_management/spend_tracking/test_provider_edge_spend_e2e.py Normal file
View file

@ -0,0 +1,50 @@
"""The provider-edge demonstrator: one spend-tracking flow wired through the
record/replay edge (LIT-5745).
This is the reference for wiring a suite to the edge: register a deployment
whose ``api_base`` comes from ``e2e_config.provider_edge_base``, then exercise
the proxy exactly as a live test would. In live mode the base is None and the
deployment talks to the real provider; in record mode it talks through the
local edge, which forwards to the provider and captures the exchange; in
replay mode the same test drives the REAL proxy and REAL database on the
recorded provider traffic alone, so key auth, routing, and the spend-log
write path are all still under test with zero provider calls.
"""
import pytest
from e2e_config import CHEAP_OPENAI_MODEL, provider_edge_base
from lifecycle import ResourceManager
from models import LiteLLMParamsBody
from spend_e2e_client import SpendClient, unique_marker, unwrap
pytestmark = pytest.mark.e2e
@pytest.mark.covers("quota_management.spend_tracking.chat_completions.logs_cost")
def test_edge_wired_chat_writes_nonzero_spend_row(
client: SpendClient, resources: ResourceManager, scoped_key: str
) -> None:
base = provider_edge_base("openai")
model = f"e2e-edge-openai-{unique_marker()}"
model_id = client.proxy.create_model(
model,
LiteLLMParamsBody(
model=f"openai/{CHEAP_OPENAI_MODEL}",
api_key="os.environ/OPENAI_API_KEY",
api_base=None if base is None else f"{base}/v1",
),
)
resources.defer(lambda: client.proxy.delete_model(model_id))
chat = unwrap(
client.chat(scoped_key, model, f"reply with one word {unique_marker()}", max_tokens=16)
)
assert chat.id
rows = client.poll_logs_for_key(
scoped_key, predicate=lambda rs: any((r.spend or 0) > 0 for r in rs)
)
matching = [row for row in rows if row.request_id == chat.id]
assert matching, f"no SpendLogs row for request_id {chat.id}; saw {len(rows)} row(s)"
assert (matching[0].spend or 0) > 0, f"spend row for {chat.id} has zero spend"

46
tests/e2e/test_fixture_bundle.py
View file

@ -1,9 +1,9 @@
"""Harness coverage for the on-disk fixture bundle format (LIT-5729).
"""Harness coverage for the on-disk fixture bundle format (LIT-5729/LIT-5745).
No proxy and no ``e2e`` marker: these pin the bundle CONTRACT - the seven-day
freshness gate that names the bundle's age, record mode's wipe safety (never
delete a directory that is not a bundle), collision-free per-test slugs, and
lossless Result round-trips - so replay can never silently drift from what
grouped-in-order loading - so replay can never silently drift from what
record wrote.
"""
@ -12,18 +12,6 @@ from __future__ import annotations
from datetime import datetime, timedelta, timezone
from pathlib import Path
import pytest
from pydantic import BaseModel
from e2e_http import (
NetworkError,
RateLimitedError,
Result,
Success,
UnauthorizedError,
UnknownApiError,
ValidationError,
)
from fixture_bundle import (
BUNDLE_FORMAT_VERSION,
MANIFEST_FILENAME,
@ -32,28 +20,22 @@ from fixture_bundle import (
FreshBundle,
LoadedBundle,
Manifest,
RecordedHttpResponse,
RecordedRequest,
RecordedResult,
StaleBundle,
UnreadableBundle,
UnsafeBundleDir,
check_freshness,
format_age,
from_result,
interaction_filename,
load_bundle,
prepare_bundle,
slug_for_test,
to_result,
)
NOW = datetime(2026, 8, 18, 12, 0, 0, tzinfo=timezone.utc)
class Payload(BaseModel):
value: str
def write_manifest(
root: Path, recorded_at: datetime, *, format_version: int = BUNDLE_FORMAT_VERSION
) -> None:
@ -74,20 +56,8 @@ def plain_request(path: str) -> RecordedRequest:
return RecordedRequest(method="post", path=path, headers={})
class TestResultRoundTrip:
@pytest.mark.parametrize(
"result",
[
Success(status_code=201, data=Payload(value="ok")),
NetworkError(message="connection refused"),
UnauthorizedError(),
RateLimitedError(retry_after_seconds=7, body="slow down"),
ValidationError(message="bad shape"),
UnknownApiError(status_code=502, body="upstream exploded"),
],
)
def test_every_result_kind_survives_disk_and_back(self, result: Result[Payload]) -> None:
assert to_result(from_result(result), Payload) == result
def plain_response() -> RecordedHttpResponse:
return RecordedHttpResponse(status_code=401, headers={}, body_b64="")
class TestFreshness:
@ -144,7 +114,7 @@ class TestPrepareBundle:
prepared(root).record(
test_key="old.py::test_old",
request=plain_request("/stale"),
response=RecordedResult(kind="unauthorized"),
response=plain_response(),
)
assert any(entry.is_dir() for entry in root.iterdir())
prepared(root)
@ -193,7 +163,7 @@ class TestRecordAndLoad:
recorder.record(
test_key=key,
request=plain_request(path),
response=RecordedResult(kind="unauthorized"),
response=plain_response(),
)
loaded = load_bundle(root)
assert isinstance(loaded, LoadedBundle)
@ -208,7 +178,7 @@ class TestRecordAndLoad:
recorder.record(
test_key=key,
request=plain_request(f"/{key[-3:]}"),
response=RecordedResult(kind="unauthorized"),
response=plain_response(),
)
loaded = load_bundle(root)
assert isinstance(loaded, LoadedBundle)

114
tests/e2e/test_fixture_mode.py Normal file
View file

@ -0,0 +1,114 @@
"""Harness coverage for fixture-mode selection and determinism (LIT-5729/LIT-5745).
No proxy and no ``e2e`` marker. Pins the mode parser, the deterministic
per-test marker sequence a replay run must regenerate, the collection-time
gate (including the stale message that names the bundle's age), and the pytest
report header. The provider-edge record/replay behavior itself is pinned in
test_provider_edge.py.
"""
from __future__ import annotations
import hashlib
from datetime import datetime, timedelta, timezone
from pathlib import Path
import pytest
from fixture_bundle import BUNDLE_FORMAT_VERSION, MANIFEST_FILENAME, Manifest
from fixture_mode import (
InvalidFixtureMode,
current_test_key,
deterministic_marker,
fixture_mode_collection_error,
fixture_report_lines,
parse_fixture_mode,
)
NOW = datetime(2026, 8, 18, 12, 0, 0, tzinfo=timezone.utc)
def write_manifest(root: Path, recorded_at: datetime) -> None:
root.mkdir(parents=True, exist_ok=True)
manifest = Manifest(
format_version=BUNDLE_FORMAT_VERSION, recorded_at=recorded_at, harness_version="abc1234"
)
(root / MANIFEST_FILENAME).write_text(manifest.model_dump_json(), encoding="utf-8")
class TestParseFixtureMode:
@pytest.mark.parametrize(
("raw", "expected"),
[("live", "live"), ("record", "record"), ("replay", "replay"), ("", "live"), (" REPLAY ", "replay")],
)
def test_known_values_normalize(self, raw: str, expected: str) -> None:
assert parse_fixture_mode(raw) == expected
def test_unknown_value_is_invalid_with_the_original_spelling(self) -> None:
assert parse_fixture_mode("cached") == InvalidFixtureMode(value="cached")
class TestDeterministicMarker:
def test_sequence_is_a_pure_function_of_test_and_ordinal(self) -> None:
"""A replay process must regenerate exactly the markers the record
process generated, so the Nth marker of a test is pinned to a pure
function of the node id and N."""
key = current_test_key()
assert deterministic_marker() == hashlib.sha1(f"{key}#0".encode()).hexdigest()[:12]
assert deterministic_marker() == hashlib.sha1(f"{key}#1".encode()).hexdigest()[:12]
class TestCurrentTestKey:
def test_names_this_test_and_strips_the_phase(self) -> None:
key = current_test_key()
assert key.endswith("TestCurrentTestKey::test_names_this_test_and_strips_the_phase")
assert "(call)" not in key
class TestCollectionGate:
def test_invalid_mode_names_the_value_and_the_choices(self, tmp_path: Path) -> None:
assert (
fixture_mode_collection_error("cached", tmp_path, now=NOW)
== "E2E_FIXTURE_MODE='cached' is not one of live, record, replay"
)
@pytest.mark.parametrize("mode_raw", ["live", "", "record"])
def test_live_and_record_never_block_collection(self, mode_raw: str, tmp_path: Path) -> None:
assert fixture_mode_collection_error(mode_raw, tmp_path / "missing", now=NOW) is None
def test_replay_with_no_bundle_says_how_to_record_one(self, tmp_path: Path) -> None:
reason = fixture_mode_collection_error("replay", tmp_path / "missing", now=NOW)
assert reason is not None
assert f"no {MANIFEST_FILENAME}" in reason
assert "E2E_FIXTURE_MODE=record" in reason
def test_stale_replay_bundle_fails_naming_its_age(self, tmp_path: Path) -> None:
root = tmp_path / "bundle"
write_manifest(root, NOW - timedelta(days=9, hours=5))
reason = fixture_mode_collection_error("replay", root, now=NOW)
assert reason is not None
assert "age 9d5h exceeds the 7-day limit" in reason
assert "re-record with E2E_FIXTURE_MODE=record" in reason
def test_fresh_replay_bundle_collects(self, tmp_path: Path) -> None:
root = tmp_path / "bundle"
write_manifest(root, NOW - timedelta(days=2))
assert fixture_mode_collection_error("replay", root, now=NOW) is None
class TestReportHeader:
def test_live_mode_prints_nothing(self, tmp_path: Path) -> None:
assert fixture_report_lines("live", tmp_path, now=NOW) == []
assert fixture_report_lines("", tmp_path, now=NOW) == []
def test_record_and_replay_name_the_bundle(self, tmp_path: Path) -> None:
root = tmp_path / "bundle"
recorded_at = NOW - timedelta(days=1)
write_manifest(root, recorded_at)
assert fixture_report_lines("record", root, now=NOW) == [
f"e2e fixture mode: record -> {root}"
]
replay_lines = fixture_report_lines("replay", root, now=NOW)
assert len(replay_lines) == 1
assert "replay" in replay_lines[0]
assert recorded_at.isoformat() in replay_lines[0]

676
tests/e2e/test_fixture_transport.py
View file

@ -1,676 +0,0 @@
"""Harness coverage for the record/replay transports (LIT-5729).
No proxy and no ``e2e`` marker. A fake in-memory ``Transport`` stands in for
the live one (dependency injection, no monkeypatching): recording must pass
every value through unchanged while writing one redacted interaction file per
call, and replay must serve identical values from the bundle alone - the
fake's call log proves nothing reaches the inner transport - failing hard
(``ReplayMiss``) on any content drift, printing the computed canonical key and
the closest recorded key (LIT-5741; the pure canonicalizer is pinned in
test_fixture_canonical.py). The collection-time gate and report header are
pinned here too, including the stale message that names the bundle's age.
"""
from __future__ import annotations
import hashlib
import sys
import threading
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from pathlib import Path
from uuid import uuid4
import pytest
from pydantic import BaseModel
from e2e_http import (
AuthHeaders,
BinaryStream,
ProbeResult,
Result,
StreamingResponse,
Success,
)
from fixture_bundle import (
BUNDLE_FORMAT_VERSION,
MANIFEST_FILENAME,
BundleRecorder,
Interaction,
LoadedBundle,
Manifest,
RecordedResult,
load_bundle,
prepare_bundle,
slug_for_test,
)
from fixture_canonical import canonicalize
from fixture_transport import (
InvalidFixtureMode,
RecordingTransport,
ReplayMiss,
ReplaySource,
ReplayTransport,
current_test_key,
deterministic_marker,
fixture_mode_collection_error,
fixture_report_lines,
parse_fixture_mode,
recorded_request,
replay_leftover_error,
select_transport,
)
from transport import Transport
NOW = datetime(2026, 8, 18, 12, 0, 0, tzinfo=timezone.utc)
class Payload(BaseModel):
value: str
class Body(BaseModel):
prompt: str
class Query(BaseModel):
q: str
class DeployParams(BaseModel):
model: str
api_key: str | None = None
aws_secret_access_key: str | None = None
class DeployBody(BaseModel):
model_name: str
litellm_params: DeployParams
STREAMING = StreamingResponse(
status_code=200,
body="",
content_type="text/event-stream",
chunks=2,
stream_events=["one", "two"],
stream_done=True,
)
BINARY = BinaryStream(status_code=200, content_type="audio/mpeg", chunk_count=3, total_bytes=42)
PROBE = ProbeResult(status_code=200, body="alive")
@dataclass
class FakeTransport:
calls: list[str] = field(default_factory=list)
def bearer(self, key: str) -> AuthHeaders:
return AuthHeaders(authorization=f"Bearer {key}")
@property
def master(self) -> AuthHeaders:
return self.bearer("sk-fake-master")
def _success[R: BaseModel](self, response_type: type[R]) -> Result[R]:
return Success(status_code=200, data=response_type.model_validate({"value": "live"}))
def post[R: BaseModel](
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
) -> Result[R]:
self.calls.append(f"post {path}")
return self._success(response_type)
def get[R: BaseModel](
self,
path: str,
*,
headers: BaseModel,
params: BaseModel,
response_type: type[R],
timeout: float | None = None,
) -> Result[R]:
self.calls.append(f"get {path}")
return self._success(response_type)
def delete[R: BaseModel](
self,
path: str,
*,
headers: BaseModel,
json: BaseModel,
response_type: type[R],
params: BaseModel | None = None,
) -> Result[R]:
self.calls.append(f"delete {path}")
return self._success(response_type)
def patch[R: BaseModel](
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
) -> Result[R]:
self.calls.append(f"patch {path}")
return self._success(response_type)
def put[R: BaseModel](
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
) -> Result[R]:
self.calls.append(f"put {path}")
return self._success(response_type)
def stream(self, path: str, *, headers: BaseModel, json: BaseModel) -> StreamingResponse:
self.calls.append(f"stream {path}")
return STREAMING
def stream_binary(
self, path: str, *, headers: BaseModel, json: BaseModel, chunk_size: int = 8192
) -> BinaryStream:
self.calls.append(f"stream_binary {path}")
return BINARY
def send(
self,
path: str,
*,
headers: BaseModel,
json: BaseModel,
params: BaseModel | None = None,
stream: bool = False,
) -> StreamingResponse:
self.calls.append(f"send {path}")
return STREAMING
def probe(self, path: str, *, params: BaseModel) -> ProbeResult:
self.calls.append(f"probe {path}")
return PROBE
def upload[R: BaseModel](
self,
path: str,
*,
headers: BaseModel,
form: BaseModel,
filename: str,
content: bytes,
file_content_type: str = "application/jsonl",
file_field: str = "file",
params: BaseModel | None = None,
response_type: type[R],
) -> Result[R]:
self.calls.append(f"upload {path}")
return self._success(response_type)
def download(self, path: str, *, headers: BaseModel) -> StreamingResponse:
self.calls.append(f"download {path}")
return STREAMING
def make_recorder(root: Path) -> BundleRecorder:
recorder = prepare_bundle(root)
assert isinstance(recorder, BundleRecorder)
return recorder
def replay_source(root: Path) -> ReplaySource:
loaded = load_bundle(root)
assert isinstance(loaded, LoadedBundle)
return ReplaySource(bundle=loaded)
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 write_manifest(root: Path, recorded_at: datetime) -> None:
root.mkdir(parents=True, exist_ok=True)
manifest = Manifest(
format_version=BUNDLE_FORMAT_VERSION, recorded_at=recorded_at, harness_version="abc1234"
)
(root / MANIFEST_FILENAME).write_text(manifest.model_dump_json(), encoding="utf-8")
class TestParseFixtureMode:
@pytest.mark.parametrize(
("raw", "expected"),
[("live", "live"), ("record", "record"), ("replay", "replay"), ("", "live"), (" REPLAY ", "replay")],
)
def test_known_values_normalize(self, raw: str, expected: str) -> None:
assert parse_fixture_mode(raw) == expected
def test_unknown_value_is_invalid_with_the_original_spelling(self) -> None:
assert parse_fixture_mode("cached") == InvalidFixtureMode(value="cached")
class TestDeterministicMarker:
def test_sequence_is_a_pure_function_of_test_and_ordinal(self) -> None:
"""A replay process must regenerate exactly the markers the record
process generated, so the Nth marker of a test is pinned to a pure
function of the node id and N."""
key = current_test_key()
assert deterministic_marker() == hashlib.sha1(f"{key}#0".encode()).hexdigest()[:12]
assert deterministic_marker() == hashlib.sha1(f"{key}#1".encode()).hexdigest()[:12]
class TestCurrentTestKey:
def test_names_this_test_and_strips_the_phase(self) -> None:
key = current_test_key()
assert key.endswith("TestCurrentTestKey::test_names_this_test_and_strips_the_phase")
assert "(call)" not in key
class TestRecordingTransport:
def test_passes_the_result_through_and_writes_one_file_per_call(self, tmp_path: Path) -> None:
fake = FakeTransport()
root = tmp_path / "bundle"
recording: Transport = RecordingTransport(inner=fake, recorder=make_recorder(root))
result = recording.post(
"/model/new", headers=fake.master, json=Body(prompt="x"), response_type=Payload
)
assert result == Success(status_code=200, data=Payload(value="live"))
assert fake.calls == ["post /model/new"]
files = this_tests_files(root)
assert [file.name for file in files] == ["0000-post-model-new.json"]
interaction = Interaction.model_validate_json(files[0].read_text(encoding="utf-8"))
assert interaction.request.method == "post"
assert interaction.request.path == "/model/new"
def test_redacts_auth_header_values_in_the_recorded_request(self, tmp_path: Path) -> None:
fake = FakeTransport()
root = tmp_path / "bundle"
recording: Transport = RecordingTransport(inner=fake, recorder=make_recorder(root))
headers = AuthHeaders.model_validate(
{"authorization": "Bearer sk-secret", "x-litellm-api-key": "sk-other"}
)
recording.post("/key/generate", headers=headers, json=Body(prompt="x"), response_type=Payload)
interaction = Interaction.model_validate_json(
this_tests_files(root)[0].read_text(encoding="utf-8")
)
assert interaction.request.headers == {
"authorization": "<redacted>",
"x-litellm-api-key": "<redacted>",
}
assert "sk-secret" not in this_tests_files(root)[0].read_text(encoding="utf-8")
def test_redacts_credential_body_fields_in_the_recorded_request(self, tmp_path: Path) -> None:
fake = FakeTransport()
root = tmp_path / "bundle"
recording: Transport = RecordingTransport(inner=fake, recorder=make_recorder(root))
recording.post(
"/model/new",
headers=fake.master,
json=DeployBody(
model_name="m",
litellm_params=DeployParams(model="openai/gpt", api_key="sk-live-provider-secret-123456"),
),
response_type=Payload,
)
raw = this_tests_files(root)[0].read_text(encoding="utf-8")
interaction = Interaction.model_validate_json(raw)
assert "sk-live-provider-secret-123456" not in raw
assert isinstance(interaction.request.body, dict)
params = interaction.request.body["litellm_params"]
assert isinstance(params, dict)
assert params["api_key"] == "<redacted>"
assert params["aws_secret_access_key"] is None
def test_upload_records_a_content_digest_not_the_bytes(self, tmp_path: Path) -> None:
fake = FakeTransport()
root = tmp_path / "bundle"
recording: Transport = RecordingTransport(inner=fake, recorder=make_recorder(root))
recording.upload(
"/v1/files",
headers=fake.master,
form=Query(q="batch"),
filename="batch.jsonl",
content=b'{"custom_id": "1"}',
response_type=Payload,
)
interaction = Interaction.model_validate_json(
this_tests_files(root)[0].read_text(encoding="utf-8")
)
assert interaction.request.file_name == "batch.jsonl"
assert interaction.request.file_bytes == len(b'{"custom_id": "1"}')
assert interaction.request.file_sha256 is not None
assert "custom_id" not in interaction.request.model_dump_json()
class TestReplayTransport:
def test_serves_recorded_values_without_touching_the_inner_transport(
self, tmp_path: Path
) -> None:
fake = FakeTransport()
root = tmp_path / "bundle"
recording: Transport = RecordingTransport(inner=fake, recorder=make_recorder(root))
recorded_post = recording.post(
"/model/new", headers=fake.master, json=Body(prompt="x"), response_type=Payload
)
recorded_get = recording.get(
"/v1/models", headers=fake.master, params=Query(q="all"), response_type=Payload
)
recorded_stream = recording.stream(
"/chat/completions", headers=fake.master, json=Body(prompt="hi")
)
recorded_probe = recording.probe("/health/liveliness", params=Query(q="1"))
recorded_binary = recording.stream_binary(
"/v1/audio/speech", headers=fake.master, json=Body(prompt="say")
)
calls_after_record = list(fake.calls)
replay: Transport = ReplayTransport(source=replay_source(root), master_key="sk-1234")
assert (
replay.post("/model/new", headers=replay.master, json=Body(prompt="x"), response_type=Payload)
== recorded_post
)
assert (
replay.get("/v1/models", headers=replay.master, params=Query(q="all"), response_type=Payload)
== recorded_get
)
assert (
replay.stream("/chat/completions", headers=replay.master, json=Body(prompt="hi"))
== recorded_stream
)
assert replay.probe("/health/liveliness", params=Query(q="1")) == recorded_probe
assert (
replay.stream_binary("/v1/audio/speech", headers=replay.master, json=Body(prompt="say"))
== recorded_binary
)
assert fake.calls == calls_after_record
def test_miss_names_the_computed_key_and_the_closest_recorded_key(self, tmp_path: Path) -> None:
fake = FakeTransport()
root = tmp_path / "bundle"
recording: Transport = RecordingTransport(inner=fake, recorder=make_recorder(root))
recording.post("/model/new", headers=fake.master, json=Body(prompt="x"), response_type=Payload)
replay: Transport = ReplayTransport(source=replay_source(root), master_key="sk-1234")
with pytest.raises(ReplayMiss) as excinfo:
replay.get("/v1/models", headers=replay.master, params=Query(q="all"), response_type=Payload)
message = str(excinfo.value)
assert "no recorded interaction matches key get /v1/models #" in message
assert "closest recorded key is post /model/new #" in message
assert "0000-post-model-new.json" in message
assert "re-record with E2E_FIXTURE_MODE=record" in message
def test_content_drift_on_the_same_route_misses_with_no_live_call(self, tmp_path: Path) -> None:
"""The naive verb+path match replayed a stale response for a request
whose content had changed, silently passing; a content key must miss,
print both canonical forms' diff, and never reach the inner transport."""
fake = FakeTransport()
root = tmp_path / "bundle"
recording: Transport = RecordingTransport(inner=fake, recorder=make_recorder(root))
recording.post("/model/new", headers=fake.master, json=Body(prompt="x"), response_type=Payload)
calls_after_record = list(fake.calls)
replay: Transport = ReplayTransport(source=replay_source(root), master_key="sk-1234")
with pytest.raises(ReplayMiss) as excinfo:
replay.post("/model/new", headers=replay.master, json=Body(prompt="y"), response_type=Payload)
message = str(excinfo.value)
assert "no recorded interaction matches key post /model/new #" in message
assert "closest recorded key is post /model/new #" in message
assert '- "prompt": "x"' in message
assert '+ "prompt": "y"' in message
assert fake.calls == calls_after_record
def test_exhausted_key_names_the_key(self, tmp_path: Path) -> None:
fake = FakeTransport()
root = tmp_path / "bundle"
recording: Transport = RecordingTransport(inner=fake, recorder=make_recorder(root))
recording.post("/model/new", headers=fake.master, json=Body(prompt="x"), response_type=Payload)
replay: Transport = ReplayTransport(source=replay_source(root), master_key="sk-1234")
replay.post("/model/new", headers=replay.master, json=Body(prompt="x"), response_type=Payload)
with pytest.raises(
ReplayMiss, match=r"every recorded interaction for key post /model/new #\w{16} is already consumed"
):
replay.post("/model/new", headers=replay.master, json=Body(prompt="x"), response_type=Payload)
def test_replays_out_of_recorded_order_across_distinct_keys(self, tmp_path: Path) -> None:
"""Concurrent tests interleave independent calls nondeterministically
(e.g. a burst of parallel chat calls), so replay matches by content,
never by recorded position."""
fake = FakeTransport()
root = tmp_path / "bundle"
recording: Transport = RecordingTransport(inner=fake, recorder=make_recorder(root))
recording.post("/model/new", headers=fake.master, json=Body(prompt="x"), response_type=Payload)
recording.post("/key/generate", headers=fake.master, json=Body(prompt="k"), response_type=Payload)
source = replay_source(root)
replay: Transport = ReplayTransport(source=source, master_key="sk-1234")
replay.post("/key/generate", headers=replay.master, json=Body(prompt="k"), response_type=Payload)
replay.post("/model/new", headers=replay.master, json=Body(prompt="x"), response_type=Payload)
assert source.leftover_error(current_test_key()) is None
def test_identical_requests_replay_their_responses_in_recorded_order(self, tmp_path: Path) -> None:
"""A poll loop makes the same request repeatedly and asserts on the
progression, so duplicates under one key stay FIFO."""
root = tmp_path / "bundle"
recorder = make_recorder(root)
recorder.record(
test_key=current_test_key(),
request=recorded_request(
"get", "/v1/models", headers=AuthHeaders(authorization="Bearer sk-x"), params=Query(q="all")
),
response=RecordedResult(kind="success", status_code=200, data={"value": "first"}),
)
recorder.record(
test_key=current_test_key(),
request=recorded_request(
"get", "/v1/models", headers=AuthHeaders(authorization="Bearer sk-x"), params=Query(q="all")
),
response=RecordedResult(kind="success", status_code=200, data={"value": "second"}),
)
replay: Transport = ReplayTransport(source=replay_source(root), master_key="sk-1234")
first = replay.get("/v1/models", headers=replay.master, params=Query(q="all"), response_type=Payload)
second = replay.get("/v1/models", headers=replay.master, params=Query(q="all"), response_type=Payload)
assert first == Success(status_code=200, data=Payload(value="first"))
assert second == Success(status_code=200, data=Payload(value="second"))
def test_concurrent_replays_of_one_key_serve_each_recording_exactly_once(self, tmp_path: Path) -> None:
"""A burst of parallel identical calls consumes one shared pool: no
response duplicated, none forgotten, nothing left over at teardown.
The tiny switch interval forces thread preemption inside pool setup
and consumption, so a non-atomic pool build or pop fails this test."""
root = tmp_path / "bundle"
recorder = make_recorder(root)
for ordinal in range(32):
recorder.record(
test_key=current_test_key(),
request=recorded_request(
"get", "/v1/models", headers=AuthHeaders(authorization="Bearer sk-x"), params=Query(q="all")
),
response=RecordedResult(kind="success", status_code=200, data={"value": f"v{ordinal:02d}"}),
)
source = replay_source(root)
replay: Transport = ReplayTransport(source=source, master_key="sk-1234")
barrier = threading.Barrier(8)
def consume_one() -> str:
result = replay.get(
"/v1/models", headers=replay.master, params=Query(q="all"), response_type=Payload
)
assert isinstance(result, Success)
return result.data.value
def consume(_: int) -> tuple[str, ...]:
barrier.wait()
return tuple(consume_one() for _call in range(4))
previous_interval = sys.getswitchinterval()
sys.setswitchinterval(1e-6)
try:
with ThreadPoolExecutor(max_workers=8) as executor:
served = sorted(value for values in executor.map(consume, range(8)) for value in values)
finally:
sys.setswitchinterval(previous_interval)
assert served == [f"v{ordinal:02d}" for ordinal in range(32)]
assert source.leftover_error(current_test_key()) is None
class TestRecordedKeySets:
def test_two_separate_recordings_of_one_flow_produce_identical_key_sets(
self, tmp_path: Path
) -> None:
"""Everything a run randomizes (markers, virtual keys, dates) must
canonicalize out, so separately recorded runs of the same suite agree
on every match key and a bundle recorded elsewhere replays here."""
def record_flow(root: Path, run_date: str) -> list[str]:
fake = FakeTransport()
recording: Transport = RecordingTransport(inner=fake, recorder=make_recorder(root))
marker = deterministic_marker()
recording.post(
"/model/new",
headers=fake.master,
json=DeployBody(
model_name=f"e2e-chat-{marker}",
litellm_params=DeployParams(model="openai/gpt", api_key=f"sk-live-{uuid4().hex}"),
),
response_type=Payload,
)
recording.post(
"/chat/completions",
headers=recording.bearer(f"sk-{uuid4().hex}"),
json=Body(prompt=f"Reply with the single word ok. {marker}"),
response_type=Payload,
)
recording.get(
"/spend/logs", headers=fake.master, params=Query(q=run_date), response_type=Payload
)
loaded = load_bundle(root)
assert isinstance(loaded, LoadedBundle)
return sorted(
canonicalize(interaction.request).key
for interactions in loaded.interactions.values()
for interaction in interactions
)
first_keys = record_flow(tmp_path / "one", "2026-08-18")
second_keys = record_flow(tmp_path / "two", "2026-08-19")
assert first_keys == second_keys
assert len(first_keys) == 3
class TestReplayLeftover:
def test_fully_consumed_recording_leaves_nothing(self, tmp_path: Path) -> None:
fake = FakeTransport()
root = tmp_path / "bundle"
recording: Transport = RecordingTransport(inner=fake, recorder=make_recorder(root))
recording.post("/model/new", headers=fake.master, json=Body(prompt="x"), response_type=Payload)
source = replay_source(root)
replay: Transport = ReplayTransport(source=source, master_key="sk-1234")
replay.post("/model/new", headers=replay.master, json=Body(prompt="x"), response_type=Payload)
assert source.leftover_error(current_test_key()) is None
def test_unconsumed_trailing_interactions_name_the_next_call(self, tmp_path: Path) -> None:
fake = FakeTransport()
root = tmp_path / "bundle"
recording: Transport = RecordingTransport(inner=fake, recorder=make_recorder(root))
recording.post("/model/new", headers=fake.master, json=Body(prompt="x"), response_type=Payload)
recording.probe("/health/liveliness", params=Query(q="1"))
source = replay_source(root)
replay: Transport = ReplayTransport(source=source, master_key="sk-1234")
replay.post("/model/new", headers=replay.master, json=Body(prompt="x"), response_type=Payload)
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. probe /health/liveliness #" in error
assert "re-record with E2E_FIXTURE_MODE=record" in error
def test_test_without_recordings_has_no_leftover(self, tmp_path: Path) -> None:
root = tmp_path / "bundle"
make_recorder(root)
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
def test_replay_mode_reads_the_shared_bundle(self, tmp_path: Path) -> None:
fake = FakeTransport()
root = tmp_path / "bundle"
recording: Transport = RecordingTransport(inner=fake, recorder=make_recorder(root))
recording.post("/model/new", headers=fake.master, json=Body(prompt="x"), response_type=Payload)
error = replay_leftover_error(mode_raw="replay", bundle_dir=root, test_key=current_test_key())
assert error is not None
assert "1 of 1 recorded interactions never consumed" in error
class TestSelectTransport:
def test_live_returns_the_live_transport_untouched(self, tmp_path: Path) -> None:
fake = FakeTransport()
for mode_raw in ("live", ""):
assert (
select_transport(fake, mode_raw=mode_raw, bundle_dir=tmp_path / "b", master_key="sk")
is fake
)
def test_record_wraps_live_and_starts_a_fresh_bundle(self, tmp_path: Path) -> None:
fake = FakeTransport()
root = tmp_path / "bundle"
write_manifest(root, NOW - timedelta(days=30))
(root / "old-test-slug").mkdir()
(root / "old-test-slug" / "0000-post-old.json").write_text("{}", encoding="utf-8")
selected = select_transport(fake, mode_raw="record", bundle_dir=root, master_key="sk")
assert isinstance(selected, RecordingTransport)
assert selected.inner is fake
assert {entry.name for entry in root.iterdir()} == {MANIFEST_FILENAME}
def test_replay_builds_a_transport_from_the_bundle_alone(self, tmp_path: Path) -> None:
fake = FakeTransport()
root = tmp_path / "bundle"
make_recorder(root)
selected = select_transport(fake, mode_raw="replay", bundle_dir=root, master_key="sk-master")
assert isinstance(selected, ReplayTransport)
assert selected.master == AuthHeaders(authorization="Bearer sk-master")
def test_invalid_mode_raises_naming_the_value(self, tmp_path: Path) -> None:
with pytest.raises(ValueError, match="cached"):
select_transport(
FakeTransport(), mode_raw="cached", bundle_dir=tmp_path / "b", master_key="sk"
)
class TestCollectionGate:
def test_invalid_mode_names_the_value_and_the_choices(self, tmp_path: Path) -> None:
assert (
fixture_mode_collection_error("cached", tmp_path, now=NOW)
== "E2E_FIXTURE_MODE='cached' is not one of live, record, replay"
)
@pytest.mark.parametrize("mode_raw", ["live", "", "record"])
def test_live_and_record_never_block_collection(self, mode_raw: str, tmp_path: Path) -> None:
assert fixture_mode_collection_error(mode_raw, tmp_path / "missing", now=NOW) is None
def test_replay_with_no_bundle_says_how_to_record_one(self, tmp_path: Path) -> None:
reason = fixture_mode_collection_error("replay", tmp_path / "missing", now=NOW)
assert reason is not None
assert f"no {MANIFEST_FILENAME}" in reason
assert "E2E_FIXTURE_MODE=record" in reason
def test_stale_replay_bundle_fails_naming_its_age(self, tmp_path: Path) -> None:
root = tmp_path / "bundle"
write_manifest(root, NOW - timedelta(days=9, hours=5))
reason = fixture_mode_collection_error("replay", root, now=NOW)
assert reason is not None
assert "age 9d5h exceeds the 7-day limit" in reason
assert "re-record with E2E_FIXTURE_MODE=record" in reason
def test_fresh_replay_bundle_collects(self, tmp_path: Path) -> None:
root = tmp_path / "bundle"
write_manifest(root, NOW - timedelta(days=2))
assert fixture_mode_collection_error("replay", root, now=NOW) is None
class TestReportHeader:
def test_live_mode_prints_nothing(self, tmp_path: Path) -> None:
assert fixture_report_lines("live", tmp_path, now=NOW) == []
assert fixture_report_lines("", tmp_path, now=NOW) == []
def test_record_and_replay_name_the_bundle(self, tmp_path: Path) -> None:
root = tmp_path / "bundle"
recorded_at = NOW - timedelta(days=1)
write_manifest(root, recorded_at)
assert fixture_report_lines("record", root, now=NOW) == [
f"e2e fixture mode: record -> {root}"
]
replay_lines = fixture_report_lines("replay", root, now=NOW)
assert len(replay_lines) == 1
assert "replay" in replay_lines[0]
assert recorded_at.isoformat() in replay_lines[0]

492
tests/e2e/test_provider_edge.py Normal file
View file

@ -0,0 +1,492 @@
"""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
import pytest
from pydantic import TypeAdapter
from e2e_http import RawResponse, forward
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,
handle_edge_request,
provider_edge_api_base,
replay_leftover_error,
start_provider_edge,
)
CHAT_PATH = "/openai/v1/chat/completions"
REPLAY_MOUNTS = {"openai": "https://replay.invalid"}
JSON_OBJECT = TypeAdapter(dict[str, object])
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()
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 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()