tiennm99/litellm
1
0
Fork 0
mirror of https://github.com/tiennm99/litellm.git synced 2026-07-15 06:19:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/litellm_internal_staging' into litellm_/funny-williams-dab711

Browse Source
This commit is contained in:
Yuneng Jiang
2026-05-14 14:01:08 -07:00
parent c82acbc0de 16bd81985d
commit d4be90cdad
23 changed files with 1765 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/_vcr_conftest_common.py
+765 -14
View File
@@ -5,25 +5,28 @@ See ``tests/llm_translation/Readme.md`` for the full design and
from __future__ import annotations
import ast
import atexit
import hashlib
import json
import os
import re
import socket
import sys
from collections import defaultdict
from typing import Iterable
import pytest
from tests._vcr_redis_persister import (
MAX_EPISODES_PER_CASSETTE,
VCR_VERBOSE_ENV,
cassette_cache_capacity_snapshot,
cassette_cache_health,
filter_non_2xx_response,
format_vcr_verdict,
make_redis_persister,
mark_test_outcome_for_cassette,
patch_vcrpy_aiohttp_record_path,
vcr_verbose_enabled,
)
CASSETTE_CACHE_HIGH_WATER_FRACTION = 0.85
@@ -231,6 +234,29 @@ def _iter_header_values(headers, name: str):
yield value
_AWS_SIGV4_CREDENTIAL_RE = re.compile(
r"AWS4-HMAC-SHA256\s+Credential=([^/\s,]+)/", re.IGNORECASE
)
def _stable_key_value(header_name: str, raw: str) -> str:
"""Return a *stable* identifier for a credential header.
For Bearer / API-key headers the entire value is stable across calls,
so we hash it as-is. For AWS SigV4 ``Authorization`` headers, only
the access-key portion of ``Credential=AKIA.../<DATE>/...`` is stable
— date, region, signed headers, and signature all rotate per request,
so hashing the full value would push every Bedrock request into a new
cassette episode. Extract just the access-key id when present.
"""
if header_name.lower() != "authorization":
return raw
match = _AWS_SIGV4_CREDENTIAL_RE.search(raw)
if match:
return f"aws-sigv4:{match.group(1)}"
return raw
def _compute_key_fingerprint(request) -> str:
headers = getattr(request, "headers", None)
parts: list[str] = []
@@ -242,7 +268,8 @@ def _compute_key_fingerprint(request) -> str:
text = text.strip()
if not text:
continue
parts.append(f"{header_name}={text}")
stable = _stable_key_value(header_name, text)
parts.append(f"{header_name}={stable}")
if not parts:
return "no-key"
digest = hashlib.sha256("\n".join(parts).encode("utf-8")).hexdigest()
@@ -470,6 +497,235 @@ def register_persister_if_enabled(vcr) -> None:
_atexit_banner_registered = True
VCR_SKIP_REASON_USER_ATTR = "vcr_skip_reason"
# Marker reasons recorded per-item / per-test for the session summary.
SKIP_REASON_RESPX = "respx_conflict"
SKIP_REASON_RESPX_MODULE = "respx_conflict_module"
SKIP_REASON_INCOMPATIBLE = "incompatible"
SKIP_REASON_FILE_OPT_OUT = "file_opt_out"
SKIP_REASON_DISABLED = "disabled"
SKIP_REASON_PRE_MARKED = "already_marked"
# Hostnames we consider an "expensive live call" if a non-VCR-marked test
# happens to hit them. Localhost/redis/databases are explicitly excluded.
_LIVE_CALL_HOST_SUFFIXES = (
".openai.com",
".anthropic.com",
".vertexai.googleapis.com",
".aiplatform.googleapis.com",
".googleapis.com",
".x.ai",
".cohere.ai",
".cohere.com",
".voyageai.com",
".perplexity.ai",
".mistral.ai",
".groq.com",
".huggingface.co",
".azure.com",
".tavily.com",
".serper.dev",
".searchapi.io",
".firecrawl.dev",
".exa.ai",
)
_LIVE_CALL_LOCAL_PREFIXES = (
"127.",
"localhost",
"::1",
"0.0.0.0",
"10.",
"172.16.",
"172.17.",
"172.18.",
"172.19.",
"172.20.",
"172.21.",
"172.22.",
"172.23.",
"172.24.",
"172.25.",
"172.26.",
"172.27.",
"172.28.",
"172.29.",
"172.30.",
"172.31.",
"192.168.",
)
class _RespxUsageVisitor(ast.NodeVisitor):
"""AST visitor that flags real respx wiring in a test module.
Substring scans of the source text are unreliable: a comment like
``# Previously used respx.mock`` or a docstring referencing respx
would falsely flag the module. We only count:
* ``@pytest.mark.respx`` / ``@respx.mock`` decorators
* ``with respx.mock(): ...`` context managers
* ``respx.mock(...)`` / ``respx.mock`` attribute access
* function parameters / fixture arguments named ``respx_mock``
"""
def __init__(self) -> None:
self.uses_respx = False
def _decorator_is_respx(self, dec: ast.expr) -> bool:
# ``@respx.mock`` (Attribute) or ``@respx.mock(...)`` (Call wrapping Attribute)
if isinstance(dec, ast.Call):
dec = dec.func
if isinstance(dec, ast.Attribute):
return (
isinstance(dec.value, ast.Name)
and dec.value.id == "respx"
and dec.attr == "mock"
)
return False
def _is_pytest_mark_respx(self, dec: ast.expr) -> bool:
# ``@pytest.mark.respx`` or ``@pytest.mark.respx(...)``.
if isinstance(dec, ast.Call):
dec = dec.func
if (
isinstance(dec, ast.Attribute)
and dec.attr == "respx"
and isinstance(dec.value, ast.Attribute)
and dec.value.attr == "mark"
and isinstance(dec.value.value, ast.Name)
and dec.value.value.id == "pytest"
):
return True
return False
def _check_decorators(self, decs: list[ast.expr]) -> None:
for d in decs:
if self._decorator_is_respx(d) or self._is_pytest_mark_respx(d):
self.uses_respx = True
def _check_args(self, args: ast.arguments) -> None:
# ``def test_foo(respx_mock): ...`` — pytest supplies the fixture
# whenever the parameter name appears, regardless of marker.
all_args = (
list(args.args)
+ list(args.kwonlyargs)
+ (list(args.posonlyargs) if hasattr(args, "posonlyargs") else [])
)
for a in all_args:
if a.arg == "respx_mock":
self.uses_respx = True
return
def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
self._check_decorators(node.decorator_list)
self._check_args(node.args)
self.generic_visit(node)
def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
self._check_decorators(node.decorator_list)
self._check_args(node.args)
self.generic_visit(node)
def visit_ClassDef(self, node: ast.ClassDef) -> None:
self._check_decorators(node.decorator_list)
self.generic_visit(node)
def _is_respx_mock_attr(self, node: ast.expr) -> bool:
return (
isinstance(node, ast.Attribute)
and isinstance(node.value, ast.Name)
and node.value.id == "respx"
and node.attr == "mock"
)
def visit_With(self, node: ast.With) -> None:
for item in node.items:
ctx = item.context_expr
if isinstance(ctx, ast.Call):
ctx = ctx.func
if self._is_respx_mock_attr(ctx):
self.uses_respx = True
self.generic_visit(node)
def visit_AsyncWith(self, node: ast.AsyncWith) -> None:
for item in node.items:
ctx = item.context_expr
if isinstance(ctx, ast.Call):
ctx = ctx.func
if self._is_respx_mock_attr(ctx):
self.uses_respx = True
self.generic_visit(node)
def visit_Call(self, node: ast.Call) -> None:
# ``respx.mock(...)`` invocation outside a ``with``/decorator —
# e.g. ``mock = respx.mock()`` at module scope.
if self._is_respx_mock_attr(node.func):
self.uses_respx = True
self.generic_visit(node)
def _module_uses_respx(item) -> bool:
"""Return True if the test's *module* actually wires up respx.
Uses an ``ast`` walk (not substring matching) so comments and
docstrings that mention respx don't count as real usage. A bare
``from respx import MockRouter`` import with no other respx
references therefore won't flag the module — that's exactly the
dead-import case this PR is trying to surface.
"""
module = getattr(item, "module", None)
src_file = getattr(module, "__file__", None) or str(getattr(item, "path", "") or "")
if not src_file or not os.path.isfile(src_file):
return False
try:
with open(src_file, encoding="utf-8") as f:
src = f.read()
except OSError:
return False
try:
tree = ast.parse(src, filename=src_file)
except SyntaxError:
# If the test file itself is broken, fall back to "no respx" —
# the test will fail collection on its own and we don't want
# the auto-marker to mask that with a misleading skip reason.
return False
visitor = _RespxUsageVisitor()
visitor.visit(tree)
return visitor.uses_respx
def _item_uses_respx(item) -> bool:
"""Return True if *this specific item* will trigger respx.
Two signals: the ``respx`` pytest marker, and the ``respx_mock``
fixture appearing in the item's resolved fixture chain. Either alone
causes vcrpy + respx to fight over the httpx transport.
"""
if item.get_closest_marker("respx") is not None:
return True
fixturenames = getattr(item, "fixturenames", None) or ()
if "respx_mock" in fixturenames:
return True
return False
# Cache the source-scan result so we don't reread each module per item.
_RESPX_MODULE_CACHE: dict[str, bool] = {}
def _module_path_uses_respx(item) -> bool:
src_file = str(getattr(item, "path", "") or "")
if not src_file:
return False
cached = _RESPX_MODULE_CACHE.get(src_file)
if cached is not None:
return cached
result = _module_uses_respx(item)
_RESPX_MODULE_CACHE[src_file] = result
return result
def apply_vcr_auto_marker_to_items(
items,
*,
@@ -478,26 +734,349 @@ def apply_vcr_auto_marker_to_items(
) -> None:
"""Auto-apply ``pytest.mark.vcr`` to collected items.
``skip_files`` are basenames to leave un-marked (e.g. respx-using
files, since respx and vcrpy both patch the httpx transport).
``skip_nodeid_suffixes`` are node-id suffixes for individual tests
that depend on live cross-call provider state.
Skip semantics (in priority order):
1. ``vcr_disabled()`` — global env-var off-switch (``LITELLM_VCR_DISABLE=1``
or no ``CASSETTE_REDIS_URL``).
2. Item already carries ``@pytest.mark.vcr`` — leave it alone.
3. Item triggers respx (per-item marker / fixture) — vcrpy and respx
both patch the httpx transport so applying both makes one silently
no-op. We tag the item ``vcr_skip_reason=respx_conflict``.
4. Module wires up respx anywhere — even tests in the file that don't
themselves use respx still inherit the patched transport when
respx fixtures activate at session level. Tagged
``respx_conflict_module``.
5. ``skip_files`` / ``skip_nodeid_suffixes`` opt-out lists from the
caller — used for tests that observe live cross-call provider state
(e.g. prompt-cache warmup) which deterministic replay can't model.
Tagged ``incompatible``.
Each skipped item gets a ``vcr_skip_reason`` attribute so the
session-end summary can show why it isn't cached.
"""
if vcr_disabled():
for item in items:
setattr(item, VCR_SKIP_REASON_USER_ATTR, SKIP_REASON_DISABLED)
return
skip_files = frozenset(skip_files)
skip_nodeid_suffixes = tuple(skip_nodeid_suffixes)
for item in items:
if item.get_closest_marker("vcr") is not None:
setattr(item, VCR_SKIP_REASON_USER_ATTR, SKIP_REASON_PRE_MARKED)
continue
if _item_uses_respx(item):
setattr(item, VCR_SKIP_REASON_USER_ATTR, SKIP_REASON_RESPX)
continue
filename = os.path.basename(str(item.path))
if filename in skip_files:
# Trust the caller's opt-out, but split by reason: if the
# module actually uses respx, label the conflict precisely so
# the summary surfaces dead respx imports vs. real conflicts.
if _module_path_uses_respx(item):
setattr(item, VCR_SKIP_REASON_USER_ATTR, SKIP_REASON_RESPX_MODULE)
else:
setattr(item, VCR_SKIP_REASON_USER_ATTR, SKIP_REASON_FILE_OPT_OUT)
continue
if any(item.nodeid.endswith(suffix) for suffix in skip_nodeid_suffixes):
continue
if item.get_closest_marker("vcr") is not None:
setattr(item, VCR_SKIP_REASON_USER_ATTR, SKIP_REASON_INCOMPATIBLE)
continue
item.add_marker(pytest.mark.vcr)
# ---------------------------------------------------------------------------
# Per-test stats accumulator + verdict classification.
#
# The session-end summary needs richer signal than the line-level verdict:
# - which tests overflowed ``MAX_EPISODES_PER_CASSETTE`` (cassette refused
# to save → live calls every CI run);
# - which tests fired live HTTP at a real LLM endpoint while VCR was not
# active for them (genuine wasted spend, not just "test mocked elsewhere");
# - skip-reason buckets so we can tell respx-conflict from
# incompatible-by-design from "module imports respx but never uses it".
# ---------------------------------------------------------------------------
# Verdict tags used in the per-test logline AND in the session summary
# breakdown.
VERDICT_HIT = "VCR HIT"
VERDICT_MISS_RECORDED = "VCR MISS:RECORDED"
VERDICT_MISS_OVERFLOW = "VCR MISS:OVERFLOW"
VERDICT_MISS_NOT_PERSISTED = "VCR MISS:NOT_PERSISTED"
VERDICT_PARTIAL = "VCR PARTIAL"
VERDICT_NOOP_NO_TRAFFIC = "VCR NOOP"
VERDICT_UNMARKED_LIVE_CALL = "VCR UNMARKED:LIVE_CALL"
VERDICT_UNMARKED_NO_TRAFFIC = "VCR UNMARKED:NO_TRAFFIC"
VERDICT_DISABLED = "VCR DISABLED"
# Per-session stats. Cleared by ``_reset_session_stats`` for unit tests.
_session_stats = {
"verdict_counts": defaultdict(int),
"overflow_tests": [], # list of nodeids
"unmarked_live_call_tests": [], # list of (nodeid, hosts)
"skip_reason_counts": defaultdict(int),
"skip_reason_examples": defaultdict(list),
}
def _reset_session_stats() -> None:
_session_stats["verdict_counts"].clear()
_session_stats["overflow_tests"].clear()
_session_stats["unmarked_live_call_tests"].clear()
_session_stats["skip_reason_counts"].clear()
_session_stats["skip_reason_examples"].clear()
# user_properties keys used to ship structured outcome data from xdist workers
# back to the controller. ``vcr_verdict`` is the human-readable line that
# ``VerboseReporterState.maybe_emit_verdict`` writes next to each test;
# ``vcr_outcome`` + ``vcr_recorded_by`` are the structured payload that
# ``aggregate_report_outcome`` folds into the controller's ``_session_stats``
# so the session-end summary actually has data in xdist mode.
_USER_PROP_VERDICT_LINE = "vcr_verdict"
_USER_PROP_OUTCOME = "vcr_outcome"
_USER_PROP_RECORDED_BY = "vcr_recorded_by"
def _emit_outcome_payload(
node,
verdict: str,
*,
skip_reason: str | None = None,
live_call_hosts: Iterable[str] | None = None,
) -> None:
"""Stash a structured VCR outcome on a pytest node so the xdist
controller can fold it into ``_session_stats``.
On a worker, ``record_vcr_outcome`` has already updated the worker-local
``_session_stats`` — but in xdist mode that state lives in the worker
process and never reaches the controller's ``pytest_terminal_summary``.
We use the report's ``user_properties`` channel (which xdist round-trips
back to the controller) to ship the outcome, and
``aggregate_report_outcome`` rebuilds the controller's stats from there.
The recorder tags ``vcr_recorded_by`` with ``PYTEST_XDIST_WORKER`` so
the controller can distinguish "recorded in this same main process —
already counted" from "recorded in a worker — needs aggregation here".
"""
node.user_properties.append(
(
_USER_PROP_OUTCOME,
{
"verdict": verdict,
"skip_reason": skip_reason,
"live_call_hosts": list(live_call_hosts) if live_call_hosts else [],
},
)
)
node.user_properties.append(
(_USER_PROP_RECORDED_BY, os.environ.get("PYTEST_XDIST_WORKER", ""))
)
def aggregate_report_outcome(report) -> None:
"""Fold a worker-produced VCR outcome into the controller's session stats.
No-op outside the xdist controller path:
* On a worker, ``_session_stats`` was already updated in-process by
``record_vcr_outcome`` — and the worker doesn't render the summary
anyway, so there's nothing for us to aggregate.
* In single-process mode, ``vcr_recorded_by`` is the empty string,
which means the same process that ran the test is now handling the
report — ``_session_stats`` already has the entry, double-counting
would be a bug.
* Only when ``vcr_recorded_by`` is a non-empty worker id (``"gw0"``
etc.) do we know the controller's ``_session_stats`` is missing this
test and needs the outcome folded in.
"""
if os.environ.get("PYTEST_XDIST_WORKER"):
return
if report.when != "teardown":
return
recorded_by = next(
(v for k, v in (report.user_properties or []) if k == _USER_PROP_RECORDED_BY),
None,
)
if not recorded_by:
return
outcome = next(
(v for k, v in (report.user_properties or []) if k == _USER_PROP_OUTCOME),
None,
)
if not outcome:
return
verdict = outcome.get("verdict")
if not verdict:
return
nodeid = report.nodeid
_session_stats["verdict_counts"][verdict] += 1
if verdict == VERDICT_MISS_OVERFLOW:
_session_stats["overflow_tests"].append(nodeid)
elif verdict == VERDICT_UNMARKED_LIVE_CALL:
_session_stats["unmarked_live_call_tests"].append(
(nodeid, list(outcome.get("live_call_hosts") or []))
)
skip_reason = outcome.get("skip_reason")
if skip_reason:
_session_stats["skip_reason_counts"][skip_reason] += 1
examples = _session_stats["skip_reason_examples"][skip_reason]
if len(examples) < 5:
examples.append(nodeid)
def session_stats_snapshot() -> dict:
"""Read-only copy of the per-session VCR stats. Used by the summary."""
return {
"verdict_counts": dict(_session_stats["verdict_counts"]),
"overflow_tests": list(_session_stats["overflow_tests"]),
"unmarked_live_call_tests": list(_session_stats["unmarked_live_call_tests"]),
"skip_reason_counts": dict(_session_stats["skip_reason_counts"]),
"skip_reason_examples": {
k: list(v) for k, v in _session_stats["skip_reason_examples"].items()
},
}
def _classify_marked_test(cassette) -> str:
"""Map cassette state → verdict tag for tests that *were* VCR-marked."""
played = getattr(cassette, "play_count", 0) or 0
dirty = getattr(cassette, "dirty", False)
total = len(cassette) if hasattr(cassette, "__len__") else 0
# "OVERFLOW" mirrors ``_RedisPersister.save_cassette``'s
# ``> MAX_EPISODES_PER_CASSETTE`` guard. Cassettes that hit this
# threshold are refused for save, so the test re-records live every
# run. Only flag when ``dirty=True`` — if a cassette grew past the
# cap historically but this run replayed it without adding new
# episodes, the persister never tries to save (no recording
# happened), so the cache state is stable and the next run will
# replay too. Flagging that case as OVERFLOW would tag healthy
# cached tests as cost leaks.
if total > MAX_EPISODES_PER_CASSETTE and dirty:
return VERDICT_MISS_OVERFLOW
if played == 0 and not dirty:
return VERDICT_NOOP_NO_TRAFFIC
if played > 0 and not dirty:
return VERDICT_HIT
if played == 0 and dirty:
return VERDICT_MISS_RECORDED
return VERDICT_PARTIAL
def _format_verdict_line(verdict: str, cassette, extra: str = "") -> str:
if cassette is None:
return f"[{verdict}]{(' ' + extra) if extra else ''}"
played = getattr(cassette, "play_count", 0) or 0
total = len(cassette) if hasattr(cassette, "__len__") else 0
base = f"[{verdict}] played={played} entries={total}"
if extra:
base = f"{base} {extra}"
return base
# ---------------------------------------------------------------------------
# Live-call detection for tests that bypass VCR.
#
# When a test isn't VCR-marked (respx_conflict, incompatible, or just
# plain unmarked), we wrap its socket calls inside the autouse
# ``_vcr_outcome_gate`` fixture so we can flag any outbound TCP connection
# to a known LLM provider. This converts "likely live call" into
# "confirmed: this test connected to host X".
# ---------------------------------------------------------------------------
_LIVE_CALL_BUFFER_KEY = "vcr_live_call_hosts"
def _is_live_call_host(host: str) -> bool:
if not host:
return False
host = host.lower()
if any(host.startswith(p) for p in _LIVE_CALL_LOCAL_PREFIXES):
return False
if any(host.endswith(suffix) for suffix in _LIVE_CALL_HOST_SUFFIXES):
return True
# AWS Bedrock endpoints are ``bedrock-runtime[-fips].{region}.amazonaws.com``
# (region between ``bedrock-runtime`` and ``amazonaws.com``), so plain
# suffix matching can't catch them.
if host.endswith(".amazonaws.com") and host.split(".", 1)[0].startswith(
"bedrock-runtime"
):
return True
return False
class _LiveCallProbe:
"""Context manager that monkeypatches ``socket.create_connection`` and
``socket.socket.connect`` for the lifetime of a test, recording any
outbound TCP connection to a known LLM host.
We don't intercept HTTP at the application layer because that would
fight with vcrpy/respx in tests that *do* mock httpx — the socket
layer is below both, so this probe is safe regardless of what's
patched above it. We also don't raise: the goal is observability, not
a hard gate.
"""
def __init__(self) -> None:
self.hosts: list[str] = []
self._orig_create_connection = None
self._orig_socket_connect = None
def __enter__(self):
self._orig_create_connection = socket.create_connection
self._orig_socket_connect = socket.socket.connect
def _wrapped_create_connection(address, *args, **kwargs):
try:
host = address[0] if isinstance(address, tuple) else None
if host and _is_live_call_host(host) and host not in self.hosts:
self.hosts.append(host)
except Exception:
pass
return self._orig_create_connection(address, *args, **kwargs)
def _wrapped_socket_connect(sock_self, address):
try:
host = address[0] if isinstance(address, tuple) else None
if host and _is_live_call_host(host) and host not in self.hosts:
self.hosts.append(host)
except Exception:
pass
return self._orig_socket_connect(sock_self, address)
socket.create_connection = _wrapped_create_connection
socket.socket.connect = _wrapped_socket_connect
return self
def __exit__(self, *exc):
if self._orig_create_connection is not None:
socket.create_connection = self._orig_create_connection
if self._orig_socket_connect is not None:
socket.socket.connect = self._orig_socket_connect
return False
def vcr_outcome_logging_enabled() -> bool:
"""Verdict logging is on whenever VCR itself is active.
The old ``LITELLM_VCR_VERBOSE=1`` gate kept logs quiet by default, but
that hides the very signal we need to know whether a paid test ran
against a real provider. CI logs already drop a one-line verdict per
test; that's what makes the cost analysis tractable. Set
``LITELLM_VCR_VERBOSE=0`` if you really want the legacy quiet mode.
"""
if vcr_disabled():
return False
if os.environ.get(VCR_VERBOSE_ENV) == "0":
return False
return True
def record_vcr_outcome(request, vcr) -> None:
"""Call from the post-yield section of an autouse fixture per test."""
cassette = vcr
@@ -507,10 +1086,78 @@ def record_vcr_outcome(request, vcr) -> None:
if cassette_path:
mark_test_outcome_for_cassette(cassette_path, test_passed)
if not vcr_verbose_enabled():
nodeid = request.node.nodeid
if cassette is not None:
verdict = _classify_marked_test(cassette)
# Track overflow tests even when verbose logging is off — the
# session summary shows them either way.
if verdict == VERDICT_MISS_OVERFLOW:
_session_stats["overflow_tests"].append(nodeid)
if not test_passed and verdict == VERDICT_MISS_RECORDED:
verdict = VERDICT_MISS_NOT_PERSISTED
_session_stats["verdict_counts"][verdict] += 1
_emit_outcome_payload(request.node, verdict)
if vcr_outcome_logging_enabled():
line = _format_verdict_line(verdict, cassette)
request.node.user_properties.append((_USER_PROP_VERDICT_LINE, line))
return
verdict = format_vcr_verdict(cassette)
request.node.user_properties.append(("vcr_verdict", verdict))
# Cassette is None ⇒ test wasn't VCR-marked. Honor the skip reason
# we tagged at collection time, and pull live-call hosts captured by
# the socket probe (if any).
skip_reason = getattr(
request.node, VCR_SKIP_REASON_USER_ATTR, SKIP_REASON_FILE_OPT_OUT
)
_session_stats["skip_reason_counts"][skip_reason] += 1
hosts = getattr(request.node, _LIVE_CALL_BUFFER_KEY, []) or []
if hosts:
verdict = VERDICT_UNMARKED_LIVE_CALL
_session_stats["unmarked_live_call_tests"].append((nodeid, list(hosts)))
extra = f"reason={skip_reason} hosts={','.join(hosts)}"
else:
verdict = VERDICT_UNMARKED_NO_TRAFFIC
extra = f"reason={skip_reason}"
_session_stats["verdict_counts"][verdict] += 1
examples = _session_stats["skip_reason_examples"][skip_reason]
if len(examples) < 5:
examples.append(nodeid)
_emit_outcome_payload(
request.node,
verdict,
skip_reason=skip_reason,
live_call_hosts=hosts,
)
if vcr_outcome_logging_enabled():
request.node.user_properties.append(
(_USER_PROP_VERDICT_LINE, _format_verdict_line(verdict, None, extra))
)
def install_live_call_probe(request, vcr) -> None:
"""Activate the live-call socket probe for non-VCR-marked tests.
Call this from inside the per-test autouse ``_vcr_outcome_gate``
fixture *before* the ``yield``. When ``vcr`` is ``None`` (test isn't
VCR-marked) we patch ``socket.connect`` for the duration of the test
and stash any LLM-host connections on ``request.node`` so
``record_vcr_outcome`` can include them in the verdict line.
Tests that *are* VCR-marked don't get the probe — vcrpy itself
intercepts above the socket layer, so any "outbound" socket would be
a recording cycle, not real spend.
"""
if vcr is not None or vcr_disabled():
return None
probe = _LiveCallProbe()
probe.__enter__()
setattr(request.node, _LIVE_CALL_BUFFER_KEY, probe.hosts)
request.addfinalizer(lambda: probe.__exit__(None, None, None))
return probe
def _format_capacity_line(snapshot: dict) -> str:
@@ -525,6 +1172,99 @@ def _format_capacity_line(snapshot: dict) -> str:
)
def emit_vcr_classification_summary(terminalreporter) -> None:
"""Render the per-classification summary at session end.
Output sections (only included when non-empty):
* **Verdict counts** — full breakdown of HIT / MISS:RECORDED /
MISS:OVERFLOW / MISS:NOT_PERSISTED / PARTIAL / NOOP /
UNMARKED:LIVE_CALL / UNMARKED:NO_TRAFFIC. The OVERFLOW and
UNMARKED:LIVE_CALL counts are the cost-leak signals.
* **Cassette overflow** (>``MAX_EPISODES_PER_CASSETTE``) — these tests
fire live every CI run because the persister refuses to save them.
Usually means the request body is non-deterministic (file handle
consumed, AWS SigV4 timestamp, random UUID).
* **Unmarked tests with live API calls** — confirmed live HTTP traffic
to a known LLM host while VCR was *not* active for the test. This
is the "convert likely → confirmed" signal: each entry is real
money the cache would otherwise prevent.
* **Skip-reason breakdown** — how many tests opted out of VCR and
why (respx_conflict, respx_conflict_module, file_opt_out,
incompatible). Bare ``file_opt_out`` entries with zero respx usage
in the module are dead skip-list rows worth pruning.
"""
if vcr_disabled():
return
if os.environ.get("PYTEST_XDIST_WORKER"):
return
snapshot = session_stats_snapshot()
counts = snapshot["verdict_counts"]
if not counts:
return
terminalreporter.write_sep("=", "VCR CACHE CLASSIFICATION SUMMARY", bold=True)
for verdict in (
VERDICT_HIT,
VERDICT_PARTIAL,
VERDICT_MISS_RECORDED,
VERDICT_MISS_OVERFLOW,
VERDICT_MISS_NOT_PERSISTED,
VERDICT_NOOP_NO_TRAFFIC,
VERDICT_UNMARKED_NO_TRAFFIC,
VERDICT_UNMARKED_LIVE_CALL,
):
n = counts.get(verdict, 0)
if not n:
continue
terminalreporter.write_line(f" [{verdict}] {n}")
overflow = snapshot["overflow_tests"]
if overflow:
terminalreporter.write_sep(
"-",
f"CASSETTE OVERFLOW (>{MAX_EPISODES_PER_CASSETTE} episodes, save refused)",
red=True,
bold=True,
)
terminalreporter.write_line(
" These tests will hit the live provider on every CI run "
"because the persister won't save cassettes that grew past "
"the limit. Stabilize the request body (file handle consumed, "
"SigV4 timestamp, UUID, or boundary leak)."
)
for nodeid in overflow:
terminalreporter.write_line(f" - {nodeid}")
live_calls = snapshot["unmarked_live_call_tests"]
if live_calls:
terminalreporter.write_sep(
"-",
"UNMARKED TESTS WITH LIVE API CALLS",
red=True,
bold=True,
)
terminalreporter.write_line(
" These tests connected to a real LLM provider host while "
"they were NOT VCR-marked. Either add @pytest.mark.vcr "
"explicitly, mock with respx, or move them off the "
"respx_conflict / incompatible skip list."
)
for nodeid, hosts in live_calls:
terminalreporter.write_line(f" - {nodeid} → {','.join(hosts)}")
reasons = snapshot["skip_reason_counts"]
if reasons:
terminalreporter.write_sep("-", "SKIP-REASON BREAKDOWN", bold=True)
for reason, n in sorted(reasons.items(), key=lambda kv: -kv[1]):
examples = snapshot["skip_reason_examples"].get(reason, [])
terminalreporter.write_line(f" {reason}: {n}")
for ex in examples:
terminalreporter.write_line(f" - {ex}")
terminalreporter.write_sep("=", bold=True)
def emit_cassette_cache_session_banner(terminalreporter) -> None:
"""Call from ``pytest_terminal_summary``. No-op on xdist workers."""
if vcr_disabled():
@@ -596,17 +1336,28 @@ class VerboseReporterState:
return self.terminal_reporter
def maybe_emit_verdict(self, report) -> None:
# Aggregate xdist-worker stats into the controller's session counters
# first — this path is independent of verbose logging because the
# structured outcome payload is always attached when VCR is active,
# and ``aggregate_report_outcome`` no-ops outside the xdist-controller
# case on its own.
aggregate_report_outcome(report)
if report.when != "teardown":
return
if os.environ.get("PYTEST_XDIST_WORKER"):
return
if not vcr_verbose_enabled():
if not vcr_outcome_logging_enabled():
return
reporter = self.resolve_terminal_reporter()
if reporter is None:
return
verdict = next(
(v for k, v in (report.user_properties or []) if k == "vcr_verdict"),
(
v
for k, v in (report.user_properties or [])
if k == _USER_PROP_VERDICT_LINE
),
None,
)
if not verdict:
tests/audio_tests/conftest.py
+9
View File
@@ -8,6 +8,9 @@ sys.path.insert(0, os.path.abspath("../.."))
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
@@ -34,6 +37,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@@ -48,3 +52,8 @@ def pytest_runtest_logreport(report):
def pytest_collection_modifyitems(config, items):
apply_vcr_auto_marker_to_items(items)
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)
tests/guardrails_tests/conftest.py
+9
View File
@@ -19,6 +19,9 @@ import litellm
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
@@ -45,6 +48,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@@ -151,3 +155,8 @@ def pytest_collection_modifyitems(config, items):
# Reorder the items list
items[:] = custom_logger_tests + other_tests
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)
tests/image_gen_tests/conftest.py
+9
View File
@@ -12,6 +12,9 @@ import litellm # noqa: E402,F401
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
@@ -48,6 +51,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@@ -62,3 +66,8 @@ def pytest_runtest_logreport(report):
def pytest_collection_modifyitems(config, items):
apply_vcr_auto_marker_to_items(items)
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)
tests/image_gen_tests/test_image_edits.py
+50 -26
View File
@@ -102,22 +102,47 @@ class BaseLLMImageEditTest(ABC):
# Get the current directory of the file being run
pwd = os.path.dirname(os.path.realpath(__file__))
TEST_IMAGES = [
open(os.path.join(pwd, "ishaan_github.png"), "rb"),
open(os.path.join(pwd, "litellm_site.png"), "rb"),
]
SINGLE_TEST_IMAGE = open(os.path.join(pwd, "ishaan_github.png"), "rb")
# Image fixtures must be regenerated per access — module-level
# ``open(...)`` handles get consumed after a single multipart upload, leaving
# subsequent tests in the same process to send empty bodies. That non-determinism
# (a) blows the recorded cassette past ``MAX_EPISODES_PER_CASSETTE`` so the
# persister refuses to save (see ``tests/_vcr_redis_persister.py``), and
# (b) re-bills the live image edit endpoint on every CI run.
def _read_image_bytes(filename: str) -> bytes:
with open(os.path.join(pwd, filename), "rb") as f:
return f.read()
_ISHAAN_GITHUB_BYTES = _read_image_bytes("ishaan_github.png")
_LITELLM_SITE_BYTES = _read_image_bytes("litellm_site.png")
def _make_test_images() -> list:
"""Return a fresh pair of image streams seeded with the fixture bytes.
Use this everywhere you'd previously have used the module-level
``TEST_IMAGES``. Each call returns brand new ``BytesIO`` objects whose
file pointers start at 0, so multipart uploads encode the full image
bytes on every test invocation. Parametrized and ``flaky``-retried
test methods call ``get_base_image_edit_call_args`` once per
invocation, so a fresh stream per call is sufficient — the factory
must not auto-rewind on EOF or the SDK's multipart writer will read
the same bytes forever (worker OOM).
"""
return [
BytesIO(_ISHAAN_GITHUB_BYTES),
BytesIO(_LITELLM_SITE_BYTES),
]
def _make_single_test_image() -> BytesIO:
return BytesIO(_ISHAAN_GITHUB_BYTES)
def get_test_images_as_bytesio():
"""Helper function to get test images as BytesIO objects"""
bytesio_images = []
for image_path in ["ishaan_github.png", "litellm_site.png"]:
with open(os.path.join(pwd, image_path), "rb") as f:
image_bytes = f.read()
bytesio_images.append(BytesIO(image_bytes))
return bytesio_images
return _make_test_images()
class TestOpenAIImageEditGPTImage1(BaseLLMImageEditTest):
@@ -129,7 +154,7 @@ class TestOpenAIImageEditGPTImage1(BaseLLMImageEditTest):
"""Return base call args for OpenAI image edit"""
return {
"model": "gpt-image-1",
"image": TEST_IMAGES,
"image": _make_test_images(),
}
@@ -143,7 +168,7 @@ class TestAzureAIFlux2ImageEdit(BaseLLMImageEditTest):
"""Return base call args for Azure AI FLUX 2 image edit"""
return {
"model": "azure_ai/flux.2-pro",
"image": SINGLE_TEST_IMAGE,
"image": _make_single_test_image(),
"api_base": os.getenv("AZURE_AI_API_BASE"),
"api_key": os.getenv("AZURE_AI_API_KEY"),
"api_version": "preview",
@@ -171,7 +196,7 @@ async def test_openai_image_edit_litellm_router():
result = await router.aimage_edit(
prompt=prompt,
model="gpt-image-1",
image=TEST_IMAGES,
image=_make_test_images(),
)
print("result from image edit", result)
@@ -275,7 +300,7 @@ async def test_azure_image_edit_litellm_sdk():
api_base=test_api_base,
api_key=test_api_key,
api_version=test_api_version,
image=TEST_IMAGES,
image=_make_test_images(),
)
# Verify the request was made correctly
@@ -389,7 +414,7 @@ async def test_openai_image_edit_cost_tracking():
result = await aimage_edit(
prompt=prompt,
model="openai/gpt-image-1",
image=TEST_IMAGES,
image=_make_test_images(),
)
# Verify the request was made correctly
@@ -480,7 +505,7 @@ async def test_azure_image_edit_cost_tracking():
prompt=prompt,
model="azure/CUSTOM_AZURE_DEPLOYMENT_NAME",
base_model="azure/gpt-image-1",
image=TEST_IMAGES,
image=_make_test_images(),
)
# Verify the request was made correctly
@@ -528,7 +553,6 @@ async def test_recraft_image_edit_api():
import requests
litellm._turn_on_debug()
global TEST_IMAGES
try:
prompt = """
Create a studio ghibli style image that combines all the reference images. Make sure the person looks like a CTO.
@@ -536,7 +560,7 @@ async def test_recraft_image_edit_api():
result = await aimage_edit(
prompt=prompt,
model="recraft/recraftv3",
image=TEST_IMAGES,
image=_make_test_images(),
)
print("result from image edit", result)
@@ -634,13 +658,13 @@ async def test_multiple_vs_single_image_edit(sync_mode):
single_result = image_edit(
prompt=prompt,
model="gpt-image-1",
image=SINGLE_TEST_IMAGE,
image=_make_single_test_image(),
)
else:
single_result = await aimage_edit(
prompt=prompt,
model="gpt-image-1",
image=SINGLE_TEST_IMAGE,
image=_make_single_test_image(),
)
print("Single image result:", single_result)
@@ -651,13 +675,13 @@ async def test_multiple_vs_single_image_edit(sync_mode):
multiple_result = image_edit(
prompt=prompt,
model="gpt-image-1",
image=TEST_IMAGES,
image=_make_test_images(),
)
else:
multiple_result = await aimage_edit(
prompt=prompt,
model="gpt-image-1",
image=TEST_IMAGES,
image=_make_test_images(),
)
print("Multiple images result:", multiple_result)
@@ -688,7 +712,7 @@ async def test_multiple_image_edit_with_different_formats():
# Test with mixed BytesIO and file objects
mixed_images = [
SINGLE_TEST_IMAGE, # File object
_make_single_test_image(), # File object
get_test_images_as_bytesio()[1], # BytesIO object
]
@@ -752,14 +776,14 @@ async def test_image_edit_array_handling():
result1 = await aimage_edit(
prompt=prompt,
model="gpt-image-1",
image=SINGLE_TEST_IMAGE,
image=_make_single_test_image(),
)
# Test 2: Multiple images (already a list)
result2 = await aimage_edit(
prompt=prompt,
model="gpt-image-1",
image=TEST_IMAGES,
image=_make_test_images(),
)
# Both valid calls should succeed
tests/litellm_utils_tests/conftest.py
+9
View File
@@ -15,6 +15,9 @@ import litellm # noqa: E402,F401
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
@@ -76,6 +79,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@@ -107,3 +111,8 @@ def pytest_collection_modifyitems(config, items):
# Reorder the items list
items[:] = custom_logger_tests + other_tests
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)
tests/llm_responses_api_testing/conftest.py
+9
View File
@@ -16,6 +16,9 @@ import litellm # noqa: E402
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
@@ -42,6 +45,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@@ -107,3 +111,8 @@ def pytest_collection_modifyitems(config, items):
other_tests.sort(key=lambda x: x.name)
items[:] = custom_logger_tests + other_tests
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)
tests/llm_translation/conftest.py
+15 -16
View File
@@ -21,27 +21,20 @@ import litellm # noqa: E402
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
)
# vcrpy and respx both patch the httpx transport — applying both makes one
# silently win, so respx-using files opt out of the auto-marker.
_RESPX_CONFLICTING_FILES = frozenset(
{
"test_gpt4o_audio.py",
"test_nvidia_nim.py",
"test_openai.py",
"test_openai_o1.py",
"test_prompt_caching.py",
"test_text_completion_unit_tests.py",
"test_xai.py",
}
)
_VCR_AUTO_MARKER_SKIP_FILES = _RESPX_CONFLICTING_FILES | frozenset(
{"test_vcr_redis_persister.py"}
)
# Per-item respx detection (``apply_vcr_auto_marker_to_items``) handles
# the vast majority of respx-vs-vcrpy conflicts automatically. The only
# entry below is the persister's own unit-test file, which exercises
# ``save_cassette`` / ``load_cassette`` against fakeredis and must not
# itself run under a live cassette context.
_VCR_AUTO_MARKER_SKIP_FILES = frozenset({"test_vcr_redis_persister.py"})
# Tests that observe live cross-call provider state (e.g. prompt-cache
# warm-up between two consecutive calls); replay can't reproduce that state.
@@ -73,6 +66,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@@ -85,6 +79,11 @@ def pytest_runtest_logreport(report):
_verbose_state.maybe_emit_verdict(report)
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)
# ---------------------------------------------------------------------------
# Capture TRUE defaults at conftest import time (before test modules pollute).
# ---------------------------------------------------------------------------
tests/llm_translation/test_gpt4o_audio.py
-1
View File
@@ -11,7 +11,6 @@ sys.path.insert(
import httpx
import pytest
from respx import MockRouter
import litellm
from litellm import Choices, Message, ModelResponse
tests/llm_translation/test_nvidia_nim.py
-1
View File
@@ -11,7 +11,6 @@ sys.path.insert(
import httpx
import pytest
from respx import MockRouter
from unittest.mock import patch, MagicMock, AsyncMock
import litellm
tests/llm_translation/test_openai.py
-1
View File
@@ -12,7 +12,6 @@ sys.path.insert(
import httpx
import pytest
from respx import MockRouter
import litellm
from litellm import Choices, Message, ModelResponse
tests/llm_translation/test_openai_o1.py
-1
View File
@@ -11,7 +11,6 @@ sys.path.insert(
import httpx
import pytest
from respx import MockRouter
import litellm
from litellm import Choices, Message, ModelResponse
tests/llm_translation/test_prompt_caching.py
-1
View File
@@ -11,7 +11,6 @@ sys.path.insert(
import httpx
import pytest
from respx import MockRouter
import litellm
from litellm import Choices, Message, ModelResponse
tests/llm_translation/test_vcr_classification.py
+804
View File
@@ -0,0 +1,804 @@
"""Unit tests for the VCR classification + observability layer.
Covers:
- per-item respx detection (module scan, marker, fixture)
- skip-reason tagging in ``apply_vcr_auto_marker_to_items``
- verdict classification (HIT / MISS:RECORDED / MISS:OVERFLOW / MISS:NOT_PERSISTED /
PARTIAL / NOOP / UNMARKED:LIVE_CALL / UNMARKED:NO_TRAFFIC)
- AWS SigV4 fingerprint stability
- session-end summary rendering
- live-call host classification
"""
from __future__ import annotations
import os
import sys
from types import SimpleNamespace
from typing import Optional
import pytest
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
from tests._vcr_conftest_common import ( # noqa: E402
SKIP_REASON_FILE_OPT_OUT,
SKIP_REASON_INCOMPATIBLE,
SKIP_REASON_PRE_MARKED,
SKIP_REASON_RESPX,
SKIP_REASON_RESPX_MODULE,
VCR_SKIP_REASON_USER_ATTR,
VERDICT_HIT,
VERDICT_MISS_NOT_PERSISTED,
VERDICT_MISS_OVERFLOW,
VERDICT_MISS_RECORDED,
VERDICT_NOOP_NO_TRAFFIC,
VERDICT_PARTIAL,
VERDICT_UNMARKED_LIVE_CALL,
VERDICT_UNMARKED_NO_TRAFFIC,
_RESPX_MODULE_CACHE,
_classify_marked_test,
_compute_key_fingerprint,
_is_live_call_host,
_reset_session_stats,
_stable_key_value,
aggregate_report_outcome,
apply_vcr_auto_marker_to_items,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
session_stats_snapshot,
)
# ---------------------------------------------------------------------------
# Test doubles
# ---------------------------------------------------------------------------
class _StubItem:
"""Pytest item double sufficient for the auto-marker logic."""
def __init__(
self,
nodeid: str,
path: str,
*,
markers: Optional[list[str]] = None,
fixturenames: Optional[list[str]] = None,
module=None,
) -> None:
self.nodeid = nodeid
self.path = path
self._markers = list(markers or [])
self.fixturenames = list(fixturenames or [])
self.module = module
self.user_properties: list = []
def get_closest_marker(self, name: str):
return name if name in self._markers else None
def add_marker(self, marker):
# ``pytest.mark.vcr`` is a MarkDecorator; rely on its ``name``.
name = getattr(marker, "name", str(marker))
self._markers.append(name)
@pytest.fixture
def vcr_enabled(monkeypatch):
monkeypatch.setenv("CASSETTE_REDIS_URL", "redis://stub")
monkeypatch.delenv("LITELLM_VCR_DISABLE", raising=False)
monkeypatch.delenv("PYTEST_XDIST_WORKER", raising=False)
@pytest.fixture(autouse=True)
def _reset_module_caches():
_reset_session_stats()
_RESPX_MODULE_CACHE.clear()
yield
_reset_session_stats()
_RESPX_MODULE_CACHE.clear()
# ---------------------------------------------------------------------------
# AWS SigV4 fingerprint stability — the Bedrock cassette overflow root cause
# ---------------------------------------------------------------------------
def test_should_extract_only_aws_access_key_from_sigv4_authorization():
"""Two Bedrock requests with the same access key but different
timestamps and signatures must produce the same fingerprint, otherwise
every CI run pushes a new episode into the cassette."""
auth_today = (
"AWS4-HMAC-SHA256 Credential=AKIAEXAMPLE12345/20260512/us-east-1/"
"bedrock/aws4_request, SignedHeaders=host;x-amz-date, "
"Signature=AAAAAAAA"
)
auth_tomorrow = (
"AWS4-HMAC-SHA256 Credential=AKIAEXAMPLE12345/20260513/us-east-1/"
"bedrock/aws4_request, SignedHeaders=host;x-amz-date, "
"Signature=BBBBBBBB"
)
today = _stable_key_value("Authorization", auth_today)
tomorrow = _stable_key_value("Authorization", auth_tomorrow)
assert today == tomorrow == "aws-sigv4:AKIAEXAMPLE12345"
def test_should_keep_bearer_authorization_unchanged():
"""OpenAI ``Bearer <key>`` headers are stable as-is — keep them."""
out = _stable_key_value("Authorization", "Bearer sk-1234")
assert out == "Bearer sk-1234"
def test_should_produce_stable_fingerprint_across_sigv4_signatures():
"""``_compute_key_fingerprint`` should not change when only the SigV4
signature/timestamp rotates."""
req_a = SimpleNamespace(
headers={
"authorization": (
"AWS4-HMAC-SHA256 Credential=AKIA1/20260101/us-east-1/"
"bedrock/aws4_request, SignedHeaders=host, Signature=AAA"
)
}
)
req_b = SimpleNamespace(
headers={
"authorization": (
"AWS4-HMAC-SHA256 Credential=AKIA1/20260512/us-east-1/"
"bedrock/aws4_request, SignedHeaders=host;x-amz-date, "
"Signature=ZZZ"
)
}
)
assert _compute_key_fingerprint(req_a) == _compute_key_fingerprint(req_b)
def test_should_distinguish_different_aws_access_keys():
"""Two different access keys must produce different fingerprints so
cassettes recorded under one identity never serve another."""
req_a = SimpleNamespace(
headers={
"authorization": "AWS4-HMAC-SHA256 Credential=AKIAONE/x/y/z/aws4_request, Signature=A"
}
)
req_b = SimpleNamespace(
headers={
"authorization": "AWS4-HMAC-SHA256 Credential=AKIATWO/x/y/z/aws4_request, Signature=A"
}
)
assert _compute_key_fingerprint(req_a) != _compute_key_fingerprint(req_b)
# ---------------------------------------------------------------------------
# Live-call host classification
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"host,expected",
[
("api.openai.com", True),
("api.anthropic.com", True),
("bedrock-runtime.us-east-1.amazonaws.com", True),
("bedrock-runtime-fips.us-east-1.amazonaws.com", True),
("api.us-east-1.bedrock-runtime.amazonaws.com", False),
("foo.bar.openai.com", True),
("127.0.0.1", False),
("localhost", False),
("10.0.0.1", False),
("172.16.0.1", False),
("redis.example.com", False),
("", False),
],
)
def test_should_classify_live_call_hosts(host, expected):
assert _is_live_call_host(host) is expected
# ---------------------------------------------------------------------------
# Verdict classification
# ---------------------------------------------------------------------------
def _cassette(played: int, dirty: bool, total: int):
class _Sized:
def __init__(self, n):
self.n = n
self.play_count = played
self.dirty = dirty
def __len__(self):
return self.n
return _Sized(total)
def test_should_classify_pure_replay_as_hit():
assert (
_classify_marked_test(_cassette(played=3, dirty=False, total=3)) == VERDICT_HIT
)
def test_should_classify_no_traffic_as_noop():
assert (
_classify_marked_test(_cassette(played=0, dirty=False, total=0))
== VERDICT_NOOP_NO_TRAFFIC
)
def test_should_classify_pure_record_as_miss_recorded():
assert (
_classify_marked_test(_cassette(played=0, dirty=True, total=1))
== VERDICT_MISS_RECORDED
)
def test_should_classify_mixed_replay_and_record_as_partial():
assert (
_classify_marked_test(_cassette(played=2, dirty=True, total=4))
== VERDICT_PARTIAL
)
def test_should_classify_overflow_only_when_dirty_episodes_were_recorded():
"""Cassettes that exceed ``MAX_EPISODES_PER_CASSETTE`` (50) are
refused for save — but only when ``dirty=True`` (new episodes were
actually recorded that the persister would refuse). Replaying an
already-large cassette with no new traffic is healthy: the persister
never tries to save, so the cache state is stable and the next run
will replay too."""
assert (
_classify_marked_test(_cassette(played=0, dirty=True, total=51))
== VERDICT_MISS_OVERFLOW
)
assert (
_classify_marked_test(_cassette(played=10, dirty=True, total=52))
== VERDICT_MISS_OVERFLOW
)
def test_should_classify_large_cassette_with_no_new_episodes_as_hit():
"""``total > 50`` + ``dirty=False`` means everything was replayed
from cache; no save attempt happens, so this is a healthy HIT, not
OVERFLOW."""
assert (
_classify_marked_test(_cassette(played=51, dirty=False, total=51))
== VERDICT_HIT
)
assert (
_classify_marked_test(_cassette(played=60, dirty=False, total=60))
== VERDICT_HIT
)
# ---------------------------------------------------------------------------
# apply_vcr_auto_marker_to_items: skip-reason tagging
# ---------------------------------------------------------------------------
def _make_module_with_source(tmp_path, src: str, name: str):
p = tmp_path / f"{name}.py"
p.write_text(src)
mod = SimpleNamespace(__file__=str(p))
return mod, str(p)
def test_should_apply_vcr_marker_to_clean_test(vcr_enabled, tmp_path):
mod, p = _make_module_with_source(tmp_path, "def test_x(): pass\n", "clean")
item = _StubItem("clean.py::test_x", p, module=mod)
apply_vcr_auto_marker_to_items([item])
assert item.get_closest_marker("vcr") == "vcr"
def test_should_skip_per_item_when_respx_marker_present(vcr_enabled, tmp_path):
mod, p = _make_module_with_source(tmp_path, "def test_x(): pass\n", "respx_marker")
item = _StubItem("respx_marker.py::test_x", p, markers=["respx"], module=mod)
apply_vcr_auto_marker_to_items([item])
assert item.get_closest_marker("vcr") is None
assert getattr(item, VCR_SKIP_REASON_USER_ATTR) == SKIP_REASON_RESPX
def test_should_skip_per_item_when_respx_mock_fixture_present(vcr_enabled, tmp_path):
mod, p = _make_module_with_source(tmp_path, "def test_x(): pass\n", "respx_fixture")
item = _StubItem(
"respx_fixture.py::test_x", p, fixturenames=["respx_mock"], module=mod
)
apply_vcr_auto_marker_to_items([item])
assert item.get_closest_marker("vcr") is None
assert getattr(item, VCR_SKIP_REASON_USER_ATTR) == SKIP_REASON_RESPX
def test_should_tag_pre_marked_items_so_summary_can_show_them(vcr_enabled, tmp_path):
mod, p = _make_module_with_source(tmp_path, "def test_x(): pass\n", "premarked")
item = _StubItem("premarked.py::test_x", p, markers=["vcr"], module=mod)
apply_vcr_auto_marker_to_items([item])
assert getattr(item, VCR_SKIP_REASON_USER_ATTR) == SKIP_REASON_PRE_MARKED
def test_should_tag_skip_files_with_respx_module_when_module_actually_uses_respx(
vcr_enabled, tmp_path
):
"""A file in ``skip_files`` whose module *does* call respx should be
labeled as a real conflict (respx_conflict_module), not a dead opt-out."""
mod, p = _make_module_with_source(
tmp_path,
"import respx\n@pytest.mark.respx\ndef test_x(): pass\n",
"real_respx",
)
item = _StubItem("real_respx.py::test_x", p, module=mod)
apply_vcr_auto_marker_to_items([item], skip_files={"real_respx.py"})
assert getattr(item, VCR_SKIP_REASON_USER_ATTR) == SKIP_REASON_RESPX_MODULE
def test_should_tag_skip_files_with_file_opt_out_when_module_does_not_use_respx(
vcr_enabled, tmp_path
):
"""A file in ``skip_files`` whose module never wires up respx is a
dead skip-list entry — surface it so we can prune."""
mod, p = _make_module_with_source(
tmp_path,
"from respx import MockRouter # dead import\ndef test_x(): pass\n",
"dead_skip",
)
item = _StubItem("dead_skip.py::test_x", p, module=mod)
apply_vcr_auto_marker_to_items([item], skip_files={"dead_skip.py"})
assert getattr(item, VCR_SKIP_REASON_USER_ATTR) == SKIP_REASON_FILE_OPT_OUT
def test_should_not_flag_respx_mentioned_in_comment_or_docstring(vcr_enabled, tmp_path):
"""Substring scans of source text false-positive on
``# Previously used respx.mock`` and similar — defeats the dead
skip-list pruning goal. AST-based detection ignores comments and
string literals."""
src = (
'"""Module docstring mentions respx.mock and @pytest.mark.respx and respx_mock."""\n'
"# Previously tried respx.mock but switched to vcrpy\n"
"# Old code did `with respx.mock(): ...`\n"
"x = '@respx.mock' # string literal, not a real decorator\n"
"def test_x():\n"
" pass\n"
)
mod, p = _make_module_with_source(tmp_path, src, "comment_respx")
item = _StubItem("comment_respx.py::test_x", p, module=mod)
apply_vcr_auto_marker_to_items([item], skip_files={"comment_respx.py"})
assert getattr(item, VCR_SKIP_REASON_USER_ATTR) == SKIP_REASON_FILE_OPT_OUT
def test_should_flag_real_respx_mark_decorator_via_ast(vcr_enabled, tmp_path):
src = "import pytest\n" "@pytest.mark.respx\n" "def test_x(respx_mock): pass\n"
mod, p = _make_module_with_source(tmp_path, src, "real_respx_mark")
item = _StubItem("real_respx_mark.py::test_x", p, module=mod)
apply_vcr_auto_marker_to_items([item], skip_files={"real_respx_mark.py"})
assert getattr(item, VCR_SKIP_REASON_USER_ATTR) == SKIP_REASON_RESPX_MODULE
def test_should_flag_real_respx_with_block_via_ast(vcr_enabled, tmp_path):
src = "import respx\n" "def test_x():\n" " with respx.mock():\n" " pass\n"
mod, p = _make_module_with_source(tmp_path, src, "real_respx_with")
item = _StubItem("real_respx_with.py::test_x", p, module=mod)
apply_vcr_auto_marker_to_items([item], skip_files={"real_respx_with.py"})
assert getattr(item, VCR_SKIP_REASON_USER_ATTR) == SKIP_REASON_RESPX_MODULE
def test_should_flag_respx_mock_call_at_module_scope_via_ast(vcr_enabled, tmp_path):
src = "import respx\nmock = respx.mock()\ndef test_x(): pass\n"
mod, p = _make_module_with_source(tmp_path, src, "real_respx_call")
item = _StubItem("real_respx_call.py::test_x", p, module=mod)
apply_vcr_auto_marker_to_items([item], skip_files={"real_respx_call.py"})
assert getattr(item, VCR_SKIP_REASON_USER_ATTR) == SKIP_REASON_RESPX_MODULE
def test_should_tag_nodeid_suffix_skips_as_incompatible(vcr_enabled, tmp_path):
mod, p = _make_module_with_source(tmp_path, "def test_x(): pass\n", "incompat")
item = _StubItem("incompat.py::test_prompt_caching", p, module=mod)
apply_vcr_auto_marker_to_items(
[item], skip_nodeid_suffixes=("::test_prompt_caching",)
)
assert getattr(item, VCR_SKIP_REASON_USER_ATTR) == SKIP_REASON_INCOMPATIBLE
# ---------------------------------------------------------------------------
# Session-end summary
# ---------------------------------------------------------------------------
class _FakeReporter:
def __init__(self):
self.lines: list[str] = []
def write_sep(self, sep, title="", **kwargs):
self.lines.append(f"=== {title}" if title else "===")
def write_line(self, line):
self.lines.append(line)
@property
def output(self):
return "\n".join(self.lines)
def test_should_render_overflow_section_when_any_test_overflowed(vcr_enabled):
"""The OVERFLOW section is the cost-leak signal: if it's empty, no
cassettes are silently being refused; if it's not empty, those tests
re-bill on every run."""
request = SimpleNamespace(
node=SimpleNamespace(
nodeid="t::overflow",
user_properties=[],
rep_call=SimpleNamespace(passed=True),
)
)
cassette = _cassette(played=0, dirty=True, total=51)
cassette._path = None # avoid mark_test_outcome side-effects
record_vcr_outcome(request, cassette)
reporter = _FakeReporter()
emit_vcr_classification_summary(reporter)
assert "VCR CACHE CLASSIFICATION SUMMARY" in reporter.output
assert "VCR MISS:OVERFLOW" in reporter.output
assert "CASSETTE OVERFLOW" in reporter.output
assert "t::overflow" in reporter.output
def test_should_render_unmarked_live_call_section_with_hosts(vcr_enabled):
request_node = SimpleNamespace(
nodeid="t::leak",
user_properties=[],
rep_call=SimpleNamespace(passed=True),
)
setattr(request_node, VCR_SKIP_REASON_USER_ATTR, SKIP_REASON_RESPX)
setattr(request_node, "vcr_live_call_hosts", ["api.openai.com"])
request = SimpleNamespace(node=request_node)
record_vcr_outcome(request, None)
snap = session_stats_snapshot()
assert snap["unmarked_live_call_tests"] == [("t::leak", ["api.openai.com"])]
assert snap["verdict_counts"][VERDICT_UNMARKED_LIVE_CALL] == 1
reporter = _FakeReporter()
emit_vcr_classification_summary(reporter)
assert "UNMARKED TESTS WITH LIVE API CALLS" in reporter.output
assert "api.openai.com" in reporter.output
assert "t::leak" in reporter.output
def test_should_record_unmarked_no_traffic_when_test_skipped_vcr_but_did_not_call_out(
vcr_enabled,
):
request_node = SimpleNamespace(
nodeid="t::clean_skip",
user_properties=[],
rep_call=SimpleNamespace(passed=True),
)
setattr(request_node, VCR_SKIP_REASON_USER_ATTR, SKIP_REASON_INCOMPATIBLE)
request = SimpleNamespace(node=request_node)
record_vcr_outcome(request, None)
snap = session_stats_snapshot()
assert snap["verdict_counts"][VERDICT_UNMARKED_NO_TRAFFIC] == 1
assert snap["skip_reason_counts"][SKIP_REASON_INCOMPATIBLE] == 1
def test_should_demote_miss_recorded_to_not_persisted_when_test_failed(vcr_enabled):
"""If a test failed, ``save_cassette`` skips persisting — that means
the next CI run will hit live again. The verdict must reflect that."""
request = SimpleNamespace(
node=SimpleNamespace(
nodeid="t::failed",
user_properties=[],
rep_call=SimpleNamespace(passed=False),
)
)
cassette = _cassette(played=0, dirty=True, total=1)
cassette._path = None
record_vcr_outcome(request, cassette)
snap = session_stats_snapshot()
assert snap["verdict_counts"].get(VERDICT_MISS_NOT_PERSISTED) == 1
def test_should_emit_no_summary_when_no_tests_observed(vcr_enabled):
reporter = _FakeReporter()
emit_vcr_classification_summary(reporter)
assert reporter.output == ""
# ---------------------------------------------------------------------------
# xdist controller aggregation
#
# _session_stats lives in module-global memory. Under xdist that memory is
# per-worker, so the controller's pytest_terminal_summary would render an
# empty summary without these aggregation hooks. The tests below simulate
# the controller receiving teardown reports produced by workers.
# ---------------------------------------------------------------------------
def _worker_report(nodeid: str, user_properties, *, when: str = "teardown"):
"""Stand-in for a pytest TestReport delivered to the xdist controller.
Only the attributes ``aggregate_report_outcome`` reads (``nodeid``,
``when``, ``user_properties``) are populated.
"""
return SimpleNamespace(
nodeid=nodeid,
when=when,
user_properties=list(user_properties),
)
def _outcome_from_worker(
verdict: str,
*,
worker_id: str = "gw0",
skip_reason=None,
live_call_hosts=None,
):
"""Build the ``user_properties`` list a worker-side ``record_vcr_outcome``
would attach. ``worker_id=""`` simulates the single-process case where
the same process that ran the test is handling the report."""
return [
(
"vcr_outcome",
{
"verdict": verdict,
"skip_reason": skip_reason,
"live_call_hosts": list(live_call_hosts) if live_call_hosts else [],
},
),
("vcr_recorded_by", worker_id),
]
def test_controller_aggregates_hit_outcome_from_worker_report(vcr_enabled):
"""An xdist controller starts with an empty _session_stats; a teardown
report carrying a worker-produced ``vcr_outcome`` must populate the
controller's verdict counts so the session summary has data to render."""
report = _worker_report(
"t::hit",
_outcome_from_worker(VERDICT_HIT),
)
aggregate_report_outcome(report)
snap = session_stats_snapshot()
assert snap["verdict_counts"][VERDICT_HIT] == 1
def test_controller_records_overflow_nodeid_from_worker_report(vcr_enabled):
"""OVERFLOW outcomes from workers must also populate
``overflow_tests`` (the named-list the summary surfaces)."""
report = _worker_report(
"t::bedrock_overflow",
_outcome_from_worker(VERDICT_MISS_OVERFLOW),
)
aggregate_report_outcome(report)
snap = session_stats_snapshot()
assert snap["verdict_counts"][VERDICT_MISS_OVERFLOW] == 1
assert snap["overflow_tests"] == ["t::bedrock_overflow"]
def test_controller_records_live_call_hosts_from_worker_report(vcr_enabled):
"""LIVE_CALL outcomes must round-trip the destination hosts so the
summary's 'UNMARKED TESTS WITH LIVE API CALLS' section has the same
detail it would in single-process mode."""
report = _worker_report(
"t::prompt_caching",
_outcome_from_worker(
VERDICT_UNMARKED_LIVE_CALL,
skip_reason=SKIP_REASON_INCOMPATIBLE,
live_call_hosts=["api.anthropic.com", "api.x.ai"],
),
)
aggregate_report_outcome(report)
snap = session_stats_snapshot()
assert snap["verdict_counts"][VERDICT_UNMARKED_LIVE_CALL] == 1
assert snap["unmarked_live_call_tests"] == [
("t::prompt_caching", ["api.anthropic.com", "api.x.ai"])
]
assert snap["skip_reason_counts"][SKIP_REASON_INCOMPATIBLE] == 1
assert "t::prompt_caching" in snap["skip_reason_examples"][SKIP_REASON_INCOMPATIBLE]
def test_controller_does_not_double_count_single_process_reports(vcr_enabled):
"""In single-process mode, ``record_vcr_outcome`` updates
``_session_stats`` in the same process that later handles the report.
The aggregator must detect this (via empty ``vcr_recorded_by``) and
skip — otherwise every verdict would be counted twice."""
report = _worker_report(
"t::single_proc",
_outcome_from_worker(VERDICT_HIT, worker_id=""),
)
aggregate_report_outcome(report)
snap = session_stats_snapshot()
assert snap["verdict_counts"] == {}
def test_controller_ignores_reports_without_vcr_outcome(vcr_enabled):
"""Tests outside the VCR plumbing (e.g. when VCR is disabled, or unit
tests that never went through ``_vcr_outcome_gate``) produce reports
with no ``vcr_outcome`` user property. The aggregator must no-op."""
report = _worker_report("t::unrelated", [("other", "value")])
aggregate_report_outcome(report)
snap = session_stats_snapshot()
assert snap["verdict_counts"] == {}
def test_controller_ignores_non_teardown_phases(vcr_enabled):
"""Only the teardown report carries the final outcome; setup/call
reports must not contribute to the counts."""
for phase in ("setup", "call"):
report = _worker_report(
"t::phase",
_outcome_from_worker(VERDICT_HIT),
when=phase,
)
aggregate_report_outcome(report)
snap = session_stats_snapshot()
assert snap["verdict_counts"] == {}
def test_controller_no_ops_when_running_inside_xdist_worker(vcr_enabled, monkeypatch):
"""Workers update their own ``_session_stats`` directly via
``record_vcr_outcome`` — re-aggregating from the report would
double-count their own work. The aggregator must bail when
``PYTEST_XDIST_WORKER`` is set."""
monkeypatch.setenv("PYTEST_XDIST_WORKER", "gw3")
report = _worker_report(
"t::on_worker",
_outcome_from_worker(VERDICT_HIT, worker_id="gw3"),
)
aggregate_report_outcome(report)
snap = session_stats_snapshot()
assert snap["verdict_counts"] == {}
def test_controller_aggregated_outcomes_drive_session_summary(vcr_enabled):
"""End-to-end: with only worker-produced reports (no in-process
``record_vcr_outcome``), the session-end summary must still render
the OVERFLOW + LIVE_CALL sections that prove the cost-leak signal
survived the xdist worker→controller hop."""
aggregate_report_outcome(
_worker_report(
"t::overflow_via_worker",
_outcome_from_worker(VERDICT_MISS_OVERFLOW),
)
)
aggregate_report_outcome(
_worker_report(
"t::live_call_via_worker",
_outcome_from_worker(
VERDICT_UNMARKED_LIVE_CALL,
skip_reason=SKIP_REASON_RESPX,
live_call_hosts=["api.openai.com"],
),
)
)
reporter = _FakeReporter()
emit_vcr_classification_summary(reporter)
assert "VCR CACHE CLASSIFICATION SUMMARY" in reporter.output
assert "CASSETTE OVERFLOW" in reporter.output
assert "t::overflow_via_worker" in reporter.output
assert "UNMARKED TESTS WITH LIVE API CALLS" in reporter.output
assert "api.openai.com" in reporter.output
assert "t::live_call_via_worker" in reporter.output
def test_record_vcr_outcome_emits_structured_payload_for_marked_tests(
vcr_enabled,
):
"""``record_vcr_outcome`` must always stash the structured outcome on
``user_properties`` (independent of verbose logging) so the controller
has something to aggregate from in xdist mode."""
request = SimpleNamespace(
node=SimpleNamespace(
nodeid="t::marked",
user_properties=[],
rep_call=SimpleNamespace(passed=True),
)
)
cassette = _cassette(played=1, dirty=False, total=1)
cassette._path = None
record_vcr_outcome(request, cassette)
outcomes = [v for k, v in request.node.user_properties if k == "vcr_outcome"]
recorded_by = [v for k, v in request.node.user_properties if k == "vcr_recorded_by"]
assert outcomes == [
{"verdict": VERDICT_HIT, "skip_reason": None, "live_call_hosts": []}
]
# No PYTEST_XDIST_WORKER set in the vcr_enabled fixture, so the
# recording-process tag is the empty string (single-process mode).
assert recorded_by == [""]
def test_record_vcr_outcome_emits_structured_payload_for_unmarked_live_call(
vcr_enabled,
):
"""The unmarked-LIVE_CALL path must ship the hosts list and the
skip-reason so the controller can rebuild both."""
request_node = SimpleNamespace(
nodeid="t::leak",
user_properties=[],
rep_call=SimpleNamespace(passed=True),
)
setattr(request_node, VCR_SKIP_REASON_USER_ATTR, SKIP_REASON_RESPX)
setattr(request_node, "vcr_live_call_hosts", ["api.openai.com"])
request = SimpleNamespace(node=request_node)
record_vcr_outcome(request, None)
outcomes = [v for k, v in request.node.user_properties if k == "vcr_outcome"]
assert outcomes == [
{
"verdict": VERDICT_UNMARKED_LIVE_CALL,
"skip_reason": SKIP_REASON_RESPX,
"live_call_hosts": ["api.openai.com"],
}
]
# ---------------------------------------------------------------------------
# Live-call probe
# ---------------------------------------------------------------------------
def test_should_skip_live_probe_when_vcr_active(vcr_enabled):
"""When the test *is* VCR-marked (cassette truthy), we don't install
the probe — vcrpy intercepts above the socket layer, so any
'connection' would be vcrpy's own bookkeeping and not real spend."""
request = SimpleNamespace(node=SimpleNamespace(), addfinalizer=lambda fn: None)
fake_cassette = SimpleNamespace(play_count=0, dirty=False)
probe = install_live_call_probe(request, fake_cassette)
assert probe is None
def test_live_call_probe_records_known_llm_hosts(vcr_enabled, monkeypatch):
"""The probe should record outbound TCP connections to known LLM
provider hosts (and ignore localhost / RFC1918 / unknown hosts)."""
finalizers = []
class _Node:
pass
request = SimpleNamespace(
node=_Node(), addfinalizer=lambda fn: finalizers.append(fn)
)
probe = install_live_call_probe(request, None)
assert probe is not None
import socket
# Manually invoke the patched function — we don't actually open a
# connection because that would hit the network. The probe records
# at the *call site* before delegating, and the original
# ``socket.create_connection`` will then fail; we swallow that.
try:
socket.create_connection(("api.openai.com", 443), timeout=0.001)
except Exception:
pass
try:
socket.create_connection(("127.0.0.1", 6379), timeout=0.001)
except Exception:
pass
# Restore via finalizers before asserting so the rest of the test
# session is unaffected.
for fn in finalizers:
fn()
hosts = getattr(request.node, "vcr_live_call_hosts", [])
assert "api.openai.com" in hosts
assert "127.0.0.1" not in hosts
tests/llm_translation/test_xai.py
-1
View File
@@ -11,7 +11,6 @@ sys.path.insert(
import httpx
import pytest
from respx import MockRouter
import litellm
from litellm import Choices, Message, ModelResponse, EmbeddingResponse, Usage
tests/local_testing/conftest.py
+17 -10
View File
@@ -25,20 +25,21 @@ import litellm
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
)
# vcrpy and respx both patch the httpx transport — applying both makes one
# silently win, so respx-using files opt out of the auto-marker.
_RESPX_CONFLICTING_FILES = frozenset(
{
"test_router.py",
"test_amazing_vertex_completion.py",
"test_azure_openai.py",
}
)
# Per-item respx detection (``apply_vcr_auto_marker_to_items``) auto-skips
# tests whose ``@pytest.mark.respx`` marker or ``respx_mock`` fixture
# would conflict with vcrpy's transport patch. We no longer maintain a
# file-level ``_RESPX_CONFLICTING_FILES`` list here — the previous
# entries (``test_router.py``) had only a stale ``from respx import
# MockRouter`` import with no actual respx wiring, so file-level
# blacklisting was masking valid cache opportunities.
# Files where VCR replay breaks the test:
# - ``test_assistants.py``: polls fresh per-session run IDs that no cassette
@@ -76,6 +77,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@@ -88,6 +90,11 @@ def pytest_runtest_logreport(report):
_verbose_state.maybe_emit_verdict(report)
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)
# ---------------------------------------------------------------------------
# Capture TRUE defaults at conftest import time. This runs before any test
# module's top-level code (e.g. `litellm.num_retries = 3`) executes, so
@@ -215,7 +222,7 @@ def setup_and_teardown():
def pytest_collection_modifyitems(config, items):
apply_vcr_auto_marker_to_items(
items,
skip_files=_RESPX_CONFLICTING_FILES | _VCR_INCOMPATIBLE_FILES,
skip_files=_VCR_INCOMPATIBLE_FILES,
skip_nodeid_suffixes=_VCR_INCOMPATIBLE_NODEID_SUFFIXES,
)
tests/local_testing/test_router.py
-1
View File
@@ -20,7 +20,6 @@ import os
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor
from unittest.mock import AsyncMock, MagicMock, patch
from respx import MockRouter
import httpx
from dotenv import load_dotenv
from pydantic import BaseModel
tests/logging_callback_tests/conftest.py
+9
View File
@@ -22,6 +22,9 @@ import litellm
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
@@ -69,6 +72,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@@ -220,3 +224,8 @@ def pytest_collection_modifyitems(config, items):
# Reorder the items list
items[:] = custom_logger_tests + other_tests
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)
tests/ocr_tests/conftest.py
+9
View File
@@ -15,6 +15,9 @@ sys.path.insert(0, os.path.abspath("../.."))
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
@@ -41,6 +44,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@@ -55,3 +59,8 @@ def pytest_runtest_logreport(report):
def pytest_collection_modifyitems(config, items):
apply_vcr_auto_marker_to_items(items)
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)
tests/pass_through_unit_tests/conftest.py
+24 -1
View File
@@ -8,11 +8,26 @@ sys.path.insert(0, os.path.abspath("../.."))
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
)
# Tests that observe live cross-call provider state — typically a
# warm-up call followed by an assertion that the *second* call sees the
# upstream's prompt-cache (Anthropic / Bedrock prompt-caching). VCR's
# deterministic replay can't model this: both calls match the same
# cassette episode, so the second call returns the first call's
# pre-warmup response. Opt these out so they run live (no caching).
_VCR_INCOMPATIBLE_NODEID_SUFFIXES = (
"::test_prompt_caching_returns_cache_read_tokens_on_second_call",
"::test_prompt_caching_streaming_second_call_returns_cache_read",
)
_verbose_state = VerboseReporterState()
@@ -34,6 +49,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@@ -47,4 +63,11 @@ 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):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)
tests/router_unit_tests/conftest.py
+9
View File
@@ -15,6 +15,9 @@ import litellm # noqa: E402,F401
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
@@ -87,6 +90,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@@ -114,3 +118,8 @@ def pytest_collection_modifyitems(config, items):
# Reorder the items list
items[:] = custom_logger_tests + other_tests
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)
tests/search_tests/conftest.py
+9
View File
@@ -16,6 +16,9 @@ sys.path.insert(0, os.path.abspath("../.."))
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
@@ -42,6 +45,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@@ -56,3 +60,8 @@ def pytest_runtest_logreport(report):
def pytest_collection_modifyitems(config, items):
apply_vcr_auto_marker_to_items(items)
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)
tests/unified_google_tests/conftest.py
+9
View File
@@ -15,6 +15,9 @@ import litellm # noqa: E402,F401
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
@@ -74,6 +77,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@@ -101,3 +105,8 @@ def pytest_collection_modifyitems(config, items):
# Reorder the items list
items[:] = custom_logger_tests + other_tests
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)