diff --git a/pyproject.toml b/pyproject.toml index 2866e27e84c..bfef331640c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -329,6 +329,7 @@ asyncio_mode = "auto" asyncio_default_fixture_loop_scope = "session" markers = [ "asyncio: mark test as an asyncio test", + "cassette_ttl(seconds): set the write-time lifetime for this test's VCR cassette", "limit_leaks: mark test with memory limit for leak detection (e.g., '40 MB')", "no_parallel: mark test to run sequentially (not in parallel) - typically for memory measurement tests", ] diff --git a/tests/_vcr_conftest_common.py b/tests/_vcr_conftest_common.py index 4d5a73779ea..cff0fa705ff 100644 --- a/tests/_vcr_conftest_common.py +++ b/tests/_vcr_conftest_common.py @@ -1,4 +1,4 @@ -"""Shared VCR (Redis-backed) plumbing imported by per-directory conftests. +"""Shared VCR cassette plumbing imported by per-directory conftests. See ``tests/llm_translation/Readme.md`` for the full design and ``tests/llm_translation/conftest.py`` for the reference wiring.""" @@ -15,18 +15,18 @@ import socket import sys import threading from collections import defaultdict -from typing import Iterable +from collections.abc import Iterable import pytest import vcr.matchers as _vcr_matchers +from tests._vcr_persister import make_persister, set_cassette_ttl_override from tests._vcr_redis_persister import ( MAX_EPISODES_PER_CASSETTE, VCR_VERBOSE_ENV, cassette_cache_capacity_snapshot, cassette_cache_health, filter_non_2xx_response, - make_redis_persister, mark_test_outcome_for_cassette, patch_vcrpy_aiohttp_record_path, ) @@ -1131,9 +1131,7 @@ def vcr_config_dict() -> dict: def vcr_disabled() -> bool: - if os.environ.get("LITELLM_VCR_DISABLE") == "1": - return True - return not os.environ.get("CASSETTE_REDIS_URL") + return os.environ.get("LITELLM_VCR_DISABLE") == "1" _atexit_banner_registered = False @@ -1183,7 +1181,7 @@ def register_persister_if_enabled(vcr) -> None: """Call from ``pytest_recording_configure(config, vcr)`` in each conftest.""" if vcr_disabled(): return - vcr.register_persister(make_redis_persister()) + vcr.register_persister(make_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) @@ -1788,6 +1786,9 @@ def record_vcr_outcome(request, vcr) -> None: test_passed = bool(rep_call and rep_call.passed) cassette_path = getattr(cassette, "_path", None) if cassette is not None else None if cassette_path: + ttl_marker = request.node.get_closest_marker("cassette_ttl") + if ttl_marker is not None: + set_cassette_ttl_override(cassette_path, ttl_marker.args[0]) mark_test_outcome_for_cassette(cassette_path, test_passed) nodeid = request.node.nodeid diff --git a/tests/_vcr_persister.py b/tests/_vcr_persister.py new file mode 100644 index 00000000000..a34a0355db3 --- /dev/null +++ b/tests/_vcr_persister.py @@ -0,0 +1,331 @@ +from __future__ import annotations + +import logging +import os +import tempfile +import warnings +from abc import ABC, abstractmethod +from collections.abc import Callable +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from vcr.persisters.filesystem import CassetteNotFoundError +from vcr.request import Request +from vcr.serialize import serialize +from vcr.serializers import compat + +CASSETTE_TTL_SECONDS = 24 * 60 * 60 +CASSETTE_TTL_SECONDS_ENV = "CASSETTE_TTL_SECONDS" +CASSETTE_BACKEND_ENV = "CASSETTE_BACKEND" +CASSETTE_REDIS_URL_ENV = "CASSETTE_REDIS_URL" +VCR_VERBOSE_ENV = "LITELLM_VCR_VERBOSE" +MAX_EPISODES_PER_CASSETTE = 50 + +TTLValue = int | str + +_log = logging.getLogger(__name__) +_passed_by_cassette_key: dict[str, bool] = {} +_ttl_by_cassette_key: dict[str, TTLValue] = {} + + +class VCRCassetteCacheWarning(UserWarning): + """Emitted when cassette persistence fails to load or save.""" + + +_cache_health = { + "save_failures": 0, + "save_failure_last_error": "", + "load_failures": 0, + "load_failure_last_error": "", +} + + +def _cassette_key(cassette_path: str) -> str: + return os.path.abspath(str(cassette_path)) + + +def _record_cache_failure(kind: str, exc: BaseException) -> None: + err = f"{type(exc).__name__}: {exc}" + if kind == "save": + _cache_health["save_failures"] = int(_cache_health["save_failures"]) + 1 + _cache_health["save_failure_last_error"] = err + elif kind == "load": + _cache_health["load_failures"] = int(_cache_health["load_failures"]) + 1 + _cache_health["load_failure_last_error"] = err + + +def cassette_cache_health() -> dict: + return dict(_cache_health) + + +def reset_cassette_cache_health() -> None: + _cache_health["save_failures"] = 0 + _cache_health["save_failure_last_error"] = "" + _cache_health["load_failures"] = 0 + _cache_health["load_failure_last_error"] = "" + + +def mark_test_outcome_for_cassette(cassette_path: str, passed: bool) -> None: + _passed_by_cassette_key[_cassette_key(cassette_path)] = passed + + +def set_cassette_ttl_override(cassette_path: str, ttl: TTLValue) -> None: + _ttl_by_cassette_key[_cassette_key(cassette_path)] = ttl + + +def _parse_ttl(value: TTLValue) -> int: + if isinstance(value, str) and value.lower() == "inf": + return 0 + return int(value) + + +def resolve_cassette_ttl(cassette_path: str, fallback: int | None = None) -> int: + key = _cassette_key(cassette_path) + override = _ttl_by_cassette_key.pop(key, None) + if override is not None: + return _parse_ttl(override) + env_value = os.environ.get(CASSETTE_TTL_SECONDS_ENV) + if env_value is not None: + return _parse_ttl(env_value) + return CASSETTE_TTL_SECONDS if fallback is None else fallback + + +class CorruptCassetteError(ValueError): + pass + + +class BaseCassettePersister(ABC): + backend_name = "cassette" + + def __init__(self, ttl_seconds: int | None = None) -> None: + self._fallback_ttl_seconds = ttl_seconds + + def load_cassette(self, cassette_path, serializer): + try: + return self._load(cassette_path, serializer) + except CassetteNotFoundError: + raise + except Exception as exc: + corrupt = isinstance(exc, CorruptCassetteError) + reported_exc = ( + exc.__cause__ if corrupt and exc.__cause__ is not None else exc + ) + _record_cache_failure("load", reported_exc) + detail = ( + "cached payload is corrupt, treating as cache miss" + if corrupt + else "treating as cache miss" + ) + msg = ( + f"VCR {self.backend_name} load failed for {cassette_path}; {detail}: " + f"{type(reported_exc).__name__}: {reported_exc}" + ) + _log.warning(msg) + warnings.warn(msg, VCRCassetteCacheWarning, stacklevel=2) + raise CassetteNotFoundError() from exc + + def save_cassette(self, cassette_path, cassette_dict, serializer): + key = _cassette_key(cassette_path) + passed = _passed_by_cassette_key.pop(key, True) + episode_count = len(cassette_dict.get("requests", []) or []) + if episode_count > MAX_EPISODES_PER_CASSETTE: + _ttl_by_cassette_key.pop(key, None) + _log.warning( + "VCR %s save refused for %s; cassette has %d episodes " + "(> MAX_EPISODES_PER_CASSETTE=%d). The test likely produces " + "non-deterministic request bodies (e.g. uuid) and is " + "appending instead of replaying. Opt it out with the " + "no-vcr list in conftest, or stabilize its request body.", + self.backend_name, + cassette_path, + episode_count, + MAX_EPISODES_PER_CASSETTE, + ) + return + if not passed: + _ttl_by_cassette_key.pop(key, None) + _log.info( + "VCR %s save skipped for %s; test did not pass — " + "leaving any prior cassette intact", + self.backend_name, + cassette_path, + ) + return + try: + ttl_seconds = resolve_cassette_ttl( + cassette_path, self._fallback_ttl_seconds + ) + self._save( + cassette_path, + cassette_dict, + serializer, + ttl_seconds=ttl_seconds, + ) + except Exception as exc: + _record_cache_failure("save", exc) + msg = ( + f"VCR {self.backend_name} save failed for {cassette_path}; cassette " + f"not persisted: {type(exc).__name__}: {exc}" + ) + _log.warning(msg) + warnings.warn(msg, VCRCassetteCacheWarning, stacklevel=2) + + def capacity_snapshot(self) -> dict | None: + return None + + @abstractmethod + def _load(self, cassette_path, serializer): + raise NotImplementedError + + @abstractmethod + def _save( + self, cassette_path, cassette_dict, serializer, *, ttl_seconds: int + ) -> None: + raise NotImplementedError + + +class FilesystemBackend(BaseCassettePersister): + backend_name = "filesystem" + + def __init__( + self, + ttl_seconds: int | None = None, + now: Callable[[], datetime] = lambda: datetime.now(timezone.utc), + replace: Callable[[str, str], None] = os.replace, + ) -> None: + super().__init__(ttl_seconds=ttl_seconds) + self._now = now + self._replace = replace + + def _load(self, cassette_path, serializer): + path = Path(cassette_path) + if not path.is_file(): + raise CassetteNotFoundError() + try: + data = serializer.deserialize(path.read_text(encoding="utf-8")) + except Exception as exc: + raise CorruptCassetteError(str(exc)) from exc + if not isinstance(data, dict): + raise CorruptCassetteError("cassette payload must be a mapping") + if "recorded_at" not in data: + raise CassetteNotFoundError() + try: + recorded_at = datetime.fromisoformat(data["recorded_at"]) + ttl_seconds = _parse_ttl(data["ttl_seconds"]) + interactions = data["interactions"] + except (KeyError, TypeError, ValueError) as exc: + raise CorruptCassetteError(str(exc)) from exc + if recorded_at.tzinfo is None: + raise CorruptCassetteError("recorded_at must include a timezone") + if ( + ttl_seconds > 0 + and (self._now() - recorded_at).total_seconds() > ttl_seconds + ): + raise CassetteNotFoundError() + try: + requests = [Request._from_dict(item["request"]) for item in interactions] + responses = [ + compat.convert_to_bytes(item["response"]) for item in interactions + ] + except (KeyError, TypeError, ValueError) as exc: + raise CorruptCassetteError(str(exc)) from exc + return requests, responses + + def _save( + self, cassette_path, cassette_dict, serializer, *, ttl_seconds: int + ) -> None: + path = Path(cassette_path) + path.parent.mkdir(parents=True, exist_ok=True) + normalized = serializer.deserialize(serialize(cassette_dict, serializer)) + data = { + "version": normalized["version"], + "recorded_at": self._now().isoformat(), + "ttl_seconds": ttl_seconds, + "interactions": normalized["interactions"], + } + payload = serializer.serialize(data) + temp_path: str | None = None + try: + with tempfile.NamedTemporaryFile( + mode="w", encoding="utf-8", dir=path.parent, delete=False + ) as temporary: + temporary.write(payload) + temp_path = temporary.name + self._replace(temp_path, str(path)) + finally: + if temp_path is not None and os.path.exists(temp_path): + os.unlink(temp_path) + + +def make_persister(): + backend = os.environ.get(CASSETTE_BACKEND_ENV) + if backend == "filesystem": + return FilesystemBackend() + if backend == "redis" or os.environ.get(CASSETTE_REDIS_URL_ENV): + from tests._vcr_redis_persister import make_redis_persister + + return make_redis_persister() + if backend is not None: + raise ValueError(f"Unsupported {CASSETTE_BACKEND_ENV}: {backend}") + return FilesystemBackend() + + +def cassette_cache_capacity_snapshot(client: Any | None = None) -> dict | None: + if client is not None: + from tests._vcr_redis_persister import make_redis_persister + + return make_redis_persister(client=client).capacity_snapshot() + return make_persister().capacity_snapshot() + + +def filter_non_2xx_response(response): + if not isinstance(response, dict): + return response + status = response.get("status") + code = status.get("code") if isinstance(status, dict) else status + if not isinstance(code, int): + return response + return response if 200 <= code < 300 else None + + +_PATCHED_AIOHTTP_RECORD = False + + +def patch_vcrpy_aiohttp_record_path() -> None: + global _PATCHED_AIOHTTP_RECORD + if _PATCHED_AIOHTTP_RECORD: + return + import vcr.stubs.aiohttp_stubs as _aiohttp_stubs + + _orig_record_response = _aiohttp_stubs.record_response + + async def _record_response_preserving_body(cassette, vcr_request, response): + await _orig_record_response(cassette, vcr_request, response) + body = getattr(response, "_body", None) or b"" + if body: + response.content.unread_data(body) + + _aiohttp_stubs.record_response = _record_response_preserving_body + _PATCHED_AIOHTTP_RECORD = True + + +def vcr_verbose_enabled() -> bool: + return os.environ.get(VCR_VERBOSE_ENV) == "1" + + +def format_vcr_verdict(cassette: Any) -> str: + if cassette is None: + return "[VCR NOOP]" + played = getattr(cassette, "play_count", 0) or 0 + dirty = getattr(cassette, "dirty", False) + total = len(cassette) if hasattr(cassette, "__len__") else 0 + if played == 0 and not dirty: + return "[VCR NOOP] (no http traffic)" + if played > 0 and not dirty: + return f"[VCR HIT] {played} replayed, 0 new ({total} cassette entries)" + if played == 0 and dirty: + return f"[VCR MISS] 0 replayed, recorded new ({total} cassette entries)" + return ( + f"[VCR PARTIAL] {played} replayed + new recordings ({total} cassette entries)" + ) diff --git a/tests/_vcr_redis_persister.py b/tests/_vcr_redis_persister.py index bb76d5fb1ee..12a85e76f65 100644 --- a/tests/_vcr_redis_persister.py +++ b/tests/_vcr_redis_persister.py @@ -1,96 +1,40 @@ from __future__ import annotations -import logging import os -import warnings -from typing import Any, Optional +from typing import Any from vcr.persisters.filesystem import CassetteNotFoundError from vcr.serialize import deserialize, serialize -CASSETTE_TTL_SECONDS = 24 * 60 * 60 +from tests._vcr_persister import ( + CASSETTE_BACKEND_ENV, + CASSETTE_REDIS_URL_ENV, + CASSETTE_TTL_SECONDS, + CASSETTE_TTL_SECONDS_ENV, + MAX_EPISODES_PER_CASSETTE, + VCR_VERBOSE_ENV, + BaseCassettePersister, + CorruptCassetteError, + FilesystemBackend, + VCRCassetteCacheWarning, + _cache_health, + _record_cache_failure, + cassette_cache_capacity_snapshot, + cassette_cache_health, + filter_non_2xx_response, + format_vcr_verdict, + make_persister, + mark_test_outcome_for_cassette, + patch_vcrpy_aiohttp_record_path, + reset_cassette_cache_health, + resolve_cassette_ttl, + set_cassette_ttl_override, + vcr_verbose_enabled, +) + REDIS_KEY_PREFIX = "litellm:vcr:cassette:" -CASSETTE_REDIS_URL_ENV = "CASSETTE_REDIS_URL" -VCR_VERBOSE_ENV = "LITELLM_VCR_VERBOSE" -MAX_EPISODES_PER_CASSETTE = 50 - _REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) -_log = logging.getLogger(__name__) -_passed_by_cassette_key: dict[str, bool] = {} - - -class VCRCassetteCacheWarning(UserWarning): - """Emitted when the cassette Redis cache fails to load or save. - - Surfaced in pytest's session-end warnings summary so failures are - visible in CI logs even when the underlying tests pass. - """ - - -# Per-process counters; surfaced via :func:`cassette_cache_health` so -# conftests can emit a session-end banner when failures occurred. -_cache_health = { - "save_failures": 0, - "save_failure_last_error": "", - "load_failures": 0, - "load_failure_last_error": "", -} - - -def _record_cache_failure(kind: str, exc: BaseException) -> None: - err = f"{type(exc).__name__}: {exc}" - if kind == "save": - _cache_health["save_failures"] = int(_cache_health["save_failures"]) + 1 - _cache_health["save_failure_last_error"] = err - elif kind == "load": - _cache_health["load_failures"] = int(_cache_health["load_failures"]) + 1 - _cache_health["load_failure_last_error"] = err - - -def cassette_cache_health() -> dict: - return dict(_cache_health) - - -def reset_cassette_cache_health() -> None: - _cache_health["save_failures"] = 0 - _cache_health["save_failure_last_error"] = "" - _cache_health["load_failures"] = 0 - _cache_health["load_failure_last_error"] = "" - - -def cassette_cache_capacity_snapshot(client: Optional[Any] = None) -> Optional[dict]: - """Probe Redis ``INFO memory`` and return used/max bytes and percent. - - Returns ``None`` if Redis is unreachable, the server didn't report - ``maxmemory``, or ``maxmemory`` is 0 (uncapped). Best-effort: any - exception turns into ``None`` so this never breaks a test session. - """ - try: - if client is None: - client = _build_default_client() - info = client.info(section="memory") - except Exception: # pragma: no cover - best-effort probe - return None - used = info.get("used_memory") - maxmem = info.get("maxmemory") - try: - used = int(used) if used is not None else None - maxmem = int(maxmem) if maxmem is not None else None - except (TypeError, ValueError): # pragma: no cover - defensive - return None - if not used or not maxmem or maxmem <= 0: - return None - return { - "used_memory_bytes": used, - "maxmemory_bytes": maxmem, - "used_pct": (used / maxmem) * 100.0, - } - - -def mark_test_outcome_for_cassette(cassette_path: str, passed: bool) -> None: - _passed_by_cassette_key[redis_key_for(cassette_path)] = passed - def redis_key_for(cassette_path: str) -> str: abs_path = os.path.abspath(str(cassette_path)) @@ -98,13 +42,12 @@ def redis_key_for(cassette_path: str) -> str: rel = os.path.relpath(abs_path, start=_REPO_ROOT) except ValueError: rel = os.path.basename(abs_path) - if rel.endswith(".yaml"): - rel = rel[: -len(".yaml")] + rel = rel.removesuffix(".yaml") rel = rel.replace("/cassettes/", "/").lstrip("./") return f"{REDIS_KEY_PREFIX}{rel}" -def _redis_url_from_env() -> Optional[str]: +def _redis_url_from_env() -> str | None: return os.environ.get(CASSETTE_REDIS_URL_ENV) or None @@ -132,154 +75,80 @@ def _build_default_client(): ) -def make_redis_persister( - client: Optional[Any] = None, - ttl_seconds: int = CASSETTE_TTL_SECONDS, -): +class RedisBackend(BaseCassettePersister): + backend_name = "redis" + + def __init__(self, client: Any, ttl_seconds: int | None = None) -> None: + super().__init__(ttl_seconds=ttl_seconds) + self._client = client + + def _load(self, cassette_path, serializer): + data = self._client.get(redis_key_for(cassette_path)) + if data is None: + raise CassetteNotFoundError() + try: + text = data.decode("utf-8") if isinstance(data, bytes) else data + return deserialize(text, serializer) + except Exception as exc: + raise CorruptCassetteError(str(exc)) from exc + + def _save( + self, cassette_path, cassette_dict, serializer, *, ttl_seconds: int + ) -> None: + data = serialize(cassette_dict, serializer) + payload = data.encode("utf-8") if isinstance(data, str) else data + key = redis_key_for(cassette_path) + if ttl_seconds <= 0: + self._client.set(key, payload) + return + self._client.set(key, payload, ex=ttl_seconds) + + def capacity_snapshot(self) -> dict | None: + try: + info = self._client.info(section="memory") + used = int(info["used_memory"]) + maxmem = int(info["maxmemory"]) + except Exception: + return None + if not used or not maxmem or maxmem <= 0: + return None + return { + "used_memory_bytes": used, + "maxmemory_bytes": maxmem, + "used_pct": (used / maxmem) * 100.0, + } + + +def make_redis_persister(client: Any | None = None, ttl_seconds: int | None = None): redis_client = client if client is not None else _build_default_client() - - try: - from redis.exceptions import RedisError - except ImportError: # pragma: no cover - redis is a hard test dep - RedisError = Exception # type: ignore[assignment,misc] - - class _RedisPersister: - @staticmethod - def load_cassette(cassette_path, serializer): - key = redis_key_for(cassette_path) - try: - data = redis_client.get(key) - except RedisError as exc: - _record_cache_failure("load", exc) - msg = ( - f"VCR redis load failed for {cassette_path}; treating " - f"as cache miss: {type(exc).__name__}: {exc}" - ) - _log.warning(msg) - warnings.warn(msg, VCRCassetteCacheWarning, stacklevel=2) - raise CassetteNotFoundError() from exc - if data is None: - raise CassetteNotFoundError() - try: - if isinstance(data, bytes): - data = data.decode("utf-8") - result = deserialize(data, serializer) - except Exception as exc: - _record_cache_failure("load", exc) - msg = ( - f"VCR redis load failed for {cassette_path}; cached " - f"payload is corrupt, treating as cache miss: " - f"{type(exc).__name__}: {exc}" - ) - _log.warning(msg) - warnings.warn(msg, VCRCassetteCacheWarning, stacklevel=2) - raise CassetteNotFoundError() from exc - # TTL is intentionally not refreshed on read. The cassette must - # lapse ``ttl_seconds`` after its last *write*, so the next run - # past that point re-records live and catches provider request or - # response contract drift instead of replaying a frozen response - # forever. Sliding the expiry forward on read would keep an - # actively-used cassette alive indefinitely and that drift check - # would never run. - return result - - @staticmethod - def save_cassette(cassette_path, cassette_dict, serializer): - key = redis_key_for(cassette_path) - passed = _passed_by_cassette_key.pop(key, True) - episode_count = len(cassette_dict.get("requests", []) or []) - if episode_count > MAX_EPISODES_PER_CASSETTE: - _log.warning( - "VCR redis save refused for %s; cassette has %d episodes " - "(> MAX_EPISODES_PER_CASSETTE=%d). The test likely produces " - "non-deterministic request bodies (e.g. uuid) and is " - "appending instead of replaying. Opt it out with the " - "no-vcr list in conftest, or stabilize its request body.", - cassette_path, - episode_count, - MAX_EPISODES_PER_CASSETTE, - ) - return - if not passed: - _log.info( - "VCR redis save skipped for %s; test did not pass — " - "leaving any prior cassette intact", - cassette_path, - ) - return - data = serialize(cassette_dict, serializer) - payload = data.encode("utf-8") if isinstance(data, str) else data - try: - redis_client.set(key, payload, ex=ttl_seconds) - except RedisError as exc: - # Cassette persistence is strictly best-effort: connection - # blips, timeouts, OOM at the maxmemory cap, READONLY - # replicas, etc. should all degrade gracefully to "test - # passed but cassette not cached" rather than failing the - # test on teardown. We still want a loud signal so the - # failure shows up in pytest's warnings summary at the - # end of the session and feeds the session-end banner. - _record_cache_failure("save", exc) - msg = ( - f"VCR redis save failed for {cassette_path}; cassette " - f"not persisted: {type(exc).__name__}: {exc}" - ) - _log.warning(msg) - warnings.warn(msg, VCRCassetteCacheWarning, stacklevel=2) - - return _RedisPersister + return RedisBackend(redis_client, ttl_seconds=ttl_seconds) -def filter_non_2xx_response(response): - if not isinstance(response, dict): - return response - status = response.get("status") - code = status.get("code") if isinstance(status, dict) else status - if not isinstance(code, int): - return response - return response if 200 <= code < 300 else None - - -_PATCHED_AIOHTTP_RECORD = False - - -def patch_vcrpy_aiohttp_record_path() -> None: - """Re-feed the response body into aiohttp's StreamReader after vcrpy's - record_response drains it, so downstream consumers (e.g. - LiteLLMAiohttpTransport.AiohttpResponseStream) can still read it.""" - global _PATCHED_AIOHTTP_RECORD - if _PATCHED_AIOHTTP_RECORD: - return - import vcr.stubs.aiohttp_stubs as _aiohttp_stubs - - _orig_record_response = _aiohttp_stubs.record_response - - async def _record_response_preserving_body(cassette, vcr_request, response): - await _orig_record_response(cassette, vcr_request, response) - body = getattr(response, "_body", None) or b"" - if body: - response.content.unread_data(body) - - _aiohttp_stubs.record_response = _record_response_preserving_body - _PATCHED_AIOHTTP_RECORD = True - - -def vcr_verbose_enabled() -> bool: - return os.environ.get(VCR_VERBOSE_ENV) == "1" - - -def format_vcr_verdict(cassette: Any) -> str: - if cassette is None: - return "[VCR NOOP]" - played = getattr(cassette, "play_count", 0) or 0 - dirty = getattr(cassette, "dirty", False) - total = len(cassette) if hasattr(cassette, "__len__") else 0 - if played == 0 and not dirty: - return "[VCR NOOP] (no http traffic)" - if played > 0 and not dirty: - return f"[VCR HIT] {played} replayed, 0 new ({total} cassette entries)" - if played == 0 and dirty: - return f"[VCR MISS] 0 replayed, recorded new ({total} cassette entries)" - return ( - f"[VCR PARTIAL] {played} replayed + new recordings ({total} cassette entries)" - ) +__all__ = [ + "CASSETTE_BACKEND_ENV", + "CASSETTE_REDIS_URL_ENV", + "CASSETTE_TTL_SECONDS", + "CASSETTE_TTL_SECONDS_ENV", + "MAX_EPISODES_PER_CASSETTE", + "VCR_VERBOSE_ENV", + "FilesystemBackend", + "RedisBackend", + "VCRCassetteCacheWarning", + "_build_default_client", + "_cache_health", + "_record_cache_failure", + "_redis_url_from_env", + "cassette_cache_capacity_snapshot", + "cassette_cache_health", + "filter_non_2xx_response", + "format_vcr_verdict", + "make_persister", + "make_redis_persister", + "mark_test_outcome_for_cassette", + "patch_vcrpy_aiohttp_record_path", + "redis_key_for", + "reset_cassette_cache_health", + "resolve_cassette_ttl", + "set_cassette_ttl_override", + "vcr_verbose_enabled", +] diff --git a/tests/llm_translation/Readme.md b/tests/llm_translation/Readme.md index 813c188ee7b..19514eec561 100644 --- a/tests/llm_translation/Readme.md +++ b/tests/llm_translation/Readme.md @@ -2,19 +2,20 @@ Unit tests for individual LLM providers. Name of the test file is the name of the LLM provider - e.g. `test_openai.py` is for OpenAI. -## Redis-backed VCR cache +## VCR cassette cache Every test in this directory is auto-decorated with `@pytest.mark.vcr` (via `conftest.py`). The first time a test runs we hit the live provider and -record the HTTP exchange into Redis under -`litellm:vcr:cassette:`. Every subsequent run within 24h replays -from Redis without touching the network. The 24h TTL means each new day's -first run records again, so upstream API drift surfaces within a day. +record the HTTP exchange into the configured cassette backend. Every subsequent +run within the cassette lifetime replays without touching the network. The +default 24h lifetime means each new day's first run records again, so upstream +API drift surfaces within a day -The persister, header scrubbing, and 2xx-only filtering are defined in -`tests/_vcr_redis_persister.py`. Files that already use `respx` (which -patches the same httpx transport vcrpy does) are excluded from the -auto-marker — see `_RESPX_CONFLICTING_FILES` in `conftest.py`. +The shared persister and filesystem backend are defined in +`tests/_vcr_persister.py`. The Redis backend and compatibility exports remain +in `tests/_vcr_redis_persister.py`. Files that already use `respx` (which +patches the same httpx transport vcrpy does) are excluded from the auto-marker, +see `_RESPX_CONFLICTING_FILES` in `conftest.py` The same VCR cache is used by other test directories that exercise live provider APIs. The reusable conftest plumbing lives in @@ -41,13 +42,40 @@ are intentionally not included: VCR.py patches the in-process httpx transport, so it cannot intercept the LLM calls that originate inside the Docker container. -### Required environment +### Backend and lifetime -`CASSETTE_REDIS_URL` — separate Redis instance from the application -Redis (`REDIS_URL`/`REDIS_HOST`) so test cassettes are not flushed by -proxy tests. Provider credentials (`ANTHROPIC_API_KEY`, `OPENAI_API_KEY`, -`AWS_*`, etc.) are needed only on cache-miss (the daily re-record), not -on replay. +| Variable | Meaning | Default | +| --- | --- | --- | +| `CASSETTE_BACKEND` | Explicitly selects `redis` or `filesystem` | Redis when `CASSETTE_REDIS_URL` is set, otherwise filesystem | +| `CASSETTE_REDIS_URL` | Dedicated Redis URL, separate from application Redis | Unset | +| `CASSETTE_TTL_SECONDS` | Write-time lifetime in seconds; `0`, negative values, and `inf` never expire | `86400` | +| `LITELLM_VCR_DISABLE` | Set to `1` to disable VCR | Unset | + +`CASSETTE_TTL_SECONDS` sets the lifetime for newly recorded cassettes and +defaults to `86400`. Values at or below zero, and `inf`, never expire. A test can +override the environment at write time: + +```python +@pytest.mark.cassette_ttl(3600) +def test_short_lived_recording(): + ... +``` + +Redis stores the lifetime using `SET EX` and keeps its existing key format, +`litellm:vcr:cassette:`. Filesystem cassettes use vcrpy's cassette path +and include `recorded_at` and `ttl_seconds` alongside `version` and +`interactions`. Provider credentials are needed only on a cache miss + +A filesystem cassette is self-describing: + +```yaml +version: 1 +recorded_at: '2026-09-02T12:00:00+00:00' +ttl_seconds: 86400 +interactions: +- request: {...} + response: {...} +``` ### Flushing the cache diff --git a/tests/llm_translation/test_vcr_conftest_common_banner.py b/tests/llm_translation/test_vcr_conftest_common_banner.py index 1c4395ef1a8..f03b281a066 100644 --- a/tests/llm_translation/test_vcr_conftest_common_banner.py +++ b/tests/llm_translation/test_vcr_conftest_common_banner.py @@ -154,7 +154,7 @@ def test_banner_silent_when_below_high_water_and_no_failures( def test_banner_silent_when_vcr_disabled( monkeypatch, health_reset, patch_capacity_snapshot ): - monkeypatch.delenv("CASSETTE_REDIS_URL", raising=False) + monkeypatch.setenv("LITELLM_VCR_DISABLE", "1") _cache_health["save_failures"] = 5 _cache_health["save_failure_last_error"] = "OutOfMemoryError: foo" patch_capacity_snapshot( diff --git a/tests/llm_translation/test_vcr_redis_persister.py b/tests/llm_translation/test_vcr_redis_persister.py index 236ed77522a..5fb5c3ffca2 100644 --- a/tests/llm_translation/test_vcr_redis_persister.py +++ b/tests/llm_translation/test_vcr_redis_persister.py @@ -2,6 +2,7 @@ from __future__ import annotations import os import sys +from datetime import datetime, timedelta, timezone import fakeredis import pytest @@ -14,6 +15,11 @@ from vcr.serializers import yamlserializer sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) +from tests._vcr_persister import ( # noqa: E402 + FilesystemBackend, + make_persister, + set_cassette_ttl_override, +) from tests._vcr_redis_persister import ( # noqa: E402 CASSETTE_TTL_SECONDS, MAX_EPISODES_PER_CASSETTE, @@ -445,3 +451,177 @@ def test_capacity_snapshot_swallows_exceptions(): raise RuntimeError("redis offline") assert cassette_cache_capacity_snapshot(client=_Boom()) is None + + +def test_filesystem_save_then_load_roundtrips_with_metadata(tmp_path): + now = datetime(2026, 9, 2, 12, 0, tzinfo=timezone.utc) + cassette_path = tmp_path / "nested" / "cassette.yaml" + persister = FilesystemBackend(now=lambda: now) + + persister.save_cassette(cassette_path, _sample_cassette_dict(), yamlserializer) + requests, responses = persister.load_cassette(cassette_path, yamlserializer) + payload = yamlserializer.deserialize(cassette_path.read_text()) + + assert payload["recorded_at"] == "2026-09-02T12:00:00+00:00" + assert payload["ttl_seconds"] == CASSETTE_TTL_SECONDS + assert requests[0].body.startswith(b'{"model":"claude"') + assert responses[0]["body"]["string"] == b'{"id":"msg_1","type":"message"}' + + +def test_filesystem_expired_cassette_is_a_miss(tmp_path): + recorded_at = datetime(2026, 9, 2, 12, 0, tzinfo=timezone.utc) + cassette_path = tmp_path / "expired.yaml" + writer = FilesystemBackend(ttl_seconds=60, now=lambda: recorded_at) + reader = FilesystemBackend(now=lambda: recorded_at + timedelta(seconds=61)) + writer.save_cassette(cassette_path, _sample_cassette_dict(), yamlserializer) + + with pytest.raises(CassetteNotFoundError): + reader.load_cassette(cassette_path, yamlserializer) + + +def test_filesystem_cassette_without_recorded_at_is_a_miss(tmp_path): + cassette_path = tmp_path / "legacy.yaml" + cassette_path.write_text( + yamlserializer.serialize({"version": 1, "interactions": []}) + ) + + with pytest.raises(CassetteNotFoundError): + FilesystemBackend().load_cassette(cassette_path, yamlserializer) + + +@pytest.mark.parametrize("ttl", [0, -1, "inf"]) +def test_filesystem_non_positive_and_infinite_ttl_never_expire(tmp_path, ttl): + recorded_at = datetime(2026, 9, 2, 12, 0, tzinfo=timezone.utc) + cassette_path = tmp_path / f"immortal-{ttl}.yaml" + set_cassette_ttl_override(cassette_path, ttl) + writer = FilesystemBackend(now=lambda: recorded_at) + reader = FilesystemBackend(now=lambda: recorded_at + timedelta(days=36500)) + writer.save_cassette(cassette_path, _sample_cassette_dict(), yamlserializer) + + requests, _ = reader.load_cassette(cassette_path, yamlserializer) + + assert len(requests) == 1 + + +def test_redis_ttl_override_controls_set_expiry(monkeypatch): + monkeypatch.setenv("CASSETTE_TTL_SECONDS", "7200") + fake, persister = _persister_with_fake_redis() + cassette_id = "tests/llm_translation/test_x/test_custom_ttl" + set_cassette_ttl_override(cassette_id, 3600) + + persister.save_cassette(cassette_id, _sample_cassette_dict(), yamlserializer) + + assert 3595 <= fake.ttl(redis_key_for(cassette_id)) <= 3600 + + +def test_redis_non_positive_ttl_uses_set_without_expiry(): + fake, persister = _persister_with_fake_redis() + cassette_id = "tests/llm_translation/test_x/test_immortal" + set_cassette_ttl_override(cassette_id, 0) + + persister.save_cassette(cassette_id, _sample_cassette_dict(), yamlserializer) + + assert fake.ttl(redis_key_for(cassette_id)) == -1 + + +def test_ttl_env_is_baked_into_filesystem_payload(tmp_path, monkeypatch): + monkeypatch.setenv("CASSETTE_TTL_SECONDS", "7200") + cassette_path = tmp_path / "env-ttl.yaml" + + FilesystemBackend().save_cassette( + cassette_path, _sample_cassette_dict(), yamlserializer + ) + + payload = yamlserializer.deserialize(cassette_path.read_text()) + assert payload["ttl_seconds"] == 7200 + + +def test_ttl_override_is_baked_into_filesystem_payload(tmp_path, monkeypatch): + monkeypatch.setenv("CASSETTE_TTL_SECONDS", "7200") + cassette_path = tmp_path / "marker-ttl.yaml" + set_cassette_ttl_override(cassette_path, 3600) + + FilesystemBackend().save_cassette( + cassette_path, _sample_cassette_dict(), yamlserializer + ) + + payload = yamlserializer.deserialize(cassette_path.read_text()) + assert payload["ttl_seconds"] == 3600 + + +def test_filesystem_corrupt_file_is_a_warned_miss(tmp_path, reset_health): + cassette_path = tmp_path / "corrupt.yaml" + cassette_path.write_text("not: [valid") + + with pytest.warns(VCRCassetteCacheWarning): + with pytest.raises(CassetteNotFoundError): + FilesystemBackend().load_cassette(cassette_path, yamlserializer) + + assert cassette_cache_health()["load_failures"] == 1 + + +def test_filesystem_atomic_write_preserves_existing_file_on_replace_failure( + tmp_path, reset_health +): + cassette_path = tmp_path / "atomic.yaml" + cassette_path.write_text("existing") + + def fail_replace(source, destination): + raise OSError("simulated replace failure") + + persister = FilesystemBackend(replace=fail_replace) + with pytest.warns(VCRCassetteCacheWarning, match="simulated replace failure"): + persister.save_cassette(cassette_path, _sample_cassette_dict(), yamlserializer) + + assert cassette_path.read_text() == "existing" + assert list(tmp_path.iterdir()) == [cassette_path] + + +def test_filesystem_failed_test_preserves_prior_cassette(tmp_path): + cassette_path = tmp_path / "failed.yaml" + persister = FilesystemBackend() + persister.save_cassette( + cassette_path, _sample_cassette_dict(), yamlserializer + ) + original = cassette_path.read_bytes() + mark_test_outcome_for_cassette(cassette_path, passed=False) + + persister.save_cassette( + cassette_path, _sample_cassette_dict(), yamlserializer + ) + + assert cassette_path.read_bytes() == original + + +def test_filesystem_episode_cap_preserves_prior_cassette(tmp_path): + cassette_path = tmp_path / "overflow.yaml" + persister = FilesystemBackend() + sample = _sample_cassette_dict() + persister.save_cassette(cassette_path, sample, yamlserializer) + original = cassette_path.read_bytes() + overflow = { + "requests": sample["requests"] * (MAX_EPISODES_PER_CASSETTE + 1), + "responses": sample["responses"] * (MAX_EPISODES_PER_CASSETTE + 1), + } + + persister.save_cassette(cassette_path, overflow, yamlserializer) + + assert cassette_path.read_bytes() == original + + +def test_filesystem_capacity_snapshot_is_none(): + assert FilesystemBackend().capacity_snapshot() is None + + +def test_factory_selects_filesystem_by_default(monkeypatch): + monkeypatch.delenv("CASSETTE_BACKEND", raising=False) + monkeypatch.delenv("CASSETTE_REDIS_URL", raising=False) + + assert isinstance(make_persister(), FilesystemBackend) + + +def test_factory_explicit_filesystem_wins_over_redis_url(monkeypatch): + monkeypatch.setenv("CASSETTE_BACKEND", "filesystem") + monkeypatch.setenv("CASSETTE_REDIS_URL", "redis://unused") + + assert isinstance(make_persister(), FilesystemBackend)