From 6604c781208941592e4f1a3f67e5262a090b7e63 Mon Sep 17 00:00:00 2001 From: Yuneng Jiang Date: Mon, 14 Sep 2026 16:55:43 -0700 Subject: [PATCH 1/6] test: add strict stateless provider replay identity --- .github/workflows/test-code-quality.yml | 7 + tests/e2e/CONTRIBUTING.md | 10 + tests/e2e/fixture_bundle.py | 73 ++++-- tests/e2e/fixture_canonical.py | 9 +- tests/e2e/fixture_mode.py | 7 +- tests/e2e/fixture_profile.py | 169 +++++++++++++ tests/e2e/provider_edge.py | 72 ++++-- tests/e2e/test_provider_edge.py | 310 +++++++++++++++++++++++- 8 files changed, 607 insertions(+), 50 deletions(-) create mode 100644 tests/e2e/fixture_profile.py diff --git a/.github/workflows/test-code-quality.yml b/.github/workflows/test-code-quality.yml index 9c7e0db7065..ae117a6b637 100644 --- a/.github/workflows/test-code-quality.yml +++ b/.github/workflows/test-code-quality.yml @@ -83,6 +83,13 @@ jobs: - name: test_e2e_changed_gate run: uv run --no-sync pytest -q --noconftest -p no:cacheprovider -c /dev/null tests/code_coverage_tests/test_e2e_changed_gate.py tests/code_coverage_tests/test_e2e_idp_stack.py + - name: test_provider_replay_harness + run: | + pwd + uv run --no-sync pytest -q --noconftest -o addopts= -p no:rerunfailures \ + tests/e2e/test_provider_edge.py tests/e2e/test_fixture_bundle.py \ + tests/e2e/test_fixture_canonical.py tests/e2e/test_fixture_mode.py + - name: router_code_coverage run: uv run --no-sync python ./tests/code_coverage_tests/router_code_coverage.py diff --git a/tests/e2e/CONTRIBUTING.md b/tests/e2e/CONTRIBUTING.md index 44564a51e26..fa177abd64e 100644 --- a/tests/e2e/CONTRIBUTING.md +++ b/tests/e2e/CONTRIBUTING.md @@ -232,3 +232,13 @@ Before you push 4. Capture screenshots of the test run and attach them to the PR as proof 5. If a test fails because it surfaced a real issue in the product, flag that explicitly in the PR rather than reworking the test until it passes + +### Strict stateless replay matching + +Set `E2E_REPLAY_MATCH_PROFILE=stateless_v1` for both recording and replay to bind OpenAI `/v1/chat/completions` and Anthropic `/v1/messages` requests to their upstream destination, ordered query pairs, semantic headers and literal JSON content. The default remains `legacy`. Strict bundles use format 5 and cannot load as legacy bundles; select the matching profile or re-record with `E2E_FIXTURE_MODE=record`. Missing profile metadata never enrolls a legacy bundle in strict matching + +Strict matching preserves dates, UUIDs, hashes, model names, tool arguments, array order and omitted/null/empty/false/zero values. JSON object key order and header name casing may change. The strict body uses tagged JSON values so number precision and JSON types survive persistence, including numbers larger than a floating-point value. Invalid UTF-8 query values fail eligibility. Duplicate JSON keys, unsupported endpoints, non-JSON bodies and unknown semantic headers fail eligibility before contacting a provider + +The semantic header set is `content-type`, `accept`, `anthropic-version`, `anthropic-beta` and `openai-beta`, including missing versus present values. Authorization records presence and scheme; `x-api-key` records presence only. Credential values and cookies are excluded. Credential query values are redacted while their position and field name remain in the identity. Never use real customer inputs in fixture qualification + +Excluded transport and telemetry headers are `host`, `content-length`, `connection`, `accept-encoding`, `user-agent`, `traceparent`, `tracestate`, `x-request-id`, `x-client-request-id` and `x-stainless-*`. Inbound transfer-encoding is unsupported; send JSON with content-length framing. The destination represents host identity and the relay carries original body bytes. Replay does not verify credentials, SDK timeout/retry behavior, transport performance, model availability or stateful remote IDs. Live relay uses original request bytes and header values, never the stored identity diff --git a/tests/e2e/fixture_bundle.py b/tests/e2e/fixture_bundle.py index 7c9dab1a687..4467d7e4ecc 100644 --- a/tests/e2e/fixture_bundle.py +++ b/tests/e2e/fixture_bundle.py @@ -30,9 +30,11 @@ from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Annotated, Final, Literal +from fixture_profile import MatchProfile, StrictIdentity from pydantic import BaseModel, Field, JsonValue BUNDLE_FORMAT_VERSION: Final = 4 +STRICT_BUNDLE_FORMAT_VERSION: Final = 5 MAX_BUNDLE_AGE: Final = timedelta(days=7) MANIFEST_FILENAME: Final = "manifest.json" @@ -41,6 +43,7 @@ class Manifest(BaseModel): format_version: int recorded_at: datetime harness_version: str + match_profile: MatchProfile = "legacy" class RecordedRequest(BaseModel): @@ -69,6 +72,7 @@ class RecordedRequest(BaseModel): file_name: str | None = None file_sha256: str | None = None file_bytes: int | None = None + strict_identity: StrictIdentity | None = None class RecordedHttpResponse(BaseModel): @@ -100,9 +104,7 @@ class RecordedStreamedResponse(BaseModel): truncated: str | None = None -type RecordedResponse = Annotated[ - RecordedHttpResponse | RecordedStreamedResponse, Field(discriminator="kind") -] +type RecordedResponse = Annotated[RecordedHttpResponse | RecordedStreamedResponse, Field(discriminator="kind")] class Interaction(BaseModel): @@ -152,6 +154,7 @@ class BundleRecorder: manifest, so record mode never reads (or merges into) an existing bundle.""" root: Path + profile: MatchProfile = "legacy" _ordinals: dict[str, int] = field(default_factory=dict) def record(self, *, test_key: str, request: RecordedRequest, response: RecordedResponse) -> None: @@ -162,7 +165,12 @@ class BundleRecorder: directory.mkdir(parents=True, exist_ok=True) interaction = Interaction(request=request, response=response) target = directory / interaction_filename(ordinal, request) - target.write_text(interaction.model_dump_json(indent=2), encoding="utf-8") + target.write_text( + interaction.model_dump_json( + indent=2, exclude={"request": {"strict_identity"}} if self.profile == "legacy" else None + ), + encoding="utf-8", + ) @dataclass(frozen=True, slots=True) @@ -171,7 +179,7 @@ class UnsafeBundleDir: reason: str -def prepare_bundle(root: Path) -> BundleRecorder | UnsafeBundleDir: +def prepare_bundle(root: Path, *, profile: MatchProfile = "legacy") -> BundleRecorder | UnsafeBundleDir: """Start a fresh bundle at ``root`` for record mode: wipe whatever bundle is there and write a new manifest. Refuses to wipe a directory that is neither empty nor a bundle (no manifest.json), so a mistyped E2E_FIXTURE_DIR can @@ -188,12 +196,15 @@ def prepare_bundle(root: Path) -> BundleRecorder | UnsafeBundleDir: shutil.rmtree(root) root.mkdir(parents=True) manifest = Manifest( - format_version=BUNDLE_FORMAT_VERSION, + format_version=BUNDLE_FORMAT_VERSION if profile == "legacy" else STRICT_BUNDLE_FORMAT_VERSION, + match_profile=profile, recorded_at=datetime.now(timezone.utc), harness_version=harness_version(), ) - (root / MANIFEST_FILENAME).write_text(manifest.model_dump_json(indent=2), encoding="utf-8") - return BundleRecorder(root=root) + (root / MANIFEST_FILENAME).write_text( + manifest.model_dump_json(indent=2, exclude={"match_profile"} if profile == "legacy" else None), encoding="utf-8" + ) + return BundleRecorder(root=root, profile=profile) @dataclass(frozen=True, slots=True) @@ -226,25 +237,30 @@ def _read_manifest(root: Path) -> Manifest | UnreadableBundle: return UnreadableBundle(reason=f"{MANIFEST_FILENAME} is invalid: {exc}") -def _supported_manifest(root: Path) -> Manifest | UnreadableBundle: +def _supported_manifest(root: Path, profile: MatchProfile = "legacy") -> Manifest | UnreadableBundle: """The manifest, refused when it was written under a different format version. A bundle is atomic (record wipes and rewrites the whole directory and never merges), so a foreign version is a hard reject rather than a partial read.""" manifest = _read_manifest(root) if isinstance(manifest, UnreadableBundle): return manifest - if manifest.format_version != BUNDLE_FORMAT_VERSION: + expected_version: Final = BUNDLE_FORMAT_VERSION if profile == "legacy" else STRICT_BUNDLE_FORMAT_VERSION + if manifest.match_profile != profile: + return UnreadableBundle( + reason="match profile mismatch; select the recorded E2E_REPLAY_MATCH_PROFILE or re-record" + ) + if manifest.format_version != expected_version: return UnreadableBundle( reason=( - f"format_version {manifest.format_version} != supported {BUNDLE_FORMAT_VERSION}; " + f"format_version {manifest.format_version} != supported {expected_version}; " "re-record with E2E_FIXTURE_MODE=record" ) ) return manifest -def check_freshness(root: Path, *, now: datetime) -> BundleFreshness: - manifest = _supported_manifest(root) +def check_freshness(root: Path, *, now: datetime, profile: MatchProfile = "legacy") -> BundleFreshness: + manifest = _supported_manifest(root, profile) if isinstance(manifest, UnreadableBundle): return manifest recorded_at = ( @@ -269,16 +285,27 @@ class LoadedBundle: interactions: dict[str, tuple[Interaction, ...]] -def load_bundle(root: Path) -> LoadedBundle | UnreadableBundle: - manifest = _supported_manifest(root) +def load_bundle(root: Path, *, profile: MatchProfile = "legacy") -> LoadedBundle | UnreadableBundle: + manifest = _supported_manifest(root, profile) if isinstance(manifest, UnreadableBundle): return manifest - interactions = { - directory.name: tuple( - Interaction.model_validate_json(file.read_text(encoding="utf-8")) - for file in sorted(directory.glob("*.json")) - ) - for directory in sorted(root.iterdir()) - if directory.is_dir() - } + try: + interactions = { + directory.name: tuple( + Interaction.model_validate_json(file.read_text(encoding="utf-8")) + for file in sorted(directory.glob("*.json")) + ) + for directory in sorted(root.iterdir()) + if directory.is_dir() + } + except (ValueError, OSError): + if profile == "legacy": + raise + return UnreadableBundle(reason="invalid stateless_v1 interaction; re-record with the selected profile") + if any( + (item.request.strict_identity is not None) != (profile == "stateless_v1") + for items in interactions.values() + for item in items + ): + return UnreadableBundle(reason="request identity/profile mismatch; re-record with the selected profile") return LoadedBundle(manifest=manifest, interactions=interactions) diff --git a/tests/e2e/fixture_canonical.py b/tests/e2e/fixture_canonical.py index c043951a108..e76d63ca33b 100644 --- a/tests/e2e/fixture_canonical.py +++ b/tests/e2e/fixture_canonical.py @@ -23,9 +23,8 @@ from dataclasses import dataclass from functools import reduce from typing import Final -from pydantic import JsonValue - from fixture_bundle import RecordedRequest +from pydantic import JsonValue VOLATILE_HEADER_NAMES: Final[frozenset[str]] = frozenset( { @@ -123,6 +122,12 @@ class CanonicalRequest: def canonicalize(request: RecordedRequest) -> CanonicalRequest: + if request.strict_identity is not None: + return CanonicalRequest( + method=request.method, + path=request.path, + content=json.dumps(request.strict_identity.model_dump(mode="json"), sort_keys=True, separators=(",", ":")), + ) file_identity: Final[JsonValue | None] = ( None if request.file_name is None and request.file_sha256 is None diff --git a/tests/e2e/fixture_mode.py b/tests/e2e/fixture_mode.py index 110f44380b4..9a7c1b6db12 100644 --- a/tests/e2e/fixture_mode.py +++ b/tests/e2e/fixture_mode.py @@ -26,6 +26,7 @@ from fixture_bundle import ( check_freshness, format_age, ) +from fixture_profile import match_profile type FixtureMode = Literal["live", "record", "replay"] @@ -82,6 +83,7 @@ def fixture_mode_collection_error(mode_raw: str, bundle_dir: Path, *, now: datet 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.""" + match_profile() mode = parse_fixture_mode(mode_raw) match mode: case InvalidFixtureMode(value=value): @@ -89,7 +91,7 @@ def fixture_mode_collection_error(mode_raw: str, bundle_dir: Path, *, now: datet case "live" | "record": return None case "replay": - freshness = check_freshness(bundle_dir, now=now) + freshness = check_freshness(bundle_dir, now=now, profile=match_profile()) match freshness: case FreshBundle(): return None @@ -110,6 +112,7 @@ def fixture_mode_collection_error(mode_raw: str, bundle_dir: Path, *, now: datet 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.""" + match_profile() mode = parse_fixture_mode(mode_raw) match mode: case InvalidFixtureMode() | "live": @@ -117,7 +120,7 @@ def fixture_report_lines(mode_raw: str, bundle_dir: Path, *, now: datetime) -> l case "record": return [f"e2e fixture mode: record -> {bundle_dir}"] case "replay": - freshness = check_freshness(bundle_dir, now=now) + freshness = check_freshness(bundle_dir, now=now, profile=match_profile()) match freshness: case FreshBundle(manifest=manifest): return [ diff --git a/tests/e2e/fixture_profile.py b/tests/e2e/fixture_profile.py new file mode 100644 index 00000000000..ad998f80ac2 --- /dev/null +++ b/tests/e2e/fixture_profile.py @@ -0,0 +1,169 @@ +from __future__ import annotations + +import json +import os +from collections.abc import Mapping +from dataclasses import dataclass +from decimal import Decimal, DecimalException +from typing import Final, Literal +from urllib.parse import parse_qsl, urlsplit + +from pydantic import BaseModel, JsonValue, TypeAdapter + +type MatchProfile = Literal["legacy", "stateless_v1"] +type ExactJson = dict[str, ExactJson] | list[ExactJson] | str | bool | Decimal | int | None + +SEMANTIC_HEADERS: Final = frozenset({"content-type", "accept", "anthropic-version", "anthropic-beta", "openai-beta"}) +AUTH_HEADERS: Final = frozenset({"authorization", "x-api-key"}) +EXCLUDED_HEADERS: Final = frozenset( + { + "host", + "content-length", + "transfer-encoding", + "connection", + "accept-encoding", + "user-agent", + "traceparent", + "tracestate", + "x-request-id", + "x-client-request-id", + "cookie", + } +) +CREDENTIAL_QUERY: Final = frozenset( + { + "api_key", + "api-key", + "apikey", + "key", + "token", + "access_token", + "signature", + "password", + "secret", + "credentials", + "authorization", + "sig", + "client_secret", + "aws_access_key_id", + "aws_secret_access_key", + "aws_session_token", + } +) +JSON_VALUE: Final[TypeAdapter[ExactJson]] = TypeAdapter(ExactJson) + + +def match_profile() -> MatchProfile: + raw: Final = os.environ.get("E2E_REPLAY_MATCH_PROFILE", "legacy") + if raw in ("legacy", "stateless_v1"): + return raw + raise ValueError("E2E_REPLAY_MATCH_PROFILE must be legacy or stateless_v1") + + +class StrictIdentity(BaseModel): + upstream: str + mount: str + query: tuple[tuple[str, str], ...] + headers: dict[str, str] + auth: dict[str, str] + body_present: bool + body: JsonValue + + +@dataclass(frozen=True, slots=True) +class IneligibleRequest: + reason: str + + +def _unique_object(pairs: list[tuple[str, ExactJson]]) -> dict[str, ExactJson]: + if len({key for key, _ in pairs}) != len(pairs): + raise ValueError("duplicate JSON object keys") + return dict(pairs) + + +def _invalid_constant(value: str) -> ExactJson: + raise ValueError("nonfinite JSON number") + + +def _exact_value(value: ExactJson) -> JsonValue: + match value: + case dict(): + return {"object": {key: _exact_value(item) for key, item in value.items()}} + case list(): + return {"array": [_exact_value(item) for item in value]} + case bool(): + return {"boolean": value} + case int() | Decimal(): + return {"number": str(value)} + case str(): + return {"string": value} + case None: + return None + + +def strict_identity( + *, + method: str, + path: str, + query: str, + headers: Mapping[str, str], + body: bytes | None, + mount: str, + upstream_base: str, +) -> StrictIdentity | IneligibleRequest: + if (mount, path, method.upper()) not in { + ("openai", "/openai/v1/chat/completions", "POST"), + ("anthropic", "/anthropic/v1/messages", "POST"), + }: + return IneligibleRequest("unsupported endpoint or method") + lowered: Final = {key.lower(): value for key, value in headers.items()} + if len(lowered) != len(headers): + return IneligibleRequest("duplicate header names") + if any( + key not in SEMANTIC_HEADERS | AUTH_HEADERS | EXCLUDED_HEADERS and not key.startswith("x-stainless-") + for key in lowered + ): + return IneligibleRequest("unsupported semantic header") + if "transfer-encoding" in lowered: + return IneligibleRequest("unsupported request transfer-encoding; send a content-length framed JSON body") + authorization: Final = lowered.get("authorization") + if authorization is not None and authorization.partition(" ")[0].lower() not in {"bearer", "basic", "digest"}: + return IneligibleRequest("unsupported authorization scheme") + destination: Final = urlsplit(upstream_base) + if destination.username or destination.password or destination.query or destination.fragment: + return IneligibleRequest("upstream destination contains credentials, query or fragment") + if destination.scheme not in ("http", "https") or not destination.netloc: + return IneligibleRequest("unsupported upstream destination") + if body and lowered.get("content-type", "").split(";", 1)[0].strip().lower() != "application/json": + return IneligibleRequest("unsupported body content-type; stateless_v1 requires JSON") + try: + parsed: Final = ( + JSON_VALUE.validate_python( + json.loads( + body, object_pairs_hook=_unique_object, parse_constant=_invalid_constant, parse_float=Decimal + ) + ) + if body + else None + ) + except (ValueError, UnicodeError, DecimalException): + return IneligibleRequest("invalid JSON or duplicate JSON object keys") + if body and not isinstance(parsed, dict): + return IneligibleRequest("stateless inference requires a JSON object") + try: + query_pairs: Final = tuple(parse_qsl(query, keep_blank_values=True, errors="strict")) + except UnicodeError: + return IneligibleRequest("invalid UTF-8 query encoding") + return StrictIdentity( + upstream=upstream_base, + mount=mount, + query=tuple((key, "" if key.lower() in CREDENTIAL_QUERY else value) for key, value in query_pairs), + headers={key: value for key, value in lowered.items() if key in SEMANTIC_HEADERS}, + auth={ + key: (value.partition(" ")[0] if key == "authorization" else "present") + for key, value in lowered.items() + if key in AUTH_HEADERS + }, + body_present=bool(body), + body=_exact_value(parsed), + ) diff --git a/tests/e2e/provider_edge.py b/tests/e2e/provider_edge.py index 6c87c7ef7ac..de36895ebb6 100644 --- a/tests/e2e/provider_edge.py +++ b/tests/e2e/provider_edge.py @@ -92,6 +92,7 @@ from fixture_mode import ( current_test_key, parse_fixture_mode, ) +from fixture_profile import IneligibleRequest, MatchProfile, match_profile, strict_identity from pydantic import JsonValue, TypeAdapter EDGE_MOUNTS: Final[Mapping[str, str]] = MappingProxyType( @@ -404,6 +405,12 @@ def _miss_message(test_key: str, slug: str, canonical: CanonicalRequest, bundle: f"under {slug}; re-record with E2E_FIXTURE_MODE=record" ) closest, closest_file = _closest_recorded(canonical, recorded) + if bundle.manifest.match_profile == "stateless_v1": + expected: Final = _JSON.validate_json(closest.content) + actual: Final = _JSON.validate_json(canonical.content) + assert isinstance(expected, dict) and isinstance(actual, dict) + changed: Final = ", ".join(key for key in expected if expected[key] != actual.get(key)) + return f"stateless_v1 replay mismatch: {changed or 'method/path'}; re-record with E2E_FIXTURE_MODE=record" diff: Final = "\n".join( islice( difflib.unified_diff( @@ -785,11 +792,33 @@ def handle_edge_request( 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))}" + return _text_reply(404, f"unknown provider mount {mount!r}; known mounts: {', '.join(sorted(mounts))}") + profile: Final = ( + backend.recorder.profile + if isinstance(backend, RecordEdge) + else backend.source.bundle.manifest.match_profile + if isinstance(backend, ReplayEdge) + else "legacy" + ) + identity: Final = ( + strict_identity( + method=method, + path=split.path, + query=split.query, + headers=headers, + body=body, + mount=mount, + upstream_base=upstream_base, ) - request: Final = edge_request( - method, split.path, split.query, body, _header_value(headers, "content-type") + if profile == "stateless_v1" + else None + ) + if isinstance(identity, IneligibleRequest): + return _text_reply(REPLAY_MISS_STATUS, f"stateless_v1 eligibility error: {identity.reason}") + request: Final = ( + RecordedRequest(method=method.lower(), path=split.path, headers={}, strict_identity=identity) + if identity is not None + else edge_request(method, split.path, split.query, body, _header_value(headers, "content-type")) ) match backend: case LiveEdge(): @@ -837,6 +866,14 @@ class _EdgeHandler(BaseHTTPRequestHandler): body: Final = self.rfile.read(length) if length else None if edge_server.observation is not None: edge_server.observation.observe(body) + strict: Final = ( + isinstance(edge_server.backend, RecordEdge) and edge_server.backend.recorder.profile == "stateless_v1" + or isinstance(edge_server.backend, ReplayEdge) + and edge_server.backend.source.bundle.manifest.match_profile == "stateless_v1" + ) + if strict and len({name.lower() for name in self.headers.keys()}) != len(self.headers): + self._write_reply(_text_reply(REPLAY_MISS_STATUS, "stateless_v1 eligibility error: duplicate headers")) + return outcome: Final = handle_edge_request( edge_server.backend, edge_server.mounts, @@ -955,16 +992,16 @@ def start_provider_edge( @functools.lru_cache(maxsize=8) -def _shared_recorder(root: Path) -> BundleRecorder: - prepared = prepare_bundle(root) +def _shared_recorder(root: Path, profile: MatchProfile = "legacy") -> BundleRecorder: + prepared = prepare_bundle(root, profile=profile) 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) +def _shared_replay_source(root: Path, profile: MatchProfile = "legacy") -> ReplaySource: + loaded = load_bundle(root, profile=profile) if isinstance(loaded, UnreadableBundle): raise ValueError(f"cannot replay from {root}: {loaded.reason}") return ReplaySource(bundle=loaded) @@ -977,11 +1014,12 @@ def _shared_edge( bind_host: str, advertise_host: str, forward_timeout: float, + profile: MatchProfile, ) -> ProviderEdge: backend: Final[EdgeBackend] = ( - RecordEdge(recorder=_shared_recorder(bundle_dir), lock=threading.Lock()) + RecordEdge(recorder=_shared_recorder(bundle_dir, profile), lock=threading.Lock()) if mode == "record" - else ReplayEdge(source=_shared_replay_source(bundle_dir)) + else ReplayEdge(source=_shared_replay_source(bundle_dir, profile)) ) return start_provider_edge( backend, @@ -998,7 +1036,7 @@ def replay_leftover_error(*, mode_raw: str, bundle_dir: Path, test_key: str) -> 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) + return _shared_replay_source(bundle_dir, match_profile()).leftover_error(test_key) def provider_edge_api_base( @@ -1021,10 +1059,10 @@ def provider_edge_api_base( 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) + 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, match_profile()).api_base( + mount + ) case _: assert_never(mode) @@ -1037,9 +1075,9 @@ def _observed_backend(mode_raw: str, bundle_dir: Path) -> EdgeBackend: case "live": return LiveEdge() case "record": - return RecordEdge(_shared_recorder(bundle_dir), threading.Lock()) + return RecordEdge(_shared_recorder(bundle_dir, match_profile()), threading.Lock()) case "replay": - return ReplayEdge(_shared_replay_source(bundle_dir)) + return ReplayEdge(_shared_replay_source(bundle_dir, match_profile())) case _: assert_never(mode) diff --git a/tests/e2e/test_provider_edge.py b/tests/e2e/test_provider_edge.py index 18f72ac0e7a..2d84e143517 100644 --- a/tests/e2e/test_provider_edge.py +++ b/tests/e2e/test_provider_edge.py @@ -24,6 +24,9 @@ from __future__ import annotations import base64 import json +import os +import subprocess +import sys import socket import threading from collections.abc import Generator, Mapping @@ -47,6 +50,7 @@ from fixture_bundle import ( slug_for_test, ) from fixture_canonical import canonicalize +from fixture_profile import MatchProfile from fixture_mode import current_test_key from provider_edge import ( REPLAY_MISS_STATUS, @@ -81,9 +85,11 @@ def json_object(body: bytes) -> dict[str, object]: class _FakeProvider(ThreadingHTTPServer): daemon_threads = True - def __init__(self, bind: tuple[str, int]) -> None: + def __init__(self, bind: tuple[str, int], *, echo_request: bool = True) -> None: super().__init__(bind, _FakeProviderHandler) self.hits: list[str] = [] + self.echo_request = echo_request + self.requests: list[tuple[dict[str, str], bytes]] = [] class _FakeProviderHandler(BaseHTTPRequestHandler): @@ -101,9 +107,8 @@ class _FakeProviderHandler(BaseHTTPRequestHandler): 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() + provider.requests.append((dict(self.headers.items()), body)) + payload = json.dumps({"echo": body.decode("utf-8"), "path": self.path, "hit": len(provider.hits)} if provider.echo_request else {"ok": True}).encode() self.send_response(200) self.send_header("content-type", "application/json") self.send_header("content-length", str(len(payload))) @@ -117,8 +122,8 @@ class _FakeProviderHandler(BaseHTTPRequestHandler): @contextmanager -def fake_provider() -> Generator[_FakeProvider]: - server = _FakeProvider(("127.0.0.1", 0)) +def fake_provider(*, echo_request: bool = True) -> Generator[_FakeProvider]: + server = _FakeProvider(("127.0.0.1", 0), echo_request=echo_request) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() try: @@ -1350,3 +1355,296 @@ class TestProviderRequestObservation: response: Final = call_edge(edge, "POST", CHAT_PATH, body=chat_body("observed-lantern")) assert response.status_code == 502 assert observation.count == 1 + + +class TestStrictIdentity: + @pytest.mark.parametrize("path", [CHAT_PATH, "/anthropic/v1/messages"]) + def test_roundtrip_rejects_semantic_changes(self, tmp_path: Path, path: str) -> None: + recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1") + assert isinstance(recorder, BundleRecorder) + original: Final = ( + b'{"model":"synthetic","messages":[{"role":"user",' + b'"content":"2031-04-05 00000000-0000-0000-0000-000000000001"}],"options":[1,2]}' + ) + headers: Final = { + "content-type": "application/json", + "accept": "application/json", + "anthropic-version": "2023-06-01", + "anthropic-beta": "feature-a", + "openai-beta": "feature-b", + "authorization": "Bearer synthetic-secret-one", + } + query: Final = "?part=one&part=two&blank=" + with fake_provider() as provider: + mounts: Final = {"openai": provider_url(provider), "anthropic": provider_url(provider)} + with running_edge(RecordEdge(recorder, threading.Lock()), mounts) as edge: + captured: Final = call_edge(edge, "POST", path + query, body=original, headers=headers) + assert captured.status_code == 200 + assert json_object(captured.body)["echo"] == original.decode() + loaded: Final = load_bundle(recorder.root, profile="stateless_v1") + assert isinstance(loaded, LoadedBundle) + assert loaded.manifest.match_profile == "stateless_v1" + source: Final = ReplaySource(loaded) + with running_edge(ReplayEdge(source), mounts) as edge: + cases: Final = ( + (original.replace(b"2031-04-05", b"2032-06-07"), headers, query, "body"), + (original.replace(b"000000000001", b"000000000002"), headers, query, "body"), + (original.replace(b"[1,2]", b"[2,1]"), headers, query, "body"), + (original.replace(b"synthetic", b"other"), headers, query, "body"), + (original, headers, "?part=three&part=two&blank=", "query"), + (original, headers, "?part=two&part=one&blank=", "query"), + *( + (original, {k: v for k, v in headers.items() if k != name}, query, "headers") + for name in ("accept", "anthropic-version", "anthropic-beta", "openai-beta") + ), + *( + (original, {**headers, name: value}, query, "headers") + for name in ("accept", "anthropic-version", "anthropic-beta", "openai-beta") + for value in ("different", "") + ), + (original, {**headers, "authorization": "Basic synthetic-secret-two"}, query, "auth"), + (original, {k: v for k, v in headers.items() if k != "authorization"}, query, "auth"), + ) + for rejected, reason in ( + (call_edge(edge, "POST", path + changed_query, body=body, headers=changed_headers), reason) + for body, changed_headers, changed_query, reason in cases + ): + assert rejected.status_code == REPLAY_MISS_STATUS + assert reason in rejected.body.decode() + assert b"synthetic-secret" not in rejected.body + reordered: Final = json.dumps(dict(reversed(list(json_object(original).items())))).encode() + accepted: Final = call_edge( + edge, "POST", path + query, body=reordered, headers={k.upper(): v for k, v in headers.items()} + ) + assert accepted.status_code == 200 + assert accepted.body == captured.body + assert source.leftover_error(current_test_key()) is None + assert len(provider.hits) == 1 + assert "synthetic-secret" not in "".join(file.read_text() for file in recorder.root.rglob("*.json")) + + @pytest.mark.parametrize( + "body", + [ + b'{"value":null}', + b'{"value":""}', + b'{"value":false}', + b'{"value":0}', + b'{"value":[]}', + b'{"value":{}}', + b'{"value":0.123456789012345678901}', + b'{"value":0.123456789012345678902}', + b'{"value":1e400}', + ], + ) + def test_json_values_remain_distinct(self, tmp_path: Path, body: bytes) -> None: + recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1") + assert isinstance(recorder, BundleRecorder) + with fake_provider() as provider: + mounts: Final = {"openai": provider_url(provider)} + with running_edge(RecordEdge(recorder, threading.Lock()), mounts) as edge: + assert ( + call_edge( + edge, "POST", CHAT_PATH, body=body, headers={"content-type": "application/json"} + ).status_code + == 200 + ) + loaded: Final = load_bundle(recorder.root, profile="stateless_v1") + assert isinstance(loaded, LoadedBundle) + with running_edge(ReplayEdge(ReplaySource(loaded)), mounts) as edge: + values: Final = ( + b"{}", + b'{"value":null}', + b'{"value":""}', + b'{"value":false}', + b'{"value":0}', + b'{"value":[]}', + b'{"value":{}}', + b'{"value":0.123456789012345678901}', + b'{"value":0.123456789012345678902}', + b'{"value":1e400}', + ) + for rejected in ( + call_edge(edge, "POST", CHAT_PATH, body=value, headers={"content-type": "application/json"}) + for value in values + if value != body + ): + assert rejected.status_code == REPLAY_MISS_STATUS + assert b"body" in rejected.body + assert ( + call_edge( + edge, "POST", CHAT_PATH, body=body, headers={"content-type": "application/json"} + ).status_code + == 200 + ) + assert len(provider.hits) == 1 + + @pytest.mark.parametrize( + "path,body,headers", + [ + (UPLOAD_PATH, b"{}", {"content-type": "application/json"}), + (CHAT_PATH + "?part=%FF", b"{}", {"content-type": "application/json"}), + (CHAT_PATH + "?part=%FE", b"{}", {"content-type": "application/json"}), + (CHAT_PATH, b"opaque", {"content-type": "application/octet-stream"}), + (CHAT_PATH, b"--boundary", {"content-type": "multipart/form-data; boundary=boundary"}), + (CHAT_PATH, b'{"x":1,"x":2}', {"content-type": "application/json"}), + (CHAT_PATH, b'{"x":1e9999999999999999999}', {"content-type": "application/json"}), + (CHAT_PATH, b"{}", {"content-type": "application/json", "x-custom-behavior": "synthetic-private-value"}), + ], + ) + def test_ineligible_capture_never_calls_provider( + self, tmp_path: Path, path: str, body: bytes, headers: dict[str, str] + ) -> None: + recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1") + assert isinstance(recorder, BundleRecorder) + with fake_provider() as provider: + with running_edge(RecordEdge(recorder, threading.Lock()), {"openai": provider_url(provider)}) as edge: + result: Final = call_edge(edge, "POST", path, body=body, headers=headers) + assert result.status_code == REPLAY_MISS_STATUS + assert b"eligibility error" in result.body + assert b"synthetic-private-value" not in result.body + assert provider.hits == [] + assert this_tests_files(recorder.root) == [] + + def test_destination_is_part_of_actual_http_identity(self, tmp_path: Path) -> None: + recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1") + assert isinstance(recorder, BundleRecorder) + with fake_provider() as provider: + with running_edge(RecordEdge(recorder, threading.Lock()), {"openai": provider_url(provider)}) as edge: + assert ( + call_edge( + edge, "POST", CHAT_PATH, body=b"{}", headers={"content-type": "application/json"} + ).status_code + == 200 + ) + loaded: Final = load_bundle(recorder.root, profile="stateless_v1") + assert isinstance(loaded, LoadedBundle) + with running_edge(ReplayEdge(ReplaySource(loaded)), {"openai": provider_url(provider) + "/other"}) as edge: + result: Final = call_edge( + edge, "POST", CHAT_PATH, body=b"{}", headers={"content-type": "application/json"} + ) + assert result.status_code == REPLAY_MISS_STATUS + assert b"upstream" in result.body + assert len(provider.hits) == 1 + + def test_credentials_are_not_identity_and_fresh_process_replays(self, tmp_path: Path) -> None: + recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1") + assert isinstance(recorder, BundleRecorder) + headers: Final = { + "content-type": "application/json", + "authorization": "Bearer synthetic-token", + "x-api-key": "synthetic-api-key", + "cookie": "synthetic-cookie", + } + path: Final = CHAT_PATH + "?api_key=synthetic-query-secret&part=one&part=two" + body: Final = b'{"model":"synthetic","messages":[]}' + with fake_provider(echo_request=False) as provider: + mounts: Final = {"openai": provider_url(provider)} + with running_edge(RecordEdge(recorder, threading.Lock()), mounts) as edge: + captured: Final = call_edge(edge, "POST", path, body=body, headers=headers) + assert captured.status_code == 200 + seen_headers, seen_body = provider.requests[0] + assert {k.lower(): v for k, v in seen_headers.items()}.items() >= headers.items() + assert seen_body == body + assert provider.hits == ["POST " + path.removeprefix("/openai")] + artifacts: Final = "".join(file.read_text() for file in recorder.root.rglob("*.json")) + for secret in ("synthetic-token", "synthetic-api-key", "synthetic-cookie", "synthetic-query-secret"): + assert secret not in artifacts + child: Final = subprocess.run( + [ + sys.executable, + "-c", + """ +import json, sys +from pathlib import Path +from fixture_bundle import LoadedBundle, load_bundle +from provider_edge import ProviderRequestObservation, observed_provider_edge, replay_leftover_error +from test_provider_edge import call_edge +from fixture_profile import MatchProfile +from fixture_mode import current_test_key +loaded = load_bundle(Path(sys.argv[1]), profile="stateless_v1") +assert isinstance(loaded, LoadedBundle) +with observed_provider_edge(ProviderRequestObservation("synthetic"), mode_raw="replay", bundle_dir=Path(sys.argv[1]), bind_host="127.0.0.1", advertise_host="127.0.0.1", mounts={"openai": sys.argv[2]}) as edge: + response = call_edge(edge, "POST", sys.argv[3], body=sys.argv[4].encode(), headers=json.loads(sys.argv[5])) + assert response.status_code == 200 + print(response.body.decode()) +assert replay_leftover_error(mode_raw="replay", bundle_dir=Path(sys.argv[1]), test_key=current_test_key()) is None +""", + str(recorder.root), + provider_url(provider), + path.replace("synthetic-query-secret", "new-query-credential"), + body.decode(), + json.dumps({**headers, "authorization": "Bearer another-credential", "x-api-key": "another-key"}), + ], + env={ + **os.environ, + "PYTHONPATH": str(Path(__file__).parent), + "E2E_REPLAY_MATCH_PROFILE": "stateless_v1", + }, + capture_output=True, + text=True, + timeout=30, + ) + assert child.returncode == 0, child.stderr + assert child.stdout.strip().encode() == captured.body + assert len(provider.hits) == 1 + + @pytest.mark.parametrize("profile,other", [("legacy", "stateless_v1"), ("stateless_v1", "legacy")]) + def test_profiles_cannot_load_each_others_bundles( + self, tmp_path: Path, profile: MatchProfile, other: MatchProfile + ) -> None: + from fixture_bundle import UnreadableBundle + + recorder: Final = prepare_bundle(tmp_path / profile, profile=profile) + assert isinstance(recorder, BundleRecorder) + mismatch: Final = load_bundle(recorder.root, profile=other) + assert isinstance(mismatch, UnreadableBundle) + assert "profile mismatch" in mismatch.reason + assert "re-record" in mismatch.reason + + @pytest.mark.parametrize("abort_after", [None, 2]) + def test_strict_stream_preserves_chunks_and_truncation(self, tmp_path: Path, abort_after: int | None) -> None: + recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1") + assert isinstance(recorder, BundleRecorder) + with chunked_provider(abort_after=abort_after) as provider: + mounts: Final = {"anthropic": provider_url(provider)} + with running_edge(RecordEdge(recorder, threading.Lock()), mounts) as edge: + _, captured, captured_ending = raw_stream_post(edge.port, "/anthropic/v1/messages", STREAM_BODY) + loaded: Final = load_bundle(recorder.root, profile="stateless_v1") + assert isinstance(loaded, LoadedBundle) + source: Final = ReplaySource(loaded) + with running_edge(ReplayEdge(source), mounts) as edge: + _, replayed, ending = raw_stream_post(edge.port, "/anthropic/v1/messages", STREAM_BODY) + assert captured == replayed == list(SSE_CHUNKS[:abort_after]) + assert ending == captured_ending + assert (ending == "terminated") == (abort_after is None) + assert source.leftover_error(current_test_key()) is None + assert len(provider.hits) == 1 + + def test_auth_scheme_survives_missing_credentials(self, tmp_path: Path) -> None: + recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1") + assert isinstance(recorder, BundleRecorder) + headers: Final = {"content-type": "application/json", "authorization": "Bearer"} + with fake_provider() as provider: + mounts: Final = {"openai": provider_url(provider)} + with running_edge(RecordEdge(recorder, threading.Lock()), mounts) as edge: + assert call_edge(edge, "POST", CHAT_PATH, body=b"{}", headers=headers).status_code == 200 + loaded: Final = load_bundle(recorder.root, profile="stateless_v1") + assert isinstance(loaded, LoadedBundle) + with running_edge(ReplayEdge(ReplaySource(loaded)), mounts) as edge: + for result in ( + call_edge(edge, "POST", CHAT_PATH, body=b"{}", headers={**headers, "authorization": scheme}) + for scheme in ("Basic", "Digest") + ): + assert result.status_code == REPLAY_MISS_STATUS + assert b"auth" in result.body + assert ( + call_edge( + edge, + "POST", + CHAT_PATH, + body=b"{}", + headers={**headers, "authorization": "Bearer synthetic-token"}, + ).status_code + == 200 + ) + assert len(provider.hits) == 1 From b9c194076ab0b537b31b2ae26482fffcd44897a9 Mon Sep 17 00:00:00 2001 From: Yuneng Jiang Date: Mon, 14 Sep 2026 17:11:56 -0700 Subject: [PATCH 2/6] test: preserve strict replay numeric spelling --- tests/e2e/CONTRIBUTING.md | 4 ++-- tests/e2e/fixture_profile.py | 24 +++++++++++++++++------- tests/e2e/test_provider_edge.py | 13 ++++++++++--- 3 files changed, 29 insertions(+), 12 deletions(-) diff --git a/tests/e2e/CONTRIBUTING.md b/tests/e2e/CONTRIBUTING.md index fa177abd64e..313085b2eed 100644 --- a/tests/e2e/CONTRIBUTING.md +++ b/tests/e2e/CONTRIBUTING.md @@ -237,8 +237,8 @@ Before you push Set `E2E_REPLAY_MATCH_PROFILE=stateless_v1` for both recording and replay to bind OpenAI `/v1/chat/completions` and Anthropic `/v1/messages` requests to their upstream destination, ordered query pairs, semantic headers and literal JSON content. The default remains `legacy`. Strict bundles use format 5 and cannot load as legacy bundles; select the matching profile or re-record with `E2E_FIXTURE_MODE=record`. Missing profile metadata never enrolls a legacy bundle in strict matching -Strict matching preserves dates, UUIDs, hashes, model names, tool arguments, array order and omitted/null/empty/false/zero values. JSON object key order and header name casing may change. The strict body uses tagged JSON values so number precision and JSON types survive persistence, including numbers larger than a floating-point value. Invalid UTF-8 query values fail eligibility. Duplicate JSON keys, unsupported endpoints, non-JSON bodies and unknown semantic headers fail eligibility before contacting a provider +Strict matching preserves dates, UUIDs, hashes, model names, tool arguments, array order and omitted/null/empty/false/zero values. JSON object key order and header name casing may change. The strict body uses tagged JSON values so number precision and JSON types survive persistence, including exact numeric spelling and numbers larger than a floating-point value. Invalid UTF-8 query values fail eligibility. Duplicate JSON keys, unsupported endpoints, non-JSON bodies and unknown semantic headers fail eligibility before contacting a provider -The semantic header set is `content-type`, `accept`, `anthropic-version`, `anthropic-beta` and `openai-beta`, including missing versus present values. Authorization records presence and scheme; `x-api-key` records presence only. Credential values and cookies are excluded. Credential query values are redacted while their position and field name remain in the identity. Never use real customer inputs in fixture qualification +The semantic header set is `content-type`, `accept`, `anthropic-version`, `anthropic-beta` and `openai-beta`, including missing versus present values. Authorization records presence and the case-insensitive scheme; `x-api-key` records presence only. Credential values and cookies are excluded. Credential query values are redacted while their position and field name remain in the identity. Never use real customer inputs in fixture qualification Excluded transport and telemetry headers are `host`, `content-length`, `connection`, `accept-encoding`, `user-agent`, `traceparent`, `tracestate`, `x-request-id`, `x-client-request-id` and `x-stainless-*`. Inbound transfer-encoding is unsupported; send JSON with content-length framing. The destination represents host identity and the relay carries original body bytes. Replay does not verify credentials, SDK timeout/retry behavior, transport performance, model availability or stateful remote IDs. Live relay uses original request bytes and header values, never the stored identity diff --git a/tests/e2e/fixture_profile.py b/tests/e2e/fixture_profile.py index ad998f80ac2..f8405be746b 100644 --- a/tests/e2e/fixture_profile.py +++ b/tests/e2e/fixture_profile.py @@ -4,14 +4,20 @@ import json import os from collections.abc import Mapping from dataclasses import dataclass -from decimal import Decimal, DecimalException from typing import Final, Literal from urllib.parse import parse_qsl, urlsplit from pydantic import BaseModel, JsonValue, TypeAdapter type MatchProfile = Literal["legacy", "stateless_v1"] -type ExactJson = dict[str, ExactJson] | list[ExactJson] | str | bool | Decimal | int | None + + +@dataclass(frozen=True, slots=True) +class NumberToken: + literal: str + + +type ExactJson = dict[str, ExactJson] | list[ExactJson] | str | bool | NumberToken | None SEMANTIC_HEADERS: Final = frozenset({"content-type", "accept", "anthropic-version", "anthropic-beta", "openai-beta"}) AUTH_HEADERS: Final = frozenset({"authorization", "x-api-key"}) @@ -93,8 +99,8 @@ def _exact_value(value: ExactJson) -> JsonValue: return {"array": [_exact_value(item) for item in value]} case bool(): return {"boolean": value} - case int() | Decimal(): - return {"number": str(value)} + case NumberToken(literal=literal): + return {"number": literal} case str(): return {"string": value} case None: @@ -140,13 +146,17 @@ def strict_identity( parsed: Final = ( JSON_VALUE.validate_python( json.loads( - body, object_pairs_hook=_unique_object, parse_constant=_invalid_constant, parse_float=Decimal + body, + object_pairs_hook=_unique_object, + parse_constant=_invalid_constant, + parse_float=NumberToken, + parse_int=NumberToken, ) ) if body else None ) - except (ValueError, UnicodeError, DecimalException): + except (ValueError, UnicodeError): return IneligibleRequest("invalid JSON or duplicate JSON object keys") if body and not isinstance(parsed, dict): return IneligibleRequest("stateless inference requires a JSON object") @@ -160,7 +170,7 @@ def strict_identity( query=tuple((key, "" if key.lower() in CREDENTIAL_QUERY else value) for key, value in query_pairs), headers={key: value for key, value in lowered.items() if key in SEMANTIC_HEADERS}, auth={ - key: (value.partition(" ")[0] if key == "authorization" else "present") + key: (value.partition(" ")[0].lower() if key == "authorization" else "present") for key, value in lowered.items() if key in AUTH_HEADERS }, diff --git a/tests/e2e/test_provider_edge.py b/tests/e2e/test_provider_edge.py index 2d84e143517..dac4e9c2fbe 100644 --- a/tests/e2e/test_provider_edge.py +++ b/tests/e2e/test_provider_edge.py @@ -1434,6 +1434,10 @@ class TestStrictIdentity: b'{"value":0.123456789012345678901}', b'{"value":0.123456789012345678902}', b'{"value":1e400}', + b'{"value":1}', + b'{"value":1e0}', + b'{"value":-0}', + b'{"value":1e9999999999999999999}', ], ) def test_json_values_remain_distinct(self, tmp_path: Path, body: bytes) -> None: @@ -1462,6 +1466,10 @@ class TestStrictIdentity: b'{"value":0.123456789012345678901}', b'{"value":0.123456789012345678902}', b'{"value":1e400}', + b'{"value":1}', + b'{"value":1e0}', + b'{"value":-0}', + b'{"value":1e9999999999999999999}', ) for rejected in ( call_edge(edge, "POST", CHAT_PATH, body=value, headers={"content-type": "application/json"}) @@ -1487,7 +1495,6 @@ class TestStrictIdentity: (CHAT_PATH, b"opaque", {"content-type": "application/octet-stream"}), (CHAT_PATH, b"--boundary", {"content-type": "multipart/form-data; boundary=boundary"}), (CHAT_PATH, b'{"x":1,"x":2}', {"content-type": "application/json"}), - (CHAT_PATH, b'{"x":1e9999999999999999999}', {"content-type": "application/json"}), (CHAT_PATH, b"{}", {"content-type": "application/json", "x-custom-behavior": "synthetic-private-value"}), ], ) @@ -1531,7 +1538,7 @@ class TestStrictIdentity: assert isinstance(recorder, BundleRecorder) headers: Final = { "content-type": "application/json", - "authorization": "Bearer synthetic-token", + "authorization": "bEaReR synthetic-token", "x-api-key": "synthetic-api-key", "cookie": "synthetic-cookie", } @@ -1643,7 +1650,7 @@ assert replay_leftover_error(mode_raw="replay", bundle_dir=Path(sys.argv[1]), te "POST", CHAT_PATH, body=b"{}", - headers={**headers, "authorization": "Bearer synthetic-token"}, + headers={**headers, "authorization": "bEaReR synthetic-token"}, ).status_code == 200 ) From 51db9405149871b3091b4c21d675503221d03ea5 Mon Sep 17 00:00:00 2001 From: Yuneng Jiang Date: Mon, 14 Sep 2026 17:20:58 -0700 Subject: [PATCH 3/6] test: relocate strict replay harness coverage --- .github/workflows/test-code-quality.yml | 5 +- .../test_provider_replay_harness.py | 329 ++++++++++++++++++ tests/e2e/CONTRIBUTING.md | 2 + tests/e2e/test_provider_edge.py | 304 ---------------- 4 files changed, 334 insertions(+), 306 deletions(-) create mode 100644 tests/code_coverage_tests/test_provider_replay_harness.py diff --git a/.github/workflows/test-code-quality.yml b/.github/workflows/test-code-quality.yml index ae117a6b637..e5261d28c29 100644 --- a/.github/workflows/test-code-quality.yml +++ b/.github/workflows/test-code-quality.yml @@ -86,9 +86,10 @@ jobs: - name: test_provider_replay_harness run: | pwd - uv run --no-sync pytest -q --noconftest -o addopts= -p no:rerunfailures \ + uv run --no-sync pytest -q --noconftest -o addopts= -o pythonpath=tests/e2e -p no:rerunfailures \ tests/e2e/test_provider_edge.py tests/e2e/test_fixture_bundle.py \ - tests/e2e/test_fixture_canonical.py tests/e2e/test_fixture_mode.py + tests/e2e/test_fixture_canonical.py tests/e2e/test_fixture_mode.py \ + tests/code_coverage_tests/test_provider_replay_harness.py - name: router_code_coverage run: uv run --no-sync python ./tests/code_coverage_tests/router_code_coverage.py diff --git a/tests/code_coverage_tests/test_provider_replay_harness.py b/tests/code_coverage_tests/test_provider_replay_harness.py new file mode 100644 index 00000000000..e7c5c96b64b --- /dev/null +++ b/tests/code_coverage_tests/test_provider_replay_harness.py @@ -0,0 +1,329 @@ +from __future__ import annotations + +import json +import os +import subprocess +import sys +import threading +from pathlib import Path +from typing import Final + +import pytest +from fixture_bundle import BundleRecorder, LoadedBundle, load_bundle, prepare_bundle +from fixture_mode import current_test_key +from fixture_profile import MatchProfile +from provider_edge import REPLAY_MISS_STATUS, RecordEdge, ReplayEdge, ReplaySource +from test_provider_edge import ( + CHAT_PATH, + SSE_CHUNKS, + STREAM_BODY, + UPLOAD_PATH, + call_edge, + chunked_provider, + fake_provider, + json_object, + provider_url, + raw_stream_post, + running_edge, + this_tests_files, +) + + +class TestStrictIdentity: + @pytest.mark.parametrize("path", [CHAT_PATH, "/anthropic/v1/messages"]) + def test_roundtrip_rejects_semantic_changes(self, tmp_path: Path, path: str) -> None: + recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1") + assert isinstance(recorder, BundleRecorder) + original: Final = ( + b'{"model":"synthetic","messages":[{"role":"user",' + b'"content":"2031-04-05 00000000-0000-0000-0000-000000000001"}],"options":[1,2]}' + ) + headers: Final = { + "content-type": "application/json", + "accept": "application/json", + "anthropic-version": "2023-06-01", + "anthropic-beta": "feature-a", + "openai-beta": "feature-b", + "authorization": "Bearer synthetic-secret-one", + } + query: Final = "?part=one&part=two&blank=" + with fake_provider() as provider: + mounts: Final = {"openai": provider_url(provider), "anthropic": provider_url(provider)} + with running_edge(RecordEdge(recorder, threading.Lock()), mounts) as edge: + captured: Final = call_edge(edge, "POST", path + query, body=original, headers=headers) + assert captured.status_code == 200 + assert json_object(captured.body)["echo"] == original.decode() + loaded: Final = load_bundle(recorder.root, profile="stateless_v1") + assert isinstance(loaded, LoadedBundle) + assert loaded.manifest.match_profile == "stateless_v1" + source: Final = ReplaySource(loaded) + with running_edge(ReplayEdge(source), mounts) as edge: + cases: Final = ( + (original.replace(b"2031-04-05", b"2032-06-07"), headers, query, "body"), + (original.replace(b"000000000001", b"000000000002"), headers, query, "body"), + (original.replace(b"[1,2]", b"[2,1]"), headers, query, "body"), + (original.replace(b"synthetic", b"other"), headers, query, "body"), + (original, headers, "?part=three&part=two&blank=", "query"), + (original, headers, "?part=two&part=one&blank=", "query"), + *( + (original, {k: v for k, v in headers.items() if k != name}, query, "headers") + for name in ("accept", "anthropic-version", "anthropic-beta", "openai-beta") + ), + *( + (original, {**headers, name: value}, query, "headers") + for name in ("accept", "anthropic-version", "anthropic-beta", "openai-beta") + for value in ("different", "") + ), + (original, {**headers, "authorization": "Basic synthetic-secret-two"}, query, "auth"), + (original, {k: v for k, v in headers.items() if k != "authorization"}, query, "auth"), + ) + for rejected, reason in ( + (call_edge(edge, "POST", path + changed_query, body=body, headers=changed_headers), reason) + for body, changed_headers, changed_query, reason in cases + ): + assert rejected.status_code == REPLAY_MISS_STATUS + assert reason in rejected.body.decode() + assert b"synthetic-secret" not in rejected.body + reordered: Final = json.dumps(dict(reversed(list(json_object(original).items())))).encode() + accepted: Final = call_edge( + edge, "POST", path + query, body=reordered, headers={k.upper(): v for k, v in headers.items()} + ) + assert accepted.status_code == 200 + assert accepted.body == captured.body + assert source.leftover_error(current_test_key()) is None + assert len(provider.hits) == 1 + assert "synthetic-secret" not in "".join(file.read_text() for file in recorder.root.rglob("*.json")) + + @pytest.mark.parametrize( + "body", + [ + b'{"value":null}', + b'{"value":""}', + b'{"value":false}', + b'{"value":0}', + b'{"value":[]}', + b'{"value":{}}', + b'{"value":0.123456789012345678901}', + b'{"value":0.123456789012345678902}', + b'{"value":1e400}', + b'{"value":1}', + b'{"value":1e0}', + b'{"value":-0}', + b'{"value":1e9999999999999999999}', + ], + ) + def test_json_values_remain_distinct(self, tmp_path: Path, body: bytes) -> None: + recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1") + assert isinstance(recorder, BundleRecorder) + with fake_provider() as provider: + mounts: Final = {"openai": provider_url(provider)} + with running_edge(RecordEdge(recorder, threading.Lock()), mounts) as edge: + assert ( + call_edge( + edge, "POST", CHAT_PATH, body=body, headers={"content-type": "application/json"} + ).status_code + == 200 + ) + loaded: Final = load_bundle(recorder.root, profile="stateless_v1") + assert isinstance(loaded, LoadedBundle) + with running_edge(ReplayEdge(ReplaySource(loaded)), mounts) as edge: + values: Final = ( + b"{}", + b'{"value":null}', + b'{"value":""}', + b'{"value":false}', + b'{"value":0}', + b'{"value":[]}', + b'{"value":{}}', + b'{"value":0.123456789012345678901}', + b'{"value":0.123456789012345678902}', + b'{"value":1e400}', + b'{"value":1}', + b'{"value":1e0}', + b'{"value":-0}', + b'{"value":1e9999999999999999999}', + ) + for rejected in ( + call_edge(edge, "POST", CHAT_PATH, body=value, headers={"content-type": "application/json"}) + for value in values + if value != body + ): + assert rejected.status_code == REPLAY_MISS_STATUS + assert b"body" in rejected.body + assert ( + call_edge( + edge, "POST", CHAT_PATH, body=body, headers={"content-type": "application/json"} + ).status_code + == 200 + ) + assert len(provider.hits) == 1 + + @pytest.mark.parametrize( + "path,body,headers", + [ + (UPLOAD_PATH, b"{}", {"content-type": "application/json"}), + (CHAT_PATH + "?part=%FF", b"{}", {"content-type": "application/json"}), + (CHAT_PATH + "?part=%FE", b"{}", {"content-type": "application/json"}), + (CHAT_PATH, b"opaque", {"content-type": "application/octet-stream"}), + (CHAT_PATH, b"--boundary", {"content-type": "multipart/form-data; boundary=boundary"}), + (CHAT_PATH, b'{"x":1,"x":2}', {"content-type": "application/json"}), + (CHAT_PATH, b"{}", {"content-type": "application/json", "x-custom-behavior": "synthetic-private-value"}), + ], + ) + def test_ineligible_capture_never_calls_provider( + self, tmp_path: Path, path: str, body: bytes, headers: dict[str, str] + ) -> None: + recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1") + assert isinstance(recorder, BundleRecorder) + with fake_provider() as provider: + with running_edge(RecordEdge(recorder, threading.Lock()), {"openai": provider_url(provider)}) as edge: + result: Final = call_edge(edge, "POST", path, body=body, headers=headers) + assert result.status_code == REPLAY_MISS_STATUS + assert b"eligibility error" in result.body + assert b"synthetic-private-value" not in result.body + assert provider.hits == [] + assert this_tests_files(recorder.root) == [] + + def test_destination_is_part_of_actual_http_identity(self, tmp_path: Path) -> None: + recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1") + assert isinstance(recorder, BundleRecorder) + with fake_provider() as provider: + with running_edge(RecordEdge(recorder, threading.Lock()), {"openai": provider_url(provider)}) as edge: + assert ( + call_edge( + edge, "POST", CHAT_PATH, body=b"{}", headers={"content-type": "application/json"} + ).status_code + == 200 + ) + loaded: Final = load_bundle(recorder.root, profile="stateless_v1") + assert isinstance(loaded, LoadedBundle) + with running_edge(ReplayEdge(ReplaySource(loaded)), {"openai": provider_url(provider) + "/other"}) as edge: + result: Final = call_edge( + edge, "POST", CHAT_PATH, body=b"{}", headers={"content-type": "application/json"} + ) + assert result.status_code == REPLAY_MISS_STATUS + assert b"upstream" in result.body + assert len(provider.hits) == 1 + + def test_credentials_are_not_identity_and_fresh_process_replays(self, tmp_path: Path) -> None: + recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1") + assert isinstance(recorder, BundleRecorder) + headers: Final = { + "content-type": "application/json", + "authorization": "bEaReR synthetic-token", + "x-api-key": "synthetic-api-key", + "cookie": "synthetic-cookie", + } + path: Final = CHAT_PATH + "?api_key=synthetic-query-secret&part=one&part=two" + body: Final = b'{"model":"synthetic","messages":[]}' + with fake_provider(echo_request=False) as provider: + mounts: Final = {"openai": provider_url(provider)} + with running_edge(RecordEdge(recorder, threading.Lock()), mounts) as edge: + captured: Final = call_edge(edge, "POST", path, body=body, headers=headers) + assert captured.status_code == 200 + seen_headers, seen_body = provider.requests[0] + assert {k.lower(): v for k, v in seen_headers.items()}.items() >= headers.items() + assert seen_body == body + assert provider.hits == ["POST " + path.removeprefix("/openai")] + artifacts: Final = "".join(file.read_text() for file in recorder.root.rglob("*.json")) + for secret in ("synthetic-token", "synthetic-api-key", "synthetic-cookie", "synthetic-query-secret"): + assert secret not in artifacts + child: Final = subprocess.run( + [ + sys.executable, + "-c", + """ +import json, sys +from pathlib import Path +from fixture_bundle import LoadedBundle, load_bundle +from provider_edge import ProviderRequestObservation, observed_provider_edge, replay_leftover_error +from test_provider_edge import call_edge +from fixture_profile import MatchProfile +from fixture_mode import current_test_key +loaded = load_bundle(Path(sys.argv[1]), profile="stateless_v1") +assert isinstance(loaded, LoadedBundle) +with observed_provider_edge(ProviderRequestObservation("synthetic"), mode_raw="replay", bundle_dir=Path(sys.argv[1]), bind_host="127.0.0.1", advertise_host="127.0.0.1", mounts={"openai": sys.argv[2]}) as edge: + response = call_edge(edge, "POST", sys.argv[3], body=sys.argv[4].encode(), headers=json.loads(sys.argv[5])) + assert response.status_code == 200 + print(response.body.decode()) +assert replay_leftover_error(mode_raw="replay", bundle_dir=Path(sys.argv[1]), test_key=current_test_key()) is None +""", + str(recorder.root), + provider_url(provider), + path.replace("synthetic-query-secret", "new-query-credential"), + body.decode(), + json.dumps({**headers, "authorization": "Bearer another-credential", "x-api-key": "another-key"}), + ], + env={ + **os.environ, + "PYTHONPATH": str(Path(__file__).resolve().parents[1] / "e2e"), + "E2E_REPLAY_MATCH_PROFILE": "stateless_v1", + }, + capture_output=True, + text=True, + timeout=30, + ) + assert child.returncode == 0, child.stderr + assert child.stdout.strip().encode() == captured.body + assert len(provider.hits) == 1 + + @pytest.mark.parametrize("profile,other", [("legacy", "stateless_v1"), ("stateless_v1", "legacy")]) + def test_profiles_cannot_load_each_others_bundles( + self, tmp_path: Path, profile: MatchProfile, other: MatchProfile + ) -> None: + from fixture_bundle import UnreadableBundle + + recorder: Final = prepare_bundle(tmp_path / profile, profile=profile) + assert isinstance(recorder, BundleRecorder) + mismatch: Final = load_bundle(recorder.root, profile=other) + assert isinstance(mismatch, UnreadableBundle) + assert "profile mismatch" in mismatch.reason + assert "re-record" in mismatch.reason + + @pytest.mark.parametrize("abort_after", [None, 2]) + def test_strict_stream_preserves_chunks_and_truncation(self, tmp_path: Path, abort_after: int | None) -> None: + recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1") + assert isinstance(recorder, BundleRecorder) + with chunked_provider(abort_after=abort_after) as provider: + mounts: Final = {"anthropic": provider_url(provider)} + with running_edge(RecordEdge(recorder, threading.Lock()), mounts) as edge: + _, captured, captured_ending = raw_stream_post(edge.port, "/anthropic/v1/messages", STREAM_BODY) + loaded: Final = load_bundle(recorder.root, profile="stateless_v1") + assert isinstance(loaded, LoadedBundle) + source: Final = ReplaySource(loaded) + with running_edge(ReplayEdge(source), mounts) as edge: + _, replayed, ending = raw_stream_post(edge.port, "/anthropic/v1/messages", STREAM_BODY) + assert captured == replayed == list(SSE_CHUNKS[:abort_after]) + assert ending == captured_ending + assert (ending == "terminated") == (abort_after is None) + assert source.leftover_error(current_test_key()) is None + assert len(provider.hits) == 1 + + def test_auth_scheme_survives_missing_credentials(self, tmp_path: Path) -> None: + recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1") + assert isinstance(recorder, BundleRecorder) + headers: Final = {"content-type": "application/json", "authorization": "Bearer"} + with fake_provider() as provider: + mounts: Final = {"openai": provider_url(provider)} + with running_edge(RecordEdge(recorder, threading.Lock()), mounts) as edge: + assert call_edge(edge, "POST", CHAT_PATH, body=b"{}", headers=headers).status_code == 200 + loaded: Final = load_bundle(recorder.root, profile="stateless_v1") + assert isinstance(loaded, LoadedBundle) + with running_edge(ReplayEdge(ReplaySource(loaded)), mounts) as edge: + for result in ( + call_edge(edge, "POST", CHAT_PATH, body=b"{}", headers={**headers, "authorization": scheme}) + for scheme in ("Basic", "Digest") + ): + assert result.status_code == REPLAY_MISS_STATUS + assert b"auth" in result.body + assert ( + call_edge( + edge, + "POST", + CHAT_PATH, + body=b"{}", + headers={**headers, "authorization": "bEaReR synthetic-token"}, + ).status_code + == 200 + ) + assert len(provider.hits) == 1 diff --git a/tests/e2e/CONTRIBUTING.md b/tests/e2e/CONTRIBUTING.md index 313085b2eed..fbd41c87f66 100644 --- a/tests/e2e/CONTRIBUTING.md +++ b/tests/e2e/CONTRIBUTING.md @@ -242,3 +242,5 @@ Strict matching preserves dates, UUIDs, hashes, model names, tool arguments, arr The semantic header set is `content-type`, `accept`, `anthropic-version`, `anthropic-beta` and `openai-beta`, including missing versus present values. Authorization records presence and the case-insensitive scheme; `x-api-key` records presence only. Credential values and cookies are excluded. Credential query values are redacted while their position and field name remain in the identity. Never use real customer inputs in fixture qualification Excluded transport and telemetry headers are `host`, `content-length`, `connection`, `accept-encoding`, `user-agent`, `traceparent`, `tracestate`, `x-request-id`, `x-client-request-id` and `x-stainless-*`. Inbound transfer-encoding is unsupported; send JSON with content-length framing. The destination represents host identity and the relay carries original body bytes. Replay does not verify credentials, SDK timeout/retry behavior, transport performance, model availability or stateful remote IDs. Live relay uses original request bytes and header values, never the stored identity + +Strict replay harness regression tests live in `tests/code_coverage_tests/test_provider_replay_harness.py`. The `test_provider_replay_harness` code-quality step runs them alongside the existing legacy harness files with `--noconftest -o pythonpath=tests/e2e`; they need only synthetic HTTP providers and temporary fixture storage diff --git a/tests/e2e/test_provider_edge.py b/tests/e2e/test_provider_edge.py index dac4e9c2fbe..72c998a8cc2 100644 --- a/tests/e2e/test_provider_edge.py +++ b/tests/e2e/test_provider_edge.py @@ -24,9 +24,6 @@ from __future__ import annotations import base64 import json -import os -import subprocess -import sys import socket import threading from collections.abc import Generator, Mapping @@ -50,7 +47,6 @@ from fixture_bundle import ( slug_for_test, ) from fixture_canonical import canonicalize -from fixture_profile import MatchProfile from fixture_mode import current_test_key from provider_edge import ( REPLAY_MISS_STATUS, @@ -1355,303 +1351,3 @@ class TestProviderRequestObservation: response: Final = call_edge(edge, "POST", CHAT_PATH, body=chat_body("observed-lantern")) assert response.status_code == 502 assert observation.count == 1 - - -class TestStrictIdentity: - @pytest.mark.parametrize("path", [CHAT_PATH, "/anthropic/v1/messages"]) - def test_roundtrip_rejects_semantic_changes(self, tmp_path: Path, path: str) -> None: - recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1") - assert isinstance(recorder, BundleRecorder) - original: Final = ( - b'{"model":"synthetic","messages":[{"role":"user",' - b'"content":"2031-04-05 00000000-0000-0000-0000-000000000001"}],"options":[1,2]}' - ) - headers: Final = { - "content-type": "application/json", - "accept": "application/json", - "anthropic-version": "2023-06-01", - "anthropic-beta": "feature-a", - "openai-beta": "feature-b", - "authorization": "Bearer synthetic-secret-one", - } - query: Final = "?part=one&part=two&blank=" - with fake_provider() as provider: - mounts: Final = {"openai": provider_url(provider), "anthropic": provider_url(provider)} - with running_edge(RecordEdge(recorder, threading.Lock()), mounts) as edge: - captured: Final = call_edge(edge, "POST", path + query, body=original, headers=headers) - assert captured.status_code == 200 - assert json_object(captured.body)["echo"] == original.decode() - loaded: Final = load_bundle(recorder.root, profile="stateless_v1") - assert isinstance(loaded, LoadedBundle) - assert loaded.manifest.match_profile == "stateless_v1" - source: Final = ReplaySource(loaded) - with running_edge(ReplayEdge(source), mounts) as edge: - cases: Final = ( - (original.replace(b"2031-04-05", b"2032-06-07"), headers, query, "body"), - (original.replace(b"000000000001", b"000000000002"), headers, query, "body"), - (original.replace(b"[1,2]", b"[2,1]"), headers, query, "body"), - (original.replace(b"synthetic", b"other"), headers, query, "body"), - (original, headers, "?part=three&part=two&blank=", "query"), - (original, headers, "?part=two&part=one&blank=", "query"), - *( - (original, {k: v for k, v in headers.items() if k != name}, query, "headers") - for name in ("accept", "anthropic-version", "anthropic-beta", "openai-beta") - ), - *( - (original, {**headers, name: value}, query, "headers") - for name in ("accept", "anthropic-version", "anthropic-beta", "openai-beta") - for value in ("different", "") - ), - (original, {**headers, "authorization": "Basic synthetic-secret-two"}, query, "auth"), - (original, {k: v for k, v in headers.items() if k != "authorization"}, query, "auth"), - ) - for rejected, reason in ( - (call_edge(edge, "POST", path + changed_query, body=body, headers=changed_headers), reason) - for body, changed_headers, changed_query, reason in cases - ): - assert rejected.status_code == REPLAY_MISS_STATUS - assert reason in rejected.body.decode() - assert b"synthetic-secret" not in rejected.body - reordered: Final = json.dumps(dict(reversed(list(json_object(original).items())))).encode() - accepted: Final = call_edge( - edge, "POST", path + query, body=reordered, headers={k.upper(): v for k, v in headers.items()} - ) - assert accepted.status_code == 200 - assert accepted.body == captured.body - assert source.leftover_error(current_test_key()) is None - assert len(provider.hits) == 1 - assert "synthetic-secret" not in "".join(file.read_text() for file in recorder.root.rglob("*.json")) - - @pytest.mark.parametrize( - "body", - [ - b'{"value":null}', - b'{"value":""}', - b'{"value":false}', - b'{"value":0}', - b'{"value":[]}', - b'{"value":{}}', - b'{"value":0.123456789012345678901}', - b'{"value":0.123456789012345678902}', - b'{"value":1e400}', - b'{"value":1}', - b'{"value":1e0}', - b'{"value":-0}', - b'{"value":1e9999999999999999999}', - ], - ) - def test_json_values_remain_distinct(self, tmp_path: Path, body: bytes) -> None: - recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1") - assert isinstance(recorder, BundleRecorder) - with fake_provider() as provider: - mounts: Final = {"openai": provider_url(provider)} - with running_edge(RecordEdge(recorder, threading.Lock()), mounts) as edge: - assert ( - call_edge( - edge, "POST", CHAT_PATH, body=body, headers={"content-type": "application/json"} - ).status_code - == 200 - ) - loaded: Final = load_bundle(recorder.root, profile="stateless_v1") - assert isinstance(loaded, LoadedBundle) - with running_edge(ReplayEdge(ReplaySource(loaded)), mounts) as edge: - values: Final = ( - b"{}", - b'{"value":null}', - b'{"value":""}', - b'{"value":false}', - b'{"value":0}', - b'{"value":[]}', - b'{"value":{}}', - b'{"value":0.123456789012345678901}', - b'{"value":0.123456789012345678902}', - b'{"value":1e400}', - b'{"value":1}', - b'{"value":1e0}', - b'{"value":-0}', - b'{"value":1e9999999999999999999}', - ) - for rejected in ( - call_edge(edge, "POST", CHAT_PATH, body=value, headers={"content-type": "application/json"}) - for value in values - if value != body - ): - assert rejected.status_code == REPLAY_MISS_STATUS - assert b"body" in rejected.body - assert ( - call_edge( - edge, "POST", CHAT_PATH, body=body, headers={"content-type": "application/json"} - ).status_code - == 200 - ) - assert len(provider.hits) == 1 - - @pytest.mark.parametrize( - "path,body,headers", - [ - (UPLOAD_PATH, b"{}", {"content-type": "application/json"}), - (CHAT_PATH + "?part=%FF", b"{}", {"content-type": "application/json"}), - (CHAT_PATH + "?part=%FE", b"{}", {"content-type": "application/json"}), - (CHAT_PATH, b"opaque", {"content-type": "application/octet-stream"}), - (CHAT_PATH, b"--boundary", {"content-type": "multipart/form-data; boundary=boundary"}), - (CHAT_PATH, b'{"x":1,"x":2}', {"content-type": "application/json"}), - (CHAT_PATH, b"{}", {"content-type": "application/json", "x-custom-behavior": "synthetic-private-value"}), - ], - ) - def test_ineligible_capture_never_calls_provider( - self, tmp_path: Path, path: str, body: bytes, headers: dict[str, str] - ) -> None: - recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1") - assert isinstance(recorder, BundleRecorder) - with fake_provider() as provider: - with running_edge(RecordEdge(recorder, threading.Lock()), {"openai": provider_url(provider)}) as edge: - result: Final = call_edge(edge, "POST", path, body=body, headers=headers) - assert result.status_code == REPLAY_MISS_STATUS - assert b"eligibility error" in result.body - assert b"synthetic-private-value" not in result.body - assert provider.hits == [] - assert this_tests_files(recorder.root) == [] - - def test_destination_is_part_of_actual_http_identity(self, tmp_path: Path) -> None: - recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1") - assert isinstance(recorder, BundleRecorder) - with fake_provider() as provider: - with running_edge(RecordEdge(recorder, threading.Lock()), {"openai": provider_url(provider)}) as edge: - assert ( - call_edge( - edge, "POST", CHAT_PATH, body=b"{}", headers={"content-type": "application/json"} - ).status_code - == 200 - ) - loaded: Final = load_bundle(recorder.root, profile="stateless_v1") - assert isinstance(loaded, LoadedBundle) - with running_edge(ReplayEdge(ReplaySource(loaded)), {"openai": provider_url(provider) + "/other"}) as edge: - result: Final = call_edge( - edge, "POST", CHAT_PATH, body=b"{}", headers={"content-type": "application/json"} - ) - assert result.status_code == REPLAY_MISS_STATUS - assert b"upstream" in result.body - assert len(provider.hits) == 1 - - def test_credentials_are_not_identity_and_fresh_process_replays(self, tmp_path: Path) -> None: - recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1") - assert isinstance(recorder, BundleRecorder) - headers: Final = { - "content-type": "application/json", - "authorization": "bEaReR synthetic-token", - "x-api-key": "synthetic-api-key", - "cookie": "synthetic-cookie", - } - path: Final = CHAT_PATH + "?api_key=synthetic-query-secret&part=one&part=two" - body: Final = b'{"model":"synthetic","messages":[]}' - with fake_provider(echo_request=False) as provider: - mounts: Final = {"openai": provider_url(provider)} - with running_edge(RecordEdge(recorder, threading.Lock()), mounts) as edge: - captured: Final = call_edge(edge, "POST", path, body=body, headers=headers) - assert captured.status_code == 200 - seen_headers, seen_body = provider.requests[0] - assert {k.lower(): v for k, v in seen_headers.items()}.items() >= headers.items() - assert seen_body == body - assert provider.hits == ["POST " + path.removeprefix("/openai")] - artifacts: Final = "".join(file.read_text() for file in recorder.root.rglob("*.json")) - for secret in ("synthetic-token", "synthetic-api-key", "synthetic-cookie", "synthetic-query-secret"): - assert secret not in artifacts - child: Final = subprocess.run( - [ - sys.executable, - "-c", - """ -import json, sys -from pathlib import Path -from fixture_bundle import LoadedBundle, load_bundle -from provider_edge import ProviderRequestObservation, observed_provider_edge, replay_leftover_error -from test_provider_edge import call_edge -from fixture_profile import MatchProfile -from fixture_mode import current_test_key -loaded = load_bundle(Path(sys.argv[1]), profile="stateless_v1") -assert isinstance(loaded, LoadedBundle) -with observed_provider_edge(ProviderRequestObservation("synthetic"), mode_raw="replay", bundle_dir=Path(sys.argv[1]), bind_host="127.0.0.1", advertise_host="127.0.0.1", mounts={"openai": sys.argv[2]}) as edge: - response = call_edge(edge, "POST", sys.argv[3], body=sys.argv[4].encode(), headers=json.loads(sys.argv[5])) - assert response.status_code == 200 - print(response.body.decode()) -assert replay_leftover_error(mode_raw="replay", bundle_dir=Path(sys.argv[1]), test_key=current_test_key()) is None -""", - str(recorder.root), - provider_url(provider), - path.replace("synthetic-query-secret", "new-query-credential"), - body.decode(), - json.dumps({**headers, "authorization": "Bearer another-credential", "x-api-key": "another-key"}), - ], - env={ - **os.environ, - "PYTHONPATH": str(Path(__file__).parent), - "E2E_REPLAY_MATCH_PROFILE": "stateless_v1", - }, - capture_output=True, - text=True, - timeout=30, - ) - assert child.returncode == 0, child.stderr - assert child.stdout.strip().encode() == captured.body - assert len(provider.hits) == 1 - - @pytest.mark.parametrize("profile,other", [("legacy", "stateless_v1"), ("stateless_v1", "legacy")]) - def test_profiles_cannot_load_each_others_bundles( - self, tmp_path: Path, profile: MatchProfile, other: MatchProfile - ) -> None: - from fixture_bundle import UnreadableBundle - - recorder: Final = prepare_bundle(tmp_path / profile, profile=profile) - assert isinstance(recorder, BundleRecorder) - mismatch: Final = load_bundle(recorder.root, profile=other) - assert isinstance(mismatch, UnreadableBundle) - assert "profile mismatch" in mismatch.reason - assert "re-record" in mismatch.reason - - @pytest.mark.parametrize("abort_after", [None, 2]) - def test_strict_stream_preserves_chunks_and_truncation(self, tmp_path: Path, abort_after: int | None) -> None: - recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1") - assert isinstance(recorder, BundleRecorder) - with chunked_provider(abort_after=abort_after) as provider: - mounts: Final = {"anthropic": provider_url(provider)} - with running_edge(RecordEdge(recorder, threading.Lock()), mounts) as edge: - _, captured, captured_ending = raw_stream_post(edge.port, "/anthropic/v1/messages", STREAM_BODY) - loaded: Final = load_bundle(recorder.root, profile="stateless_v1") - assert isinstance(loaded, LoadedBundle) - source: Final = ReplaySource(loaded) - with running_edge(ReplayEdge(source), mounts) as edge: - _, replayed, ending = raw_stream_post(edge.port, "/anthropic/v1/messages", STREAM_BODY) - assert captured == replayed == list(SSE_CHUNKS[:abort_after]) - assert ending == captured_ending - assert (ending == "terminated") == (abort_after is None) - assert source.leftover_error(current_test_key()) is None - assert len(provider.hits) == 1 - - def test_auth_scheme_survives_missing_credentials(self, tmp_path: Path) -> None: - recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1") - assert isinstance(recorder, BundleRecorder) - headers: Final = {"content-type": "application/json", "authorization": "Bearer"} - with fake_provider() as provider: - mounts: Final = {"openai": provider_url(provider)} - with running_edge(RecordEdge(recorder, threading.Lock()), mounts) as edge: - assert call_edge(edge, "POST", CHAT_PATH, body=b"{}", headers=headers).status_code == 200 - loaded: Final = load_bundle(recorder.root, profile="stateless_v1") - assert isinstance(loaded, LoadedBundle) - with running_edge(ReplayEdge(ReplaySource(loaded)), mounts) as edge: - for result in ( - call_edge(edge, "POST", CHAT_PATH, body=b"{}", headers={**headers, "authorization": scheme}) - for scheme in ("Basic", "Digest") - ): - assert result.status_code == REPLAY_MISS_STATUS - assert b"auth" in result.body - assert ( - call_edge( - edge, - "POST", - CHAT_PATH, - body=b"{}", - headers={**headers, "authorization": "bEaReR synthetic-token"}, - ).status_code - == 200 - ) - assert len(provider.hits) == 1 From 37bde0bdbe83642cac1ed2e696c3c89c36981ef7 Mon Sep 17 00:00:00 2001 From: Yuneng Jiang Date: Mon, 14 Sep 2026 17:27:18 -0700 Subject: [PATCH 4/6] test: keep provider request snapshots immutable --- tests/e2e/test_provider_edge.py | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/tests/e2e/test_provider_edge.py b/tests/e2e/test_provider_edge.py index 72c998a8cc2..81be81e7b59 100644 --- a/tests/e2e/test_provider_edge.py +++ b/tests/e2e/test_provider_edge.py @@ -31,6 +31,7 @@ from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path +from types import MappingProxyType from typing import Final import pytest @@ -85,7 +86,10 @@ class _FakeProvider(ThreadingHTTPServer): super().__init__(bind, _FakeProviderHandler) self.hits: list[str] = [] self.echo_request = echo_request - self.requests: list[tuple[dict[str, str], bytes]] = [] + self.requests: tuple[tuple[Mapping[str, str], bytes], ...] = () + + def capture_request(self, headers: Mapping[str, str], body: bytes) -> None: + self.requests = (*self.requests, (MappingProxyType(dict(headers)), body)) class _FakeProviderHandler(BaseHTTPRequestHandler): @@ -103,8 +107,12 @@ class _FakeProviderHandler(BaseHTTPRequestHandler): 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}") - provider.requests.append((dict(self.headers.items()), body)) - payload = json.dumps({"echo": body.decode("utf-8"), "path": self.path, "hit": len(provider.hits)} if provider.echo_request else {"ok": True}).encode() + provider.capture_request(dict(self.headers.items()), body) + payload: Final = json.dumps( + {"echo": body.decode("utf-8"), "path": self.path, "hit": len(provider.hits)} + if provider.echo_request + else {"ok": True} + ).encode() self.send_response(200) self.send_header("content-type", "application/json") self.send_header("content-length", str(len(payload))) From 16fb44f23ab97af1e1b1e84e92be169804bb3b38 Mon Sep 17 00:00:00 2001 From: Yuneng Jiang Date: Mon, 14 Sep 2026 21:03:55 -0700 Subject: [PATCH 5/6] ci: run provider replay harness in CircleCI --- .circleci/config.yml | 21 +++++++++++++++++++++ .github/workflows/test-code-quality.yml | 8 -------- tests/e2e/CONTRIBUTING.md | 2 +- 3 files changed, 22 insertions(+), 9 deletions(-) diff --git a/.circleci/config.yml b/.circleci/config.yml index 7c241af5853..6b5d7a67189 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -2915,6 +2915,25 @@ jobs: exit 1 fi + provider_replay_harness: + docker: + - *python312_image + working_directory: ~/project + resource_class: medium + steps: + - setup_litellm_test_deps + - run: + name: Test provider replay harness + command: | + mkdir -p test-results/provider-replay-harness + uv run --no-sync pytest -q --noconftest -o addopts= -o pythonpath=tests/e2e -p no:rerunfailures \ + --junitxml=test-results/provider-replay-harness/junit.xml \ + tests/e2e/test_provider_edge.py tests/e2e/test_fixture_bundle.py \ + tests/e2e/test_fixture_canonical.py tests/e2e/test_fixture_mode.py \ + tests/code_coverage_tests/test_provider_replay_harness.py + - store_test_results: + path: test-results/provider-replay-harness + integration_contracts: parameters: suite: @@ -2967,6 +2986,8 @@ workflows: only: - main - /litellm_.*/ + - provider_replay_harness: + filters: *main_branches - base_sdk_install: filters: *main_branches - local_testing_part1: diff --git a/.github/workflows/test-code-quality.yml b/.github/workflows/test-code-quality.yml index e07593cdfb6..987f66773f2 100644 --- a/.github/workflows/test-code-quality.yml +++ b/.github/workflows/test-code-quality.yml @@ -83,14 +83,6 @@ jobs: - name: test_e2e_changed_gate run: uv run --no-sync pytest -q --noconftest -p no:cacheprovider -c /dev/null tests/code_coverage_tests/test_e2e_changed_gate.py tests/code_coverage_tests/test_e2e_idp_stack.py - - name: test_provider_replay_harness - run: | - pwd - uv run --no-sync pytest -q --noconftest -o addopts= -o pythonpath=tests/e2e -p no:rerunfailures \ - tests/e2e/test_provider_edge.py tests/e2e/test_fixture_bundle.py \ - tests/e2e/test_fixture_canonical.py tests/e2e/test_fixture_mode.py \ - tests/code_coverage_tests/test_provider_replay_harness.py - - name: router_code_coverage run: uv run --no-sync python ./tests/code_coverage_tests/router_code_coverage.py diff --git a/tests/e2e/CONTRIBUTING.md b/tests/e2e/CONTRIBUTING.md index 5beaf68df79..53a05931ca7 100644 --- a/tests/e2e/CONTRIBUTING.md +++ b/tests/e2e/CONTRIBUTING.md @@ -247,4 +247,4 @@ The semantic header set is `content-type`, `accept`, `anthropic-version`, `anthr Excluded transport and telemetry headers are `host`, `content-length`, `connection`, `accept-encoding`, `user-agent`, `traceparent`, `tracestate`, `x-request-id`, `x-client-request-id` and `x-stainless-*`. Inbound transfer-encoding is unsupported; send JSON with content-length framing. The destination represents host identity and the relay carries original body bytes. Replay does not verify credentials, SDK timeout/retry behavior, transport performance, model availability or stateful remote IDs. Live relay uses original request bytes and header values, never the stored identity -Strict replay harness regression tests live in `tests/code_coverage_tests/test_provider_replay_harness.py`. The `test_provider_replay_harness` code-quality step runs them alongside the existing legacy harness files with `--noconftest -o pythonpath=tests/e2e`; they need only synthetic HTTP providers and temporary fixture storage +Strict replay harness regression tests live in `tests/code_coverage_tests/test_provider_replay_harness.py`. The CircleCI `provider_replay_harness` job runs them alongside the existing legacy harness files with `--noconftest -o pythonpath=tests/e2e`; they need only synthetic HTTP providers and temporary fixture storage From 6969bd9c548ca0507ac132c684be333e306258b7 Mon Sep 17 00:00:00 2001 From: Yuneng Jiang Date: Mon, 14 Sep 2026 21:12:16 -0700 Subject: [PATCH 6/6] ci: run replay harness on every admitted CircleCI pipeline --- .circleci/config.yml | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/.circleci/config.yml b/.circleci/config.yml index 6b5d7a67189..bb4ad0f4019 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -2986,8 +2986,7 @@ workflows: only: - main - /litellm_.*/ - - provider_replay_harness: - filters: *main_branches + - provider_replay_harness - base_sdk_install: filters: *main_branches - local_testing_part1: