mirror of
https://github.com/tiennm99/litellm.git
synced 2026-08-01 12:22:11 +00:00
fix(tests/vcr): make Redis cassette cache replay deterministically (zero VCR misses on consecutive runs) (#28826)
* test(vcr): make Redis-backed cassettes replay deterministically across runs - Pin LITELLM_LOCAL_MODEL_COST_MAP=True in the shared VCR harness so the per-test importlib.reload(litellm) no longer fetches the model cost map from raw.githubusercontent.com. That live fetch was being recorded into cassettes; for tests that subsequently skip it was the only recorded episode, so the persister refused to save it (skipped tests don't persist) and the test re-recorded it live every run (MISS:NOT_PERSISTED). - Compare-time symmetric matcher tolerance for Google OAuth (ya29.*) tokens, observability/telemetry payloads, credential-exchange bodies, and volatile UUID/timestamp tokens, so existing cassettes select a recorded episode instead of growing past the 50-episode cap and re-recording live. - Don't record fire-and-forget telemetry (langfuse/arize/otel/...) into non-telemetry tests' cassettes. Several modules set litellm.success_callback at import time, so observability logging is globally enabled and an async flush from the background logging worker lands in an unrelated test's VCR window, saved as a spurious MISS:RECORDED (observed: a Langfuse batch from another completion landing on test_lowest_latency_routing_buffer). Such a request now passes through live (telemetry hosts aren't real-spend hosts); tests that actually assert on telemetry keep recording it. - Dedupe + cap the VCR diagnostic dump so the classification summary survives CircleCI's ~400KB step-output truncation. - Stabilize a non-deterministic rate-limit test body; mark AWS Secrets Manager lifecycle tests VCR-incompatible (uniquely-named secrets can't be replayed). - Mark test_router_text_completion_client VCR-incompatible: it fires 300 identical requests to verify async-client reuse, but vcrpy patches the HTTP transport so replay never exercises the real connection pool the test validates, and recording 300 near-identical episodes overflows the 50-episode cap (MISS:OVERFLOW every run). It hits a free mock endpoint. - Mark the Vertex AI MaaS Mistral OCR tests (vertex_ai/mistral-ocr-2505) VCR-incompatible: the MaaS model is not provisioned in the CI GCP project, so the live :rawPredict call fails and the test skips every run, leaving no cassette to record (MISS:NOT_PERSISTED every run). Sibling direct-Mistral and Azure OCR tests are unaffected and still replay from cache. * fix(tests/vcr): refresh cassette TTL on read so replayed cassettes don't expire The Redis VCR persister loaded cassettes with a plain GET, which does not touch the key's TTL. A cassette that is only ever replayed (HIT/NOOP, never re-recorded) therefore expired exactly 24h after its last *write*, no matter how often it was read. Whichever CI run happened to cross that boundary re-recorded the cassette live and surfaced a spurious VCR MISS on otherwise deterministic cassettes — the residual per-run flakiness floor (a different random subset of read-only cassettes expiring each run). Slide the expiry forward on every successful load (best-effort EXPIRE), so any cassette used at least once per TTL window stays alive indefinitely and the 2nd/3rd run of a day replays cleanly. * fix(tests/vcr): recover from spurious GET-None for existing cassette keys Under concurrent CI load, the persister's load GET was observed returning None for a cassette key that demonstrably existed on the (single, non- clustered) Redis master — an external monitor saw the key present with a healthy TTL at the same instant the in-process client read None. Because None is a valid GET result (not a RedisError), the retry-on-error client config never engaged, so the cassette re-recorded live (a phantom MISS:RECORDED); for flaky/networked tests the failed live call then triggered a pytest rerun, which is why a rotating subset of otherwise deterministic tests missed each run. On a None result, re-check EXISTS and re-read once. If the key really exists, use the recovered value and log [vcr-transient-miss-recovered] (also counted in cassette_cache_health). A genuinely absent key (a new cassette) still falls through to CassetteNotFoundError. * chore(tests/vcr): TEMP diagnostic for persistent-miss cassette load path Logs GET/EXISTS at load time for the three cassettes that re-record every run despite being present in Redis, to capture what the in-process client sees. To be reverted before merge. * chore(tests/vcr): write load diagnostic to Redis (truncation-proof) CI stdout truncates to the last ~400KB, dropping the early loaddbg lines for the alphabetically-first failing test. Push the load probe to a Redis list instead so it survives. To be reverted before merge. * fix(tests/vcr): don't drop stored telemetry episodes during cassette load Root cause of the residual per-run misses on present cassettes: vcrpy's Cassette._load() replays each *stored* interaction through Cassette.append(), which runs before_record_request on it — and a None return there silently drops that episode. The telemetry-leak suppressor (_should_drop_telemetry_record) returns None for telemetry requests, so when a non-telemetry-named test (or the alphabetically-first test in a worker, whose _current_test_nodeid is still empty) loaded a cassette containing a Langfuse ingestion episode, the episode was dropped on read — forcing an endless live re-record (a phantom MISS:RECORDED on a cassette that was demonstrably present in Redis). Verified by reproducing Cassette._load() against the real cassette: empty/non-telemetry nodeid -> 0 episodes survive; with the guard -> 1 survives. Fix: guard the suppressor with a thread-local set around Cassette._load (via a small idempotent monkeypatch), so the drop only ever stops *new* incidental telemetry from being recorded and never filters the existing cassette on read. Also drops the speculative GET-None recovery + its diagnostics from the previous commits: the load diagnostic showed GET returns the cassette bytes fine (get=1440B), so the persister never returned a spurious None — the loss happened later in vcrpy's append. The proven TTL-refresh-on-read fix is retained. * fix(tests/vcr): drop incidental telemetry export POSTs to stop rotating async-flush misses litellm's observability loggers flush on a background thread, so a Langfuse ingestion POST scheduled by one telemetry test can fire mid-way through a *later* telemetry-named test (after that test's own httpx mock has exited) and be recorded by VCR as a phantom episode — a non-deterministic MISS:RECORDED / PARTIAL that rotates onto a different telemetry test from run to run. Telemetry export POSTs are fire-and-forget; no test asserts on a *recorded* export response except the pass-through proxy test (which forwards a client POST to Langfuse ingestion and replays its 207). So _should_drop_telemetry_record now drops incidental export POSTs for every test except that one. Dropping returns None (live fire-and-forget, never stored), so it can only turn a phantom miss into a harmless live call, never the reverse; recorded read-back GETs that telemetry tests assert on are matched by method and left untouched. * fix(tests/vcr): restore assertion in test_banner_silent_when_vcr_disabled The assertion that the banner is suppressed when VCR is disabled was inadvertently moved into test_diagnostic_log_silent_when_no_dir when the diagnostic-log tests were added, leaving the disabled-VCR test verifying nothing. Co-authored-by: Yassin Kortam <yassin@berri.ai> --------- Co-authored-by: Cursor Agent <cursoragent@cursor.com> Co-authored-by: Yassin Kortam <yassin@berri.ai>
This commit is contained in:
Mateo Wang
co-authored by
Cursor Agent
Yassin Kortam
Cursor Agent
Yassin Kortam
parent
f75a7c6b22
commit
533eab4dbd
+409
-10
@@ -13,10 +13,12 @@ import os
|
||||
import re
|
||||
import socket
|
||||
import sys
|
||||
import threading
|
||||
from collections import defaultdict
|
||||
from typing import Iterable
|
||||
|
||||
import pytest
|
||||
import vcr.matchers as _vcr_matchers
|
||||
|
||||
from tests._vcr_redis_persister import (
|
||||
MAX_EPISODES_PER_CASSETTE,
|
||||
@@ -29,11 +31,28 @@ from tests._vcr_redis_persister import (
|
||||
patch_vcrpy_aiohttp_record_path,
|
||||
)
|
||||
|
||||
# Force litellm to use its bundled model-cost-map backup instead of fetching it
|
||||
# from raw.githubusercontent.com on import. Several VCR conftests reload litellm
|
||||
# in an autouse fixture (``importlib.reload(litellm)``); ``litellm.__init__``
|
||||
# calls ``get_model_cost_map()`` which issues a live ``httpx.get`` unless this is
|
||||
# set. While a cassette is active that fetch gets *recorded* as an extra episode
|
||||
# (it was present in ~710 of ~1900 cached cassettes). For tests that then skip,
|
||||
# it is the only recorded episode, so the persister refuses to save it (skipped
|
||||
# tests don't persist) and the test re-records it live and is classified
|
||||
# MISS:NOT_PERSISTED on every run. Pinning to the local backup removes the
|
||||
# network call entirely, so skip tests record nothing (NOOP) and passing tests
|
||||
# stop carrying a volatile github episode. This matches the established idiom in
|
||||
# the unit-test suite, which sets the same flag (see e.g.
|
||||
# tests/test_litellm/test_cost_calculator.py). ``setdefault`` so an explicit
|
||||
# override still wins.
|
||||
os.environ.setdefault("LITELLM_LOCAL_MODEL_COST_MAP", "True")
|
||||
|
||||
CASSETTE_CACHE_HIGH_WATER_FRACTION = 0.85
|
||||
|
||||
|
||||
SAFE_BODY_MATCHER_NAME = "safe_body"
|
||||
KEY_FINGERPRINT_MATCHER_NAME = "key_fingerprint"
|
||||
TOLERANT_QUERY_MATCHER_NAME = "tolerant_query"
|
||||
KEY_FINGERPRINT_HEADER = "x-litellm-key-fp"
|
||||
|
||||
VCR_DIAG_DIR_ENV = "LITELLM_VCR_DIAG_DIR"
|
||||
@@ -73,6 +92,17 @@ def reset_vcr_diag_dir() -> None:
|
||||
pass
|
||||
|
||||
|
||||
# CircleCI truncates a step's retrievable output to the last ~400 KB. The
|
||||
# diagnostic log is emitted right *before* the final pytest summary line but
|
||||
# *after* the VCR CLASSIFICATION SUMMARY, so an unbounded dump (the body/key
|
||||
# matchers log one block per *episode comparison*, even on an eventual HIT)
|
||||
# pushes the classification summary out of the retrievable window and makes
|
||||
# misses impossible to read in CI. Dedupe identical blocks (the same mismatch
|
||||
# is logged against every non-matching episode) and cap the total emitted size
|
||||
# so the summary always survives.
|
||||
VCR_DIAG_EMIT_MAX_LINES = 400
|
||||
|
||||
|
||||
def emit_vcr_diagnostic_log(terminalreporter) -> None:
|
||||
directory = _vcr_diag_dir()
|
||||
if not os.path.isdir(directory):
|
||||
@@ -83,25 +113,56 @@ def emit_vcr_diagnostic_log(terminalreporter) -> None:
|
||||
return
|
||||
if not files:
|
||||
return
|
||||
terminalreporter.write_sep("=", "VCR DIAGNOSTIC LOG", bold=True)
|
||||
terminalreporter.write_line(
|
||||
f" source dir: {directory} (also archived as a CI artifact)"
|
||||
)
|
||||
|
||||
# Collect every line, tagged by source file, deduplicating identical lines
|
||||
# (with an occurrence count) so the repeated per-episode mismatch blocks
|
||||
# collapse to one representative each.
|
||||
seen_counts: dict[str, int] = defaultdict(int)
|
||||
ordered: list[tuple[str, str]] = [] # (source_file, line)
|
||||
read_errors: list[str] = []
|
||||
for name in files:
|
||||
path = os.path.join(directory, name)
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8") as fh:
|
||||
content = fh.read()
|
||||
except OSError as exc:
|
||||
terminalreporter.write_line(
|
||||
read_errors.append(
|
||||
f" [failed to read {name}: {type(exc).__name__}: {exc}]"
|
||||
)
|
||||
continue
|
||||
if not content.strip():
|
||||
continue
|
||||
terminalreporter.write_sep("-", name, bold=False)
|
||||
for line in content.splitlines():
|
||||
terminalreporter.write_line(line)
|
||||
if not line.strip():
|
||||
continue
|
||||
seen_counts[line] += 1
|
||||
if seen_counts[line] == 1:
|
||||
ordered.append((name, line))
|
||||
|
||||
if not ordered and not read_errors:
|
||||
return
|
||||
|
||||
terminalreporter.write_sep("=", "VCR DIAGNOSTIC LOG", bold=True)
|
||||
terminalreporter.write_line(
|
||||
f" source dir: {directory} (deduplicated; full log archived as a CI artifact)"
|
||||
)
|
||||
for line in read_errors:
|
||||
terminalreporter.write_line(line)
|
||||
|
||||
emitted = 0
|
||||
last_source = None
|
||||
for name, line in ordered:
|
||||
if emitted >= VCR_DIAG_EMIT_MAX_LINES:
|
||||
terminalreporter.write_line(
|
||||
f" ... {len(ordered) - emitted} more unique diagnostic line(s) "
|
||||
"suppressed to keep the classification summary retrievable in CI."
|
||||
)
|
||||
break
|
||||
if name != last_source:
|
||||
terminalreporter.write_sep("-", name, bold=False)
|
||||
last_source = name
|
||||
count = seen_counts.get(line, 1)
|
||||
suffix = f" (x{count})" if count > 1 else ""
|
||||
terminalreporter.write_line(line + suffix)
|
||||
emitted += 1
|
||||
terminalreporter.write_sep("=", bold=True)
|
||||
|
||||
|
||||
@@ -326,6 +387,285 @@ def _canonical_body(request) -> tuple[bytes, str]:
|
||||
return b"", pre_type
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Volatile-token body normalization (compare-time only).
|
||||
#
|
||||
# Many tests append a cache-buster to the request body so the *live* call
|
||||
# isn't served from an upstream prompt/response cache during recording:
|
||||
# ``f"...{time.time()}"``, ``f"...{uuid.uuid4()}"``. LiteLLM's own
|
||||
# observability payloads (langfuse/otel) likewise carry per-call UUIDs and
|
||||
# ISO-8601 timestamps. None of that affects what the test asserts (response
|
||||
# shape, cost, caching behaviour), but it makes the request body differ on
|
||||
# every run, so vcrpy never matches and the cassette keeps appending episodes
|
||||
# until it overflows ``MAX_EPISODES_PER_CASSETTE`` and re-records live forever.
|
||||
#
|
||||
# We canonicalize these volatile substrings to fixed placeholders *only for
|
||||
# matching* (in ``_safe_body_matcher``), never in what we store — so the
|
||||
# cassette on disk keeps the real bytes for debuggability, and the
|
||||
# normalization is applied symmetrically to both the incoming and the stored
|
||||
# request. Because it's symmetric and compare-time, it can never mask a
|
||||
# response-level discrepancy; it only changes which recorded episode is
|
||||
# selected. This mirrors the existing SigV4 / multipart-boundary / b64-image
|
||||
# normalizations already in this module, and means the already-bloated
|
||||
# cassettes start replaying immediately without a flush + re-record.
|
||||
_VCR_UUID_RE = re.compile(
|
||||
rb"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}"
|
||||
)
|
||||
# ISO-8601 timestamps, e.g. ``2026-05-25T03:40:37.262045Z`` /
|
||||
# ``2026-05-25T03:40:37+00:00``.
|
||||
_VCR_ISO_TS_RE = re.compile(
|
||||
rb"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:?\d{2})?"
|
||||
)
|
||||
# Unix epoch as 13-digit milliseconds, then 10-digit ``time.time()`` float,
|
||||
# then 10-digit integer seconds. Anchored to ``1`` + 9/12 digits, which keeps
|
||||
# them inside the 2001-2033 / 2001-2033 epoch windows and avoids matching
|
||||
# ordinary identifiers. Order matters: the longer/float forms are substituted
|
||||
# before the bare-integer form so the integer rule can't bite off a prefix.
|
||||
_VCR_UNIX_MS_RE = re.compile(rb"(?<![\d.])1[0-9]{12}(?![\d.])")
|
||||
_VCR_UNIX_FLOAT_RE = re.compile(rb"(?<![\d.])1[0-9]{9}\.[0-9]+")
|
||||
_VCR_UNIX_INT_RE = re.compile(rb"(?<![\d.])1[0-9]{9}(?![\d.])")
|
||||
|
||||
|
||||
def _normalize_volatile_tokens(body: bytes) -> bytes:
|
||||
"""Replace per-run cache-busters (UUIDs / timestamps) with placeholders.
|
||||
|
||||
Compare-time only — see the module note above. Returns ``body`` unchanged
|
||||
when it contains none of these patterns, so deterministic requests are
|
||||
unaffected.
|
||||
"""
|
||||
if not body:
|
||||
return body
|
||||
body = _VCR_UUID_RE.sub(b"<vcr-uuid>", body)
|
||||
body = _VCR_ISO_TS_RE.sub(b"<vcr-iso-ts>", body)
|
||||
body = _VCR_UNIX_MS_RE.sub(b"<vcr-unix-ms>", body)
|
||||
body = _VCR_UNIX_FLOAT_RE.sub(b"<vcr-unix-float>", body)
|
||||
body = _VCR_UNIX_INT_RE.sub(b"<vcr-unix-int>", body)
|
||||
return body
|
||||
|
||||
|
||||
# Hosts whose request body is a rotating credential exchange (a freshly signed
|
||||
# JWT ``assertion=...`` or refresh-token grant). The body changes on every run
|
||||
# and carries no information the test asserts on, so matching on
|
||||
# method+scheme+host+port+path+query is sufficient — skip the body comparison.
|
||||
_CREDENTIAL_EXCHANGE_HOSTS = (
|
||||
"oauth2.googleapis.com",
|
||||
"sts.googleapis.com",
|
||||
"accounts.google.com",
|
||||
"metadata.google.internal",
|
||||
"169.254.169.254",
|
||||
)
|
||||
|
||||
|
||||
def _request_host(request) -> str:
|
||||
uri = getattr(request, "uri", None) or getattr(request, "url", "") or ""
|
||||
uri = str(uri)
|
||||
if "//" not in uri:
|
||||
return ""
|
||||
rest = uri.split("//", 1)[1]
|
||||
return rest.split("/", 1)[0].split("@")[-1].split(":")[0].lower()
|
||||
|
||||
|
||||
def _is_credential_exchange_request(request) -> bool:
|
||||
return _request_host(request) in _CREDENTIAL_EXCHANGE_HOSTS
|
||||
|
||||
|
||||
# Observability / telemetry backends LiteLLM logs to. A telemetry export is a
|
||||
# snapshot of the *whole* call — fresh span/trace UUIDs, ISO-8601 timestamps,
|
||||
# durations, token costs, the LiteLLM build SHA (``release``), and the recorded
|
||||
# LLM response content — and tests often round-trip a fresh ``trace_id`` back
|
||||
# through the backend's query API to verify logging happened. None of that is
|
||||
# reproducible under deterministic replay, and none of it is what the test
|
||||
# asserts on (it checks redaction / presence, or a locally-computed trace id).
|
||||
# So for these hosts we match on method+scheme+host+port+path only: the
|
||||
# expensive LLM call still matches normally and stays cached, while the cheap
|
||||
# telemetry POST/GET replays from the recorded response. This is why the body
|
||||
# and query matchers below both short-circuit for telemetry hosts.
|
||||
_TELEMETRY_HOST_SUFFIXES = (
|
||||
"langfuse.com",
|
||||
"arize.com",
|
||||
"phoenix.arize.com",
|
||||
"traceloop.com",
|
||||
"braintrust.dev",
|
||||
"comet.com",
|
||||
"wandb.ai",
|
||||
"honeycomb.io",
|
||||
"signoz.io",
|
||||
)
|
||||
|
||||
|
||||
def _is_telemetry_request(request) -> bool:
|
||||
host = _request_host(request)
|
||||
if not host:
|
||||
return False
|
||||
return any(host == s or host.endswith("." + s) for s in _TELEMETRY_HOST_SUFFIXES)
|
||||
|
||||
|
||||
# Nodeid of the test currently executing, set per-test by
|
||||
# ``install_live_call_probe`` (runs in the autouse gate at setup). Used to
|
||||
# decide whether an incidental telemetry POST should be recorded — see
|
||||
# ``_should_drop_telemetry_record``. xdist workers are separate processes and
|
||||
# tests run sequentially within a worker, so a plain module global is safe.
|
||||
_current_test_nodeid: str = ""
|
||||
|
||||
# Test files/dirs that legitimately record & replay telemetry HTTP (they assert
|
||||
# on the outgoing observability payload or query the backend back). Identified
|
||||
# by a substring of the test path. Everything else is treated as a non-telemetry
|
||||
# test for which a telemetry call is incidental leakage (see below).
|
||||
_TELEMETRY_TEST_PATH_MARKERS = (
|
||||
"langfuse",
|
||||
"arize",
|
||||
"phoenix",
|
||||
"traceloop",
|
||||
"braintrust",
|
||||
"comet",
|
||||
"wandb",
|
||||
"honeycomb",
|
||||
"signoz",
|
||||
"otel",
|
||||
"opentelemetry",
|
||||
"telemetry",
|
||||
"observability",
|
||||
"logging", # tests/logging_callback_tests, logging_testing dirs
|
||||
)
|
||||
|
||||
|
||||
def _current_test_records_telemetry() -> bool:
|
||||
nodeid = _current_test_nodeid.lower()
|
||||
return any(marker in nodeid for marker in _TELEMETRY_TEST_PATH_MARKERS)
|
||||
|
||||
|
||||
# Test paths that legitimately RECORD AND REPLAY a telemetry *export* POST and
|
||||
# assert on its response. Only the pass-through proxy test does this: it
|
||||
# forwards a client POST to Langfuse's ``/api/public/ingestion`` and asserts the
|
||||
# upstream multi-status (207) it replays from the cassette. Every other
|
||||
# telemetry test either mocks the export client and asserts on the mock (the
|
||||
# langfuse e2e suite) or asserts on a read-back GET / an in-memory span exporter
|
||||
# — for those the export POST is fire-and-forget and must not be recorded (see
|
||||
# ``_should_drop_telemetry_record``).
|
||||
_TELEMETRY_EXPORT_REPLAY_TEST_MARKERS = ("pass_through",)
|
||||
|
||||
|
||||
def _current_test_replays_telemetry_export() -> bool:
|
||||
nodeid = _current_test_nodeid.lower()
|
||||
return any(m in nodeid for m in _TELEMETRY_EXPORT_REPLAY_TEST_MARKERS)
|
||||
|
||||
|
||||
def _is_telemetry_export_request(request) -> bool:
|
||||
"""A telemetry *export* — a span/trace/event ingestion call, always a POST
|
||||
to an observability host. Read-backs (verifying a trace landed) are GETs."""
|
||||
if not _is_telemetry_request(request):
|
||||
return False
|
||||
return str(getattr(request, "method", "") or "").upper() == "POST"
|
||||
|
||||
|
||||
# Thread-local "we are inside Cassette._load" flag. vcrpy's ``Cassette._load``
|
||||
# replays each *stored* interaction through ``Cassette.append``, which runs
|
||||
# ``before_record_request`` on it; a ``None`` return there silently drops the
|
||||
# stored episode. ``_should_drop_telemetry_record`` must therefore NOT fire
|
||||
# during load, or it would delete already-recorded telemetry episodes the
|
||||
# instant a non-telemetry-named test (or the very first test in a worker, whose
|
||||
# ``_current_test_nodeid`` is still empty) loads them — forcing an endless live
|
||||
# re-record (a phantom MISS:RECORDED on a cassette that was present in Redis).
|
||||
# The drop is only ever meant to stop *new* incidental telemetry from being
|
||||
# recorded, never to filter the existing cassette on read. ``_load`` and its
|
||||
# ``append`` calls run synchronously in one thread, so a thread-local correctly
|
||||
# scopes the guard and never masks a concurrent background-flush record.
|
||||
_vcr_load_guard = threading.local()
|
||||
|
||||
|
||||
def _vcr_load_in_progress() -> bool:
|
||||
return getattr(_vcr_load_guard, "active", False)
|
||||
|
||||
|
||||
def patch_vcrpy_cassette_load_guard() -> None:
|
||||
"""Wrap ``Cassette._load`` so ``_should_drop_telemetry_record`` is inert
|
||||
while stored episodes are being replayed into the in-memory cassette."""
|
||||
import vcr.cassette as _cassette_mod
|
||||
|
||||
if getattr(_cassette_mod.Cassette._load, "_litellm_load_guarded", False):
|
||||
return
|
||||
_orig_load = _cassette_mod.Cassette._load
|
||||
|
||||
def _guarded_load(self):
|
||||
_vcr_load_guard.active = True
|
||||
try:
|
||||
return _orig_load(self)
|
||||
finally:
|
||||
_vcr_load_guard.active = False
|
||||
|
||||
_guarded_load._litellm_load_guarded = True
|
||||
_cassette_mod.Cassette._load = _guarded_load
|
||||
|
||||
|
||||
def _should_drop_telemetry_record(request) -> bool:
|
||||
"""Whether to refuse to record this request into the active cassette.
|
||||
|
||||
Several test modules set ``litellm.success_callback = ["langfuse"]`` (and
|
||||
similar) at *import* time, which globally enables observability logging for
|
||||
the whole worker. Unrelated tests then emit telemetry whose async flush
|
||||
(litellm's background logging worker) lands in a *later* test's VCR window
|
||||
and gets saved as a spurious episode — a non-deterministic MISS:RECORDED on
|
||||
whichever test happened to be active (observed on
|
||||
``test_lowest_latency_routing_buffer`` carrying a Langfuse batch from an
|
||||
unrelated completion). Refusing to record telemetry for non-telemetry tests
|
||||
makes the leak a harmless live fire-and-forget call instead (telemetry hosts
|
||||
are not in ``_LIVE_CALL_HOST_SUFFIXES``, so the probe doesn't flag it, and
|
||||
vcrpy treats a ``None`` from ``before_record_request`` as "don't record" and
|
||||
"can't replay" → the request passes through live and is never stored).
|
||||
Tests that actually assert on telemetry keep recording it.
|
||||
|
||||
Crucially, this never fires while ``Cassette._load`` is replaying stored
|
||||
interactions (see ``_vcr_load_in_progress``): dropping there would delete an
|
||||
already-recorded telemetry episode on read and force a live re-record.
|
||||
|
||||
The async-flush leak also rotates *within* the telemetry test set: litellm's
|
||||
observability loggers flush on a background thread, so an export POST
|
||||
scheduled by one telemetry test fires mid-way through a *later*
|
||||
telemetry-named test (after that test's own ``httpx`` mock has exited) and
|
||||
is recorded as a phantom episode — a non-deterministic MISS:RECORDED /
|
||||
PARTIAL that lands on a different telemetry test from run to run. Telemetry
|
||||
*export* POSTs are fire-and-forget; no test asserts on a recorded export
|
||||
response except the pass-through proxy test (which forwards to Langfuse
|
||||
ingestion and replays its 207). So drop incidental export POSTs everywhere
|
||||
else too — dropping returns ``None`` (live fire-and-forget, never stored),
|
||||
which can only turn a phantom miss into a harmless live call, never the
|
||||
reverse. Recorded read-back GETs that telemetry tests assert on are matched
|
||||
by method and so are left untouched.
|
||||
"""
|
||||
if _vcr_load_in_progress():
|
||||
return False
|
||||
if not _is_telemetry_request(request):
|
||||
return False
|
||||
if (
|
||||
_is_telemetry_export_request(request)
|
||||
and not _current_test_replays_telemetry_export()
|
||||
):
|
||||
return True
|
||||
return not _current_test_records_telemetry()
|
||||
|
||||
|
||||
# Google APIs (Vertex AI, Gemini, OAuth2/STS). Auth is a ``ya29.*`` OAuth2
|
||||
# access token minted fresh on every run, so the per-request key fingerprint
|
||||
# rotates and never matches a recording. The logical credential — the GCP
|
||||
# project — is part of the matched URL path (``/projects/<project>/...``), so
|
||||
# skipping the fingerprint comparison for these hosts keeps cache isolation by
|
||||
# project while letting the existing recordings replay without a re-record.
|
||||
# (We also collapse ``ya29.*`` tokens to one marker in ``_stable_key_value`` so
|
||||
# *new* recordings store a stable fingerprint; this matcher relaxation is what
|
||||
# rescues the cassettes already recorded under the old per-token fingerprints.)
|
||||
_GOOGLE_HOST_SUFFIXES = (
|
||||
"googleapis.com",
|
||||
"google.internal",
|
||||
)
|
||||
|
||||
|
||||
def _is_google_host_request(request) -> bool:
|
||||
host = _request_host(request)
|
||||
if not host:
|
||||
return False
|
||||
return any(host == s or host.endswith("." + s) for s in _GOOGLE_HOST_SUFFIXES)
|
||||
|
||||
|
||||
def _safe_body_matcher(r1, r2) -> None:
|
||||
"""Compare request bodies as bytes; never invokes ``json.loads``.
|
||||
|
||||
@@ -334,11 +674,24 @@ def _safe_body_matcher(r1, r2) -> None:
|
||||
(e.g. the Bedrock batch S3 PUT) before it can return "no match".
|
||||
This matcher is strictly more conservative — the only equivalence
|
||||
it gives up vs. the default is "JSON key order doesn't matter".
|
||||
|
||||
Two compare-time relaxations layer on top, both symmetric so they can
|
||||
never hide a response-level discrepancy:
|
||||
|
||||
* Requests to a rotating-credential-exchange host (Google OAuth2/STS
|
||||
token endpoints) skip the body comparison — the signed-JWT body
|
||||
changes every run. The host matcher still gates the overall match.
|
||||
* Volatile cache-buster tokens (UUIDs / epoch timestamps) are
|
||||
canonicalized away via ``_normalize_volatile_tokens``.
|
||||
"""
|
||||
if _is_credential_exchange_request(r1) or _is_telemetry_request(r1):
|
||||
return
|
||||
body1, pre1 = _canonical_body(r1)
|
||||
body2, pre2 = _canonical_body(r2)
|
||||
if body1 == body2:
|
||||
return
|
||||
if _normalize_volatile_tokens(body1) == _normalize_volatile_tokens(body2):
|
||||
return
|
||||
_emit_body_mismatch_diagnostic(r1, r2, body1, body2, pre1, pre2)
|
||||
raise AssertionError("request bodies differ")
|
||||
|
||||
@@ -398,6 +751,10 @@ _AWS_SIGV4_CREDENTIAL_RE = re.compile(
|
||||
r"AWS4-HMAC-SHA256\s+Credential=([^/\s,]+)/", re.IGNORECASE
|
||||
)
|
||||
|
||||
# Google OAuth2 access tokens always start with ``ya29.`` regardless of how
|
||||
# they were minted (service account, metadata server, impersonation).
|
||||
_GOOGLE_OAUTH_BEARER_RE = re.compile(r"^Bearer\s+ya29\.", re.IGNORECASE)
|
||||
|
||||
|
||||
def _stable_key_value(header_name: str, raw: str) -> str:
|
||||
"""Return a *stable* identifier for a credential header.
|
||||
@@ -414,6 +771,14 @@ def _stable_key_value(header_name: str, raw: str) -> str:
|
||||
match = _AWS_SIGV4_CREDENTIAL_RE.search(raw)
|
||||
if match:
|
||||
return f"aws-sigv4:{match.group(1)}"
|
||||
# Google OAuth2 access tokens (``ya29.*``) are minted fresh from the
|
||||
# service-account credentials on every run, so hashing the raw token
|
||||
# would push every Vertex/Gemini request into a new cassette episode —
|
||||
# exactly the SigV4 failure mode above. The logical credential (the GCP
|
||||
# project) is already part of the matched URL path, so collapse all such
|
||||
# tokens to one stable marker.
|
||||
if _GOOGLE_OAUTH_BEARER_RE.match(raw):
|
||||
return "google-oauth2"
|
||||
return raw
|
||||
|
||||
|
||||
@@ -560,6 +925,12 @@ def _before_record_request(request):
|
||||
this hook is idempotent. The boundary normalizer is also
|
||||
idempotent for the same reason.
|
||||
"""
|
||||
# Refuse to record incidental telemetry leaked from a globally-enabled
|
||||
# observability callback into a non-telemetry test (see
|
||||
# ``_should_drop_telemetry_record``). Returning ``None`` tells vcrpy not to
|
||||
# store the interaction; the request passes through live (fire-and-forget).
|
||||
if _should_drop_telemetry_record(request):
|
||||
return None
|
||||
headers = getattr(request, "headers", None)
|
||||
if headers is None:
|
||||
return request
|
||||
@@ -626,6 +997,12 @@ def _coalesce_chunks_to_bytes(chunks):
|
||||
|
||||
|
||||
def _key_fingerprint_matcher(r1, r2) -> None:
|
||||
# Google OAuth2 access tokens rotate every run; the project in the URL
|
||||
# path (matched separately) is the stable credential identity, so skip the
|
||||
# fingerprint comparison for Google hosts. See ``_is_google_host_request``.
|
||||
if _is_google_host_request(r1):
|
||||
return
|
||||
|
||||
def _fp(req):
|
||||
for value in _iter_header_values(
|
||||
getattr(req, "headers", None), KEY_FINGERPRINT_HEADER
|
||||
@@ -649,6 +1026,20 @@ def _key_fingerprint_matcher(r1, r2) -> None:
|
||||
raise AssertionError("API key fingerprints differ")
|
||||
|
||||
|
||||
def _tolerant_query_matcher(r1, r2) -> None:
|
||||
"""vcrpy's ``query`` matcher, but tolerant of telemetry round-trips.
|
||||
|
||||
Observability backends are queried back with a freshly-generated
|
||||
``trace_id`` (e.g. ``GET /observations?traceId=litellm-test-<uuid>``).
|
||||
Comparing the query string would miss on every run. For telemetry hosts
|
||||
we skip the query comparison entirely (the host+path matchers still gate
|
||||
the match); every other host uses vcrpy's stock query matcher unchanged.
|
||||
"""
|
||||
if _is_telemetry_request(r1):
|
||||
return
|
||||
_vcr_matchers.query(r1, r2)
|
||||
|
||||
|
||||
def vcr_config_dict() -> dict:
|
||||
return {
|
||||
"decode_compressed_response": True,
|
||||
@@ -660,7 +1051,7 @@ def vcr_config_dict() -> dict:
|
||||
"host",
|
||||
"port",
|
||||
"path",
|
||||
"query",
|
||||
TOLERANT_QUERY_MATCHER_NAME,
|
||||
KEY_FINGERPRINT_MATCHER_NAME,
|
||||
SAFE_BODY_MATCHER_NAME,
|
||||
),
|
||||
@@ -725,7 +1116,9 @@ def register_persister_if_enabled(vcr) -> None:
|
||||
vcr.register_persister(make_redis_persister())
|
||||
vcr.register_matcher(SAFE_BODY_MATCHER_NAME, _safe_body_matcher)
|
||||
vcr.register_matcher(KEY_FINGERPRINT_MATCHER_NAME, _key_fingerprint_matcher)
|
||||
vcr.register_matcher(TOLERANT_QUERY_MATCHER_NAME, _tolerant_query_matcher)
|
||||
patch_vcrpy_aiohttp_record_path()
|
||||
patch_vcrpy_cassette_load_guard()
|
||||
global _atexit_banner_registered
|
||||
if not _atexit_banner_registered:
|
||||
atexit.register(_print_atexit_banner)
|
||||
@@ -1386,6 +1779,12 @@ def install_live_call_probe(request, vcr) -> None:
|
||||
intercepts above the socket layer, so any "outbound" socket would be
|
||||
a recording cycle, not real spend.
|
||||
"""
|
||||
# Track the current test for telemetry-leak suppression (applies to every
|
||||
# test, VCR-marked or not). See ``_should_drop_telemetry_record``.
|
||||
global _current_test_nodeid
|
||||
_current_test_nodeid = str(
|
||||
getattr(getattr(request, "node", None), "nodeid", "") or ""
|
||||
)
|
||||
if vcr is not None or vcr_disabled():
|
||||
return None
|
||||
probe = _LiveCallProbe()
|
||||
|
||||
@@ -146,8 +146,9 @@ def make_redis_persister(
|
||||
class _RedisPersister:
|
||||
@staticmethod
|
||||
def load_cassette(cassette_path, serializer):
|
||||
key = redis_key_for(cassette_path)
|
||||
try:
|
||||
data = redis_client.get(redis_key_for(cassette_path))
|
||||
data = redis_client.get(key)
|
||||
except RedisError as exc:
|
||||
_record_cache_failure("load", exc)
|
||||
msg = (
|
||||
@@ -162,7 +163,7 @@ def make_redis_persister(
|
||||
try:
|
||||
if isinstance(data, bytes):
|
||||
data = data.decode("utf-8")
|
||||
return deserialize(data, serializer)
|
||||
result = deserialize(data, serializer)
|
||||
except Exception as exc:
|
||||
_record_cache_failure("load", exc)
|
||||
msg = (
|
||||
@@ -173,6 +174,22 @@ def make_redis_persister(
|
||||
_log.warning(msg)
|
||||
warnings.warn(msg, VCRCassetteCacheWarning, stacklevel=2)
|
||||
raise CassetteNotFoundError() from exc
|
||||
# Slide the expiry forward on every successful read. A plain GET
|
||||
# does not touch the key's TTL, so a cassette that is only ever
|
||||
# replayed (HIT/NOOP, never re-recorded) expires exactly
|
||||
# ``ttl_seconds`` after its last *write* no matter how often it is
|
||||
# read — and whichever CI run happens to cross that boundary
|
||||
# re-records it live, surfacing as a spurious VCR MISS that no
|
||||
# amount of matcher tolerance can prevent. Refreshing the TTL on
|
||||
# read keeps any cassette used at least once per TTL window alive
|
||||
# indefinitely, so the second/third run of a day replays cleanly.
|
||||
# Best-effort: a failed refresh must never turn a successful load
|
||||
# into a miss.
|
||||
try:
|
||||
redis_client.expire(key, ttl_seconds)
|
||||
except RedisError:
|
||||
pass
|
||||
return result
|
||||
|
||||
@staticmethod
|
||||
def save_cassette(cassette_path, cassette_dict, serializer):
|
||||
|
||||
@@ -38,7 +38,22 @@ _VCR_INCOMPATIBLE_FILES = frozenset(
|
||||
}
|
||||
)
|
||||
|
||||
_VCR_INCOMPATIBLE_NODEID_SUFFIXES: tuple[str, ...] = ()
|
||||
# AWS Secrets Manager resource-lifecycle tests. Each run creates a secret
|
||||
# under a per-run unique name (``litellm_test_<uuid>``) and either asserts the
|
||||
# API response echoes that exact unique name or reads it straight back. The
|
||||
# name *must* be unique per run because AWS enforces a >=7-day deletion
|
||||
# recovery window — a fixed name can't be re-created on the daily VCR
|
||||
# re-record. Deterministic replay returns the previously-recorded (different)
|
||||
# name, so the unique-name round-trip cannot be reproduced offline. The
|
||||
# config-parsing tests in the same file (settings / STS endpoint) make no such
|
||||
# unique-resource calls and stay VCR-cached.
|
||||
_VCR_INCOMPATIBLE_NODEID_SUFFIXES: tuple[str, ...] = (
|
||||
"::test_write_and_read_simple_secret",
|
||||
"::test_write_and_read_json_secret",
|
||||
"::test_read_nonexistent_secret",
|
||||
"::test_primary_secret_functionality",
|
||||
"::test_write_secret_with_description_and_tags",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="function", autouse=True)
|
||||
|
||||
@@ -9,7 +9,9 @@ import pytest
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
|
||||
|
||||
from tests._vcr_conftest_common import ( # noqa: E402
|
||||
VCR_DIAG_EMIT_MAX_LINES,
|
||||
emit_cassette_cache_session_banner,
|
||||
emit_vcr_diagnostic_log,
|
||||
)
|
||||
from tests._vcr_redis_persister import ( # noqa: E402
|
||||
_cache_health,
|
||||
@@ -165,6 +167,55 @@ def test_banner_silent_when_vcr_disabled(
|
||||
assert reporter.output == ""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Diagnostic-log dedup + cap. CircleCI truncates step output to the last
|
||||
# ~400 KB; an unbounded diagnostic dump pushes the VCR classification summary
|
||||
# out of the retrievable window, so the dump must dedupe and cap.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_diagnostic_log_dedupes_repeated_blocks(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("LITELLM_VCR_DIAG_DIR", str(tmp_path))
|
||||
(tmp_path / "123.log").write_text(
|
||||
"\n".join(["[vcr-key-fingerprint-matcher] differ"] * 40 + ["unique line"]),
|
||||
encoding="utf-8",
|
||||
)
|
||||
reporter = _FakeTerminalReporter()
|
||||
|
||||
emit_vcr_diagnostic_log(reporter)
|
||||
|
||||
out = reporter.output
|
||||
# The repeated block collapses to a single line with an occurrence count.
|
||||
assert out.count("[vcr-key-fingerprint-matcher] differ") == 1
|
||||
assert "(x40)" in out
|
||||
assert "unique line" in out
|
||||
|
||||
|
||||
def test_diagnostic_log_caps_unique_lines(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("LITELLM_VCR_DIAG_DIR", str(tmp_path))
|
||||
total = VCR_DIAG_EMIT_MAX_LINES + 50
|
||||
(tmp_path / "123.log").write_text(
|
||||
"\n".join(f"unique-diagnostic-{i}" for i in range(total)), encoding="utf-8"
|
||||
)
|
||||
reporter = _FakeTerminalReporter()
|
||||
|
||||
emit_vcr_diagnostic_log(reporter)
|
||||
|
||||
out = reporter.output
|
||||
emitted = sum(1 for ln in out.splitlines() if ln.startswith("unique-diagnostic-"))
|
||||
assert emitted == VCR_DIAG_EMIT_MAX_LINES
|
||||
assert "more unique diagnostic line(s) suppressed" in out
|
||||
|
||||
|
||||
def test_diagnostic_log_silent_when_no_dir(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("LITELLM_VCR_DIAG_DIR", str(tmp_path / "does-not-exist"))
|
||||
reporter = _FakeTerminalReporter()
|
||||
|
||||
emit_vcr_diagnostic_log(reporter)
|
||||
|
||||
assert reporter.output == ""
|
||||
|
||||
|
||||
def test_banner_silent_on_xdist_worker(
|
||||
monkeypatch, vcr_enabled, health_reset, patch_capacity_snapshot
|
||||
):
|
||||
@@ -179,3 +230,164 @@ def test_banner_silent_on_xdist_worker(
|
||||
emit_cassette_cache_session_banner(reporter)
|
||||
|
||||
assert reporter.output == ""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Telemetry-leak suppression. Several modules set ``litellm.success_callback``
|
||||
# at import time, so observability logging is globally enabled and an async
|
||||
# flush can land in an unrelated test's VCR window and be saved as a spurious
|
||||
# MISS:RECORDED episode. ``_should_drop_telemetry_record`` refuses to record a
|
||||
# telemetry call for a non-telemetry test (it passes through live instead),
|
||||
# while tests that actually assert on telemetry keep recording.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class _FakeRequest:
|
||||
def __init__(
|
||||
self, host, scheme="https", method="POST", path="/api/public/ingestion"
|
||||
):
|
||||
self.host = host
|
||||
self.scheme = scheme
|
||||
self.uri = f"{scheme}://{host}{path}"
|
||||
self.headers = {}
|
||||
self.method = method
|
||||
self.body = b"{}"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def current_test(monkeypatch):
|
||||
"""Set the module-global current-test nodeid the suppressor reads."""
|
||||
import tests._vcr_conftest_common as common
|
||||
|
||||
def _set(nodeid):
|
||||
monkeypatch.setattr(common, "_current_test_nodeid", nodeid)
|
||||
|
||||
return _set
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"nodeid,host,method,expected_drop",
|
||||
[
|
||||
# Non-telemetry test: incidental telemetry leak is dropped (not recorded).
|
||||
(
|
||||
"tests/local_testing/test_lowest_latency_routing.py::test_lowest_latency_routing_buffer[1]",
|
||||
"us.cloud.langfuse.com",
|
||||
"POST",
|
||||
True,
|
||||
),
|
||||
(
|
||||
"tests/local_testing/test_function_call_parsing.py::test_parse",
|
||||
"us.cloud.langfuse.com",
|
||||
"POST",
|
||||
True,
|
||||
),
|
||||
(
|
||||
"tests/llm_translation/test_x.py::test_y",
|
||||
"otlp.arize.com",
|
||||
"POST",
|
||||
True,
|
||||
),
|
||||
# Non-telemetry host on a non-telemetry test: never dropped.
|
||||
(
|
||||
"tests/local_testing/test_lowest_latency_routing.py::test_lowest_latency_routing_buffer[1]",
|
||||
"api.openai.com",
|
||||
"POST",
|
||||
False,
|
||||
),
|
||||
# Telemetry EXPORT POSTs are fire-and-forget and dropped even for
|
||||
# telemetry-named tests: litellm's background flush makes them rotate
|
||||
# into a later telemetry test's window as a phantom MISS:RECORDED. The
|
||||
# e2e suite mocks the export client and asserts on the mock; read-back
|
||||
# tests assert on a GET — neither needs the recorded export POST.
|
||||
(
|
||||
"tests/local_testing/test_alangfuse.py::test_langfuse_logging",
|
||||
"us.cloud.langfuse.com",
|
||||
"POST",
|
||||
True,
|
||||
),
|
||||
(
|
||||
"tests/logging_callback_tests/test_langfuse_e2e_test.py::test_e2e",
|
||||
"us.cloud.langfuse.com",
|
||||
"POST",
|
||||
True,
|
||||
),
|
||||
(
|
||||
"tests/logging_callback_tests/test_dynamic_otel_keys.py::test_keys",
|
||||
"otlp.arize.com",
|
||||
"POST",
|
||||
True,
|
||||
),
|
||||
# Read-back GETs that telemetry tests assert on are kept (matched by
|
||||
# method, so the export-POST drop does not touch them).
|
||||
(
|
||||
"tests/local_testing/test_alangfuse.py::test_langfuse_logging",
|
||||
"us.cloud.langfuse.com",
|
||||
"GET",
|
||||
False,
|
||||
),
|
||||
# ...but a read-back GET on a NON-telemetry test is still incidental.
|
||||
(
|
||||
"tests/local_testing/test_function_call_parsing.py::test_parse",
|
||||
"us.cloud.langfuse.com",
|
||||
"GET",
|
||||
True,
|
||||
),
|
||||
# The pass-through proxy test forwards a client POST to Langfuse
|
||||
# ingestion and asserts the replayed 207 — its export POST is kept.
|
||||
(
|
||||
"tests/local_testing/test_pass_through_endpoints.py::test_aaapass_through_endpoint_pass_through_keys_langfuse[False-0-207]",
|
||||
"us.cloud.langfuse.com",
|
||||
"POST",
|
||||
False,
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_should_drop_telemetry_record(
|
||||
current_test, nodeid, host, method, expected_drop
|
||||
):
|
||||
import tests._vcr_conftest_common as common
|
||||
|
||||
current_test(nodeid)
|
||||
req = _FakeRequest(host, method=method)
|
||||
assert common._should_drop_telemetry_record(req) is expected_drop
|
||||
|
||||
|
||||
def test_drop_is_suppressed_while_loading_stored_episodes(current_test):
|
||||
"""During ``Cassette._load`` the drop MUST be inert.
|
||||
|
||||
vcrpy replays each stored interaction through ``Cassette.append`` →
|
||||
``before_record_request``; a ``None`` there silently drops the stored
|
||||
episode. If the telemetry drop fired on load, an already-recorded
|
||||
telemetry episode would be deleted the instant a non-telemetry-named
|
||||
test loaded it, forcing an endless live re-record (a phantom
|
||||
MISS:RECORDED on a cassette that was present in Redis). The drop must
|
||||
only stop *new* incidental recordings, never filter the cassette on read.
|
||||
"""
|
||||
import tests._vcr_conftest_common as common
|
||||
|
||||
# A non-telemetry test loading a stored Langfuse episode: dropped on
|
||||
# record, but must be KEPT while loading.
|
||||
current_test("tests/local_testing/test_lowest_latency_routing.py::test_buf")
|
||||
req = _FakeRequest("us.cloud.langfuse.com")
|
||||
|
||||
assert common._should_drop_telemetry_record(req) is True # record path
|
||||
|
||||
common._vcr_load_guard.active = True
|
||||
try:
|
||||
assert common._vcr_load_in_progress() is True
|
||||
assert common._should_drop_telemetry_record(req) is False # load path
|
||||
finally:
|
||||
common._vcr_load_guard.active = False
|
||||
assert common._should_drop_telemetry_record(req) is True
|
||||
|
||||
|
||||
def test_load_guard_patch_is_idempotent():
|
||||
import vcr.cassette as cassette_mod
|
||||
|
||||
import tests._vcr_conftest_common as common
|
||||
|
||||
common.patch_vcrpy_cassette_load_guard()
|
||||
first = cassette_mod.Cassette._load
|
||||
common.patch_vcrpy_cassette_load_guard()
|
||||
assert cassette_mod.Cassette._load is first
|
||||
assert getattr(cassette_mod.Cassette._load, "_litellm_load_guarded", False)
|
||||
|
||||
@@ -79,6 +79,59 @@ def test_load_missing_key_raises_cassette_not_found():
|
||||
persister.load_cassette("never/recorded", yamlserializer)
|
||||
|
||||
|
||||
def test_load_refreshes_ttl_so_replayed_cassettes_do_not_expire():
|
||||
"""A successful read must slide the cassette's expiry forward.
|
||||
|
||||
Regression: ``load_cassette`` used a plain ``GET``, which does not
|
||||
touch the key's TTL. A cassette that is only ever replayed (HIT/NOOP,
|
||||
never re-recorded) therefore expired exactly ``CASSETTE_TTL_SECONDS``
|
||||
after its last *write* no matter how often it was read, and whichever
|
||||
CI run crossed that 24h boundary re-recorded it live — a spurious VCR
|
||||
MISS on otherwise-deterministic cassettes. Reading must refresh the
|
||||
TTL so an actively-used cassette never expires.
|
||||
"""
|
||||
fake, persister = _persister_with_fake_redis()
|
||||
cassette_id = "tests/llm_translation/test_x/test_ttl_refresh"
|
||||
key = redis_key_for(cassette_id)
|
||||
|
||||
persister.save_cassette(cassette_id, _sample_cassette_dict(), yamlserializer)
|
||||
# Simulate a cassette written ~most-of-a-day ago: only a little TTL left.
|
||||
fake.expire(key, 60)
|
||||
assert fake.ttl(key) <= 60
|
||||
|
||||
persister.load_cassette(cassette_id, yamlserializer)
|
||||
|
||||
refreshed = fake.ttl(key)
|
||||
assert CASSETTE_TTL_SECONDS - 5 <= refreshed <= CASSETTE_TTL_SECONDS
|
||||
|
||||
|
||||
def test_load_ttl_refresh_failure_does_not_break_load():
|
||||
"""A failed TTL refresh must never turn a successful load into a miss."""
|
||||
|
||||
class _RefreshFailsRedis:
|
||||
def __init__(self, inner):
|
||||
self._inner = inner
|
||||
|
||||
def get(self, *args, **kwargs):
|
||||
return self._inner.get(*args, **kwargs)
|
||||
|
||||
def set(self, *args, **kwargs):
|
||||
return self._inner.set(*args, **kwargs)
|
||||
|
||||
def expire(self, *args, **kwargs):
|
||||
raise RedisConnectionError("simulated outage")
|
||||
|
||||
client = _RefreshFailsRedis(fakeredis.FakeStrictRedis())
|
||||
persister = make_redis_persister(client=client)
|
||||
cassette_id = "tests/llm_translation/test_x/test_ttl_refresh_fail"
|
||||
|
||||
persister.save_cassette(cassette_id, _sample_cassette_dict(), yamlserializer)
|
||||
requests, responses = persister.load_cassette(cassette_id, yamlserializer)
|
||||
|
||||
assert len(requests) == 1
|
||||
assert len(responses) == 1
|
||||
|
||||
|
||||
def test_redis_key_normalizes_path_passed_by_pytest_recording():
|
||||
raw = "tests/llm_translation/cassettes/test_anthropic/test_streaming.yaml"
|
||||
assert (
|
||||
|
||||
@@ -68,7 +68,19 @@ _VCR_INCOMPATIBLE_FILES = frozenset(
|
||||
}
|
||||
)
|
||||
|
||||
_VCR_INCOMPATIBLE_NODEID_SUFFIXES: tuple[str, ...] = ()
|
||||
# Individual tests (vs. whole files above) that VCR replay can't model:
|
||||
# - ``test_router_text_completion_client``: a concurrency test that fires 300
|
||||
# identical requests to verify the async OpenAI client is *reused* across
|
||||
# calls (per its own comment, it "fails when we create a new Async OpenAI
|
||||
# client per request"). vcrpy patches the HTTP transport, so replay never
|
||||
# opens real connections and cannot exercise the client pool the test exists
|
||||
# to validate. Recording instead stores ~300 near-identical episodes, which
|
||||
# blows past MAX_EPISODES_PER_CASSETTE (50) so the cassette is refused on
|
||||
# every run (MISS:OVERFLOW). The endpoint is a free mock, so the live calls
|
||||
# carry no real provider cost.
|
||||
_VCR_INCOMPATIBLE_NODEID_SUFFIXES: tuple[str, ...] = (
|
||||
"test_router.py::test_router_text_completion_client",
|
||||
)
|
||||
|
||||
|
||||
_verbose_state = VerboseReporterState()
|
||||
|
||||
@@ -123,8 +123,6 @@ def test_setting_mpr_limits_per_model(
|
||||
|
||||
|
||||
async def _handle_router_calls(router):
|
||||
import random
|
||||
|
||||
pre_fill = """
|
||||
Lorem ipsum dolor sit amet, consectetur adipiscing elit. Nunc ut finibus massa. Quisque a magna magna. Quisque neque diam, varius sit amet tellus eu, elementum fermentum sapien. Integer ut erat eget arcu rutrum blandit. Morbi a metus purus. Nulla porta, urna at finibus malesuada, velit ante suscipit orci, vitae laoreet dui ligula ut augue. Cras elementum pretium dui, nec luctus nulla aliquet ut. Nam faucibus, diam nec semper interdum, nisl nisi viverra nulla, vitae sodales elit ex a purus. Donec tristique malesuada lobortis. Donec posuere iaculis nisl, vitae accumsan libero dignissim dignissim. Suspendisse finibus leo et ex mattis tempor. Praesent at nisl vitae quam egestas lacinia. Donec in justo non erat aliquam accumsan sed vitae ex. Vivamus gravida diam vel ipsum tincidunt dignissim.
|
||||
|
||||
@@ -141,7 +139,11 @@ async def _handle_router_calls(router):
|
||||
[
|
||||
{
|
||||
"role": "user",
|
||||
"content": f"{pre_fill * 3}\n\nRecite the Declaration of independence at a speed of {random.random() * 100} words per minute.",
|
||||
# Fixed speed (was random.random()*100) so the request body is
|
||||
# deterministic and the VCR cassette replays instead of
|
||||
# appending a new episode every run. This is a rate-limiting
|
||||
# test; the prompt content is irrelevant to what it asserts.
|
||||
"content": f"{pre_fill * 3}\n\nRecite the Declaration of independence at a speed of 50.0 words per minute.",
|
||||
}
|
||||
],
|
||||
stream=True,
|
||||
|
||||
@@ -26,6 +26,28 @@ from tests._vcr_conftest_common import ( # noqa: E402,F401
|
||||
vcr_config_dict,
|
||||
)
|
||||
|
||||
# Vertex AI MaaS Mistral OCR tests that cannot be VCR-cached in CI.
|
||||
#
|
||||
# ``vertex_ai/mistral-ocr-2505`` is a Model-as-a-Service partner model that
|
||||
# must be explicitly enabled in the GCP project's Model Garden. It is not
|
||||
# provisioned in the CI project (``litellm-ci-cd``), so the live
|
||||
# ``:rawPredict`` call fails on every run and ``BaseOCRTest`` catches the
|
||||
# provider error and skips. Because the doomed live call is recorded but the
|
||||
# test then skips, the persister refuses to save it (skipped tests don't
|
||||
# persist) and the cassette is never seeded — so the test re-records live and
|
||||
# is classified MISS:NOT_PERSISTED on every single run, forever. No cassette
|
||||
# can be recorded until the model is provisioned. Mark the tests VCR-
|
||||
# incompatible so they are honestly accounted as live calls (UNMARKED:LIVE_CALL)
|
||||
# rather than phantom cache misses; behaviour is unchanged (they still run and
|
||||
# still skip on the provider error). The sibling direct-Mistral and Azure OCR
|
||||
# tests replay from cache normally and are unaffected. Remove these entries if
|
||||
# the MaaS model is enabled in the CI project.
|
||||
_VCR_INCOMPATIBLE_NODEID_SUFFIXES: tuple[str, ...] = (
|
||||
"test_ocr_vertex_ai.py::TestVertexAIMistralOCR::test_ocr_response_structure",
|
||||
"test_ocr_vertex_ai.py::TestVertexAIMistralOCR::test_basic_ocr_with_url[True]",
|
||||
"test_ocr_vertex_ai.py::TestVertexAIMistralOCR::test_basic_ocr_with_url[False]",
|
||||
)
|
||||
|
||||
_verbose_state = VerboseReporterState()
|
||||
|
||||
|
||||
@@ -62,7 +84,10 @@ def pytest_runtest_logreport(report):
|
||||
|
||||
|
||||
def pytest_collection_modifyitems(config, items):
|
||||
apply_vcr_auto_marker_to_items(items)
|
||||
apply_vcr_auto_marker_to_items(
|
||||
items,
|
||||
skip_nodeid_suffixes=_VCR_INCOMPATIBLE_NODEID_SUFFIXES,
|
||||
)
|
||||
|
||||
|
||||
def pytest_terminal_summary(terminalreporter, exitstatus, config):
|
||||
|
||||
@@ -14,15 +14,22 @@ from tests._vcr_conftest_common import ( # noqa: E402
|
||||
KEY_FINGERPRINT_HEADER,
|
||||
KEY_FINGERPRINT_MATCHER_NAME,
|
||||
SAFE_BODY_MATCHER_NAME,
|
||||
TOLERANT_QUERY_MATCHER_NAME,
|
||||
_before_record_request,
|
||||
_is_credential_exchange_request,
|
||||
_is_telemetry_request,
|
||||
_key_fingerprint_matcher,
|
||||
_normalize_volatile_tokens,
|
||||
_safe_body_matcher,
|
||||
_tolerant_query_matcher,
|
||||
vcr_config_dict,
|
||||
)
|
||||
|
||||
|
||||
def _req(body):
|
||||
return SimpleNamespace(body=body, headers={"Content-Type": "application/json"})
|
||||
def _req(body, uri="https://api.openai.com/v1/chat/completions"):
|
||||
return SimpleNamespace(
|
||||
body=body, uri=uri, headers={"Content-Type": "application/json"}
|
||||
)
|
||||
|
||||
|
||||
def _req_with_headers(headers, body=b""):
|
||||
@@ -150,6 +157,132 @@ def test_before_record_request_is_deterministic_across_distinct_requests():
|
||||
)
|
||||
|
||||
|
||||
def test_google_oauth_bearer_tokens_collapse_to_one_fingerprint():
|
||||
"""Rotating ``ya29.*`` access tokens must share one fingerprint so
|
||||
Vertex/Gemini cassettes match across runs (cf. AWS SigV4 access-key
|
||||
stabilization)."""
|
||||
run1 = _before_record_request(
|
||||
_req_with_headers({"Authorization": "Bearer ya29.FIRST-token-aaaaaaaa"})
|
||||
)
|
||||
run2 = _before_record_request(
|
||||
_req_with_headers({"Authorization": "Bearer ya29.SECOND-token-bbbbbbbb"})
|
||||
)
|
||||
assert run1.headers[KEY_FINGERPRINT_HEADER] == run2.headers[KEY_FINGERPRINT_HEADER]
|
||||
_key_fingerprint_matcher(run1, run2)
|
||||
|
||||
|
||||
def test_non_google_bearer_tokens_still_distinguished():
|
||||
"""The ya29 collapse must not make every Bearer token identical."""
|
||||
google = _before_record_request(
|
||||
_req_with_headers({"Authorization": "Bearer ya29.something"})
|
||||
)
|
||||
real = _before_record_request(
|
||||
_req_with_headers({"Authorization": "Bearer sk-real-openai-key"})
|
||||
)
|
||||
assert (
|
||||
google.headers[KEY_FINGERPRINT_HEADER] != real.headers[KEY_FINGERPRINT_HEADER]
|
||||
)
|
||||
|
||||
|
||||
def test_normalize_volatile_tokens_collapses_uuid_and_timestamps():
|
||||
a = b'{"content": "news today b92ed205-0fa9-4e79-939c-2365023e9cb3"}'
|
||||
b = b'{"content": "news today 1a4e1afa-2915-4dcf-b043-33b991cae879"}'
|
||||
assert _normalize_volatile_tokens(a) == _normalize_volatile_tokens(b)
|
||||
|
||||
c = b'{"input": "embed data 1779581429.9713597"}'
|
||||
d = b'{"input": "embed data 1779583432.6874988"}'
|
||||
assert _normalize_volatile_tokens(c) == _normalize_volatile_tokens(d)
|
||||
|
||||
e = b'{"timestamp": "2026-05-25T03:40:37.262045Z"}'
|
||||
f = b'{"timestamp": "2026-05-25T06:10:20.830356Z"}'
|
||||
assert _normalize_volatile_tokens(e) == _normalize_volatile_tokens(f)
|
||||
|
||||
|
||||
def test_normalize_volatile_tokens_leaves_deterministic_bodies_unchanged():
|
||||
body = b'{"model":"claude-haiku-4-5-20251001","temperature":0.0,"n":2}'
|
||||
assert _normalize_volatile_tokens(body) == body
|
||||
|
||||
|
||||
def test_safe_body_matcher_matches_bodies_differing_only_by_cachebuster():
|
||||
a = _req(b'{"messages":[{"content":"hi 1779579395.5545585"}],"model":"gpt-4.1"}')
|
||||
b = _req(b'{"messages":[{"content":"hi 1779579663.595344"}],"model":"gpt-4.1"}')
|
||||
_safe_body_matcher(a, b) # must not raise
|
||||
|
||||
|
||||
def test_safe_body_matcher_still_rejects_genuinely_different_bodies():
|
||||
a = _req(b'{"messages":[{"content":"hello"}]}')
|
||||
b = _req(b'{"messages":[{"content":"goodbye"}]}')
|
||||
with pytest.raises(AssertionError):
|
||||
_safe_body_matcher(a, b)
|
||||
|
||||
|
||||
def test_credential_exchange_request_skips_body_comparison():
|
||||
assert _is_credential_exchange_request(
|
||||
_req(b"assertion=AAA", uri="https://oauth2.googleapis.com/token")
|
||||
)
|
||||
assert not _is_credential_exchange_request(
|
||||
_req(b"x", uri="https://api.openai.com/v1/chat/completions")
|
||||
)
|
||||
# Freshly-signed JWT assertions differ every run but must still match.
|
||||
a = _req(
|
||||
b"grant_type=x&assertion=eyJ0AAAA", uri="https://oauth2.googleapis.com/token"
|
||||
)
|
||||
b = _req(
|
||||
b"grant_type=x&assertion=eyJ0BBBB", uri="https://oauth2.googleapis.com/token"
|
||||
)
|
||||
_safe_body_matcher(a, b) # must not raise
|
||||
|
||||
|
||||
def test_match_on_uses_tolerant_query_not_builtin():
|
||||
cfg = vcr_config_dict()
|
||||
assert TOLERANT_QUERY_MATCHER_NAME in cfg["match_on"]
|
||||
assert "query" not in cfg["match_on"]
|
||||
|
||||
|
||||
def test_telemetry_request_detection():
|
||||
assert _is_telemetry_request(
|
||||
_req(b"x", uri="https://us.cloud.langfuse.com/api/public/ingestion")
|
||||
)
|
||||
assert _is_telemetry_request(_req(b"x", uri="https://otlp.arize.com/v1/traces"))
|
||||
assert not _is_telemetry_request(
|
||||
_req(b"x", uri="https://api.openai.com/v1/chat/completions")
|
||||
)
|
||||
|
||||
|
||||
def test_safe_body_matcher_skips_telemetry_body():
|
||||
a = _req(
|
||||
b'{"batch":[{"id":"aaa","timestamp":"2026-05-25T03:40:37Z"}]}',
|
||||
uri="https://us.cloud.langfuse.com/api/public/ingestion",
|
||||
)
|
||||
b = _req(
|
||||
b'{"batch":[{"id":"zzz","timestamp":"2026-05-25T09:99:99Z","extra":1}]}',
|
||||
uri="https://us.cloud.langfuse.com/api/public/ingestion",
|
||||
)
|
||||
_safe_body_matcher(a, b) # must not raise despite wholly different bodies
|
||||
|
||||
|
||||
def test_tolerant_query_skips_telemetry_but_enforces_others():
|
||||
from vcr.request import Request
|
||||
|
||||
def _greq(uri):
|
||||
return Request(method="GET", uri=uri, body=b"", headers={})
|
||||
|
||||
# Telemetry GET with a fresh trace_id in the query must still match.
|
||||
a = _greq(
|
||||
"https://us.cloud.langfuse.com/api/public/observations?traceId=litellm-test-AAA"
|
||||
)
|
||||
b = _greq(
|
||||
"https://us.cloud.langfuse.com/api/public/observations?traceId=litellm-test-BBB"
|
||||
)
|
||||
_tolerant_query_matcher(a, b) # must not raise
|
||||
|
||||
# Non-telemetry hosts keep vcrpy's strict query comparison.
|
||||
c = _greq("https://api.openai.com/v1/models?page=1")
|
||||
d = _greq("https://api.openai.com/v1/models?page=2")
|
||||
with pytest.raises(AssertionError):
|
||||
_tolerant_query_matcher(c, d)
|
||||
|
||||
|
||||
def test_before_record_request_is_idempotent_on_the_same_request_object():
|
||||
"""vcrpy invokes ``before_record_request`` more than once per request.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user