Merge pull request #26838 from BerriAI/litellm_vcr-cassette-llm-tests-af37
tests(vcr): redis-backed vcrpy cache for offline LLM e2e replay
This commit is contained in:
yuneng-jiang
GitHub
commit
d07cdd4481
3
Makefile
3
Makefile
@ -185,3 +185,6 @@ test-llm-translation-single: install-test-deps
|
||||
$(UV_RUN) pytest tests/llm_translation/$(FILE) \
|
||||
--junitxml=test-results/junit.xml \
|
||||
-v --tb=short --maxfail=100 --timeout=300
|
||||
|
||||
test-llm-translation-flush-vcr-cache:
|
||||
$(UV_RUN) python tests/_flush_vcr_cache.py
|
||||
|
||||
@ -77,8 +77,8 @@ def get_proxy_server_request_headers(litellm_params: Optional[dict]) -> dict:
|
||||
if litellm_params is None:
|
||||
return {}
|
||||
|
||||
proxy_request_headers = (
|
||||
litellm_params.get("proxy_server_request", {}).get("headers", {}) or {}
|
||||
)
|
||||
proxy_request_headers = (litellm_params.get("proxy_server_request") or {}).get(
|
||||
"headers"
|
||||
) or {}
|
||||
|
||||
return proxy_request_headers
|
||||
|
||||
@ -149,6 +149,8 @@ dev = [
|
||||
"parameterized==0.9.0",
|
||||
"openapi-core==0.22.0; python_version < '3.14'",
|
||||
"pytest-timeout==2.4.0",
|
||||
"vcrpy==8.1.1",
|
||||
"pytest-recording==0.13.4",
|
||||
]
|
||||
proxy-dev = [
|
||||
"prisma==0.11.0",
|
||||
|
||||
44
tests/_flush_vcr_cache.py
Normal file
44
tests/_flush_vcr_cache.py
Normal file
@ -0,0 +1,44 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
import redis
|
||||
|
||||
from tests._vcr_redis_persister import CASSETTE_REDIS_URL_ENV, _redis_url_from_env
|
||||
|
||||
PREFIX = "litellm:vcr:cassette:"
|
||||
SCAN_BATCH = 500
|
||||
|
||||
|
||||
def _client() -> redis.Redis:
|
||||
url = _redis_url_from_env()
|
||||
if not url:
|
||||
sys.exit(f"Set {CASSETTE_REDIS_URL_ENV} to flush the VCR cache")
|
||||
return redis.Redis.from_url(
|
||||
url,
|
||||
socket_timeout=5,
|
||||
socket_connect_timeout=5,
|
||||
decode_responses=False,
|
||||
)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
client = _client()
|
||||
deleted = 0
|
||||
pipeline = client.pipeline(transaction=False)
|
||||
pending = 0
|
||||
for key in client.scan_iter(match=f"{PREFIX}*", count=SCAN_BATCH):
|
||||
pipeline.delete(key)
|
||||
pending += 1
|
||||
if pending >= SCAN_BATCH:
|
||||
deleted += sum(pipeline.execute())
|
||||
pipeline = client.pipeline(transaction=False)
|
||||
pending = 0
|
||||
if pending:
|
||||
deleted += sum(pipeline.execute())
|
||||
print(f"Deleted {deleted} VCR cassette key(s) under {PREFIX!r}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
182
tests/_vcr_redis_persister.py
Normal file
182
tests/_vcr_redis_persister.py
Normal file
@ -0,0 +1,182 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
from typing import Any, Optional
|
||||
|
||||
from vcr.persisters.filesystem import CassetteNotFoundError
|
||||
from vcr.serialize import deserialize, serialize
|
||||
|
||||
CASSETTE_TTL_SECONDS = 24 * 60 * 60
|
||||
REDIS_KEY_PREFIX = "litellm:vcr:cassette:"
|
||||
CASSETTE_REDIS_URL_ENV = "CASSETTE_REDIS_URL"
|
||||
VCR_VERBOSE_ENV = "LITELLM_VCR_VERBOSE"
|
||||
MAX_EPISODES_PER_CASSETTE = 50
|
||||
|
||||
_log = logging.getLogger(__name__)
|
||||
_passed_by_cassette_key: dict[str, bool] = {}
|
||||
|
||||
|
||||
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:
|
||||
rel = os.path.relpath(str(cassette_path))
|
||||
if rel.endswith(".yaml"):
|
||||
rel = rel[: -len(".yaml")]
|
||||
rel = rel.replace("/cassettes/", "/").lstrip("./")
|
||||
return f"{REDIS_KEY_PREFIX}{rel}"
|
||||
|
||||
|
||||
def _redis_url_from_env() -> Optional[str]:
|
||||
return os.environ.get(CASSETTE_REDIS_URL_ENV) or None
|
||||
|
||||
|
||||
def _build_default_client():
|
||||
import redis
|
||||
from redis.backoff import ExponentialBackoff
|
||||
from redis.exceptions import ConnectionError as RedisConnectionError
|
||||
from redis.exceptions import TimeoutError as RedisTimeoutError
|
||||
from redis.retry import Retry
|
||||
|
||||
url = _redis_url_from_env()
|
||||
if not url:
|
||||
raise RuntimeError(
|
||||
f"Set {CASSETTE_REDIS_URL_ENV} to enable the VCR persister. "
|
||||
"Cassette Redis is intentionally separate from the application "
|
||||
"Redis (REDIS_URL/REDIS_HOST) to avoid being flushed by tests."
|
||||
)
|
||||
return redis.Redis.from_url(
|
||||
url,
|
||||
socket_timeout=5,
|
||||
socket_connect_timeout=5,
|
||||
decode_responses=False,
|
||||
retry=Retry(ExponentialBackoff(cap=2, base=0.1), retries=2),
|
||||
retry_on_error=[RedisConnectionError, RedisTimeoutError],
|
||||
)
|
||||
|
||||
|
||||
def make_redis_persister(
|
||||
client: Optional[Any] = None,
|
||||
ttl_seconds: int = CASSETTE_TTL_SECONDS,
|
||||
):
|
||||
redis_client = client if client is not None else _build_default_client()
|
||||
|
||||
try:
|
||||
from redis.exceptions import ConnectionError as RedisConnectionError
|
||||
from redis.exceptions import TimeoutError as RedisTimeoutError
|
||||
|
||||
_transient_errors: tuple = (RedisConnectionError, RedisTimeoutError)
|
||||
except ImportError: # pragma: no cover - redis is a hard test dep
|
||||
_transient_errors = ()
|
||||
|
||||
class _RedisPersister:
|
||||
@staticmethod
|
||||
def load_cassette(cassette_path, serializer):
|
||||
try:
|
||||
data = redis_client.get(redis_key_for(cassette_path))
|
||||
except _transient_errors as exc:
|
||||
_log.warning(
|
||||
"VCR redis load failed for %s; treating as cache miss: %s",
|
||||
cassette_path,
|
||||
exc,
|
||||
)
|
||||
raise CassetteNotFoundError() from exc
|
||||
if data is None:
|
||||
raise CassetteNotFoundError()
|
||||
if isinstance(data, bytes):
|
||||
data = data.decode("utf-8")
|
||||
return deserialize(data, serializer)
|
||||
|
||||
@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 _transient_errors as exc:
|
||||
_log.warning(
|
||||
"VCR redis save failed for %s; cassette not persisted: %s",
|
||||
cassette_path,
|
||||
exc,
|
||||
)
|
||||
|
||||
return _RedisPersister
|
||||
|
||||
|
||||
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)"
|
||||
)
|
||||
@ -169,3 +169,4 @@ langchain-mcp-adapters: >=0.2.1 # MIT License
|
||||
langgraph: >=1.0.10 # MIT License
|
||||
langgraph-prebuilt: >=1.0.8 # MIT License - https://github.com/langchain-ai/langgraph/blob/main/LICENSE
|
||||
pytest-rerunfailures: >=15.1 # MPL 2.0 license
|
||||
pytest-recording: >=0.13.4 # MIT license
|
||||
|
||||
@ -1,5 +1,6 @@
|
||||
# conftest.py
|
||||
|
||||
import asyncio
|
||||
import importlib
|
||||
import os
|
||||
import sys
|
||||
@ -9,9 +10,161 @@ import pytest
|
||||
sys.path.insert(
|
||||
0, os.path.abspath("../..")
|
||||
) # Adds the parent directory to the system path
|
||||
import litellm
|
||||
|
||||
import asyncio
|
||||
import litellm # noqa: E402
|
||||
|
||||
from tests._vcr_redis_persister import ( # noqa: E402
|
||||
filter_non_2xx_response,
|
||||
format_vcr_verdict,
|
||||
make_redis_persister,
|
||||
mark_test_outcome_for_cassette,
|
||||
patch_vcrpy_aiohttp_record_path,
|
||||
vcr_verbose_enabled,
|
||||
)
|
||||
|
||||
|
||||
_controller_pluginmanager = None
|
||||
_controller_terminal_reporter = None
|
||||
|
||||
|
||||
_FILTERED_REQUEST_HEADERS = (
|
||||
"authorization",
|
||||
"x-api-key",
|
||||
"anthropic-api-key",
|
||||
"anthropic-version",
|
||||
"openai-api-key",
|
||||
"azure-api-key",
|
||||
"api-key",
|
||||
"cookie",
|
||||
"x-amz-security-token",
|
||||
"x-amz-date",
|
||||
"x-amz-content-sha256",
|
||||
"amz-sdk-invocation-id",
|
||||
"amz-sdk-request",
|
||||
"x-goog-api-key",
|
||||
"x-goog-user-project",
|
||||
)
|
||||
|
||||
_FILTERED_RESPONSE_HEADERS = (
|
||||
"set-cookie",
|
||||
"x-request-id",
|
||||
"request-id",
|
||||
"cf-ray",
|
||||
"anthropic-organization-id",
|
||||
"openai-organization",
|
||||
"x-amzn-requestid",
|
||||
"x-amzn-trace-id",
|
||||
"date",
|
||||
)
|
||||
|
||||
|
||||
def _scrub_response(response):
|
||||
if not isinstance(response, dict):
|
||||
return response
|
||||
headers = response.get("headers") or {}
|
||||
if isinstance(headers, dict):
|
||||
for header in list(headers):
|
||||
if header.lower() in _FILTERED_RESPONSE_HEADERS:
|
||||
headers.pop(header, None)
|
||||
return response
|
||||
|
||||
|
||||
def _before_record_response(response):
|
||||
return filter_non_2xx_response(_scrub_response(response))
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def vcr_config():
|
||||
return {
|
||||
"filter_headers": list(_FILTERED_REQUEST_HEADERS),
|
||||
"decode_compressed_response": True,
|
||||
"record_mode": "new_episodes",
|
||||
"allow_playback_repeats": True,
|
||||
"match_on": (
|
||||
"method",
|
||||
"scheme",
|
||||
"host",
|
||||
"port",
|
||||
"path",
|
||||
"query",
|
||||
"body",
|
||||
),
|
||||
"before_record_response": _before_record_response,
|
||||
}
|
||||
|
||||
|
||||
def _vcr_disabled() -> bool:
|
||||
if os.environ.get("LITELLM_VCR_DISABLE") == "1":
|
||||
return True
|
||||
return not os.environ.get("CASSETTE_REDIS_URL")
|
||||
|
||||
|
||||
def pytest_recording_configure(config, vcr):
|
||||
if _vcr_disabled():
|
||||
return
|
||||
vcr.register_persister(make_redis_persister())
|
||||
patch_vcrpy_aiohttp_record_path()
|
||||
|
||||
|
||||
@pytest.hookimpl(hookwrapper=True)
|
||||
def pytest_runtest_makereport(item, call):
|
||||
outcome = yield
|
||||
rep = outcome.get_result()
|
||||
setattr(item, f"rep_{rep.when}", rep)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _vcr_outcome_gate(request, vcr):
|
||||
yield
|
||||
cassette = vcr
|
||||
rep_call = getattr(request.node, "rep_call", 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:
|
||||
mark_test_outcome_for_cassette(cassette_path, test_passed)
|
||||
|
||||
if not vcr_verbose_enabled():
|
||||
return
|
||||
verdict = format_vcr_verdict(cassette)
|
||||
request.node.user_properties.append(("vcr_verdict", verdict))
|
||||
|
||||
|
||||
def pytest_configure(config):
|
||||
global _controller_pluginmanager
|
||||
if os.environ.get("PYTEST_XDIST_WORKER"):
|
||||
return
|
||||
_controller_pluginmanager = config.pluginmanager
|
||||
|
||||
|
||||
def _resolve_terminal_reporter():
|
||||
global _controller_terminal_reporter
|
||||
if _controller_terminal_reporter is not None:
|
||||
return _controller_terminal_reporter
|
||||
if _controller_pluginmanager is None:
|
||||
return None
|
||||
_controller_terminal_reporter = _controller_pluginmanager.getplugin(
|
||||
"terminalreporter"
|
||||
)
|
||||
return _controller_terminal_reporter
|
||||
|
||||
|
||||
def pytest_runtest_logreport(report):
|
||||
if report.when != "teardown":
|
||||
return
|
||||
if os.environ.get("PYTEST_XDIST_WORKER"):
|
||||
return
|
||||
if not vcr_verbose_enabled():
|
||||
return
|
||||
reporter = _resolve_terminal_reporter()
|
||||
if reporter is None:
|
||||
return
|
||||
verdict = next(
|
||||
(v for k, v in (report.user_properties or []) if k == "vcr_verdict"),
|
||||
None,
|
||||
)
|
||||
if not verdict:
|
||||
return
|
||||
reporter.write_line(f"{verdict} :: {report.nodeid}")
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
@ -61,15 +214,18 @@ def setup_and_teardown():
|
||||
|
||||
|
||||
def pytest_collection_modifyitems(config, items):
|
||||
# Separate tests in 'test_amazing_proxy_custom_logger.py' and other tests
|
||||
if not _vcr_disabled():
|
||||
for item in items:
|
||||
if item.get_closest_marker("vcr") is not None:
|
||||
continue
|
||||
item.add_marker(pytest.mark.vcr)
|
||||
|
||||
custom_logger_tests = [
|
||||
item for item in items if "custom_logger" in item.parent.name
|
||||
]
|
||||
other_tests = [item for item in items if "custom_logger" not in item.parent.name]
|
||||
|
||||
# Sort tests based on their names
|
||||
custom_logger_tests.sort(key=lambda x: x.name)
|
||||
other_tests.sort(key=lambda x: x.name)
|
||||
|
||||
# Reorder the items list
|
||||
items[:] = custom_logger_tests + other_tests
|
||||
|
||||
@ -1,3 +1,41 @@
|
||||
Unit tests for individual LLM providers.
|
||||
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.
|
||||
Name of the test file is the name of the LLM provider - e.g. `test_openai.py` is for OpenAI.
|
||||
|
||||
## Redis-backed VCR 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:<test_id>`. 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.
|
||||
|
||||
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`.
|
||||
|
||||
### Required environment
|
||||
|
||||
`REDIS_HOST`, `REDIS_PORT`, `REDIS_PASSWORD` — same vars CircleCI uses for
|
||||
its other Redis-backed jobs. Provider credentials
|
||||
(`ANTHROPIC_API_KEY`, `OPENAI_API_KEY`, `AWS_*`, etc.) are needed only on
|
||||
cache-miss (the daily re-record), not on replay.
|
||||
|
||||
### Flushing the cache
|
||||
|
||||
When you want the next run to re-record immediately instead of waiting
|
||||
for the 24h TTL:
|
||||
|
||||
```bash
|
||||
make test-llm-translation-flush-vcr-cache
|
||||
```
|
||||
|
||||
### Disabling VCR
|
||||
|
||||
Skip the cache entirely (every call goes live, no recording):
|
||||
|
||||
```bash
|
||||
LITELLM_VCR_DISABLE=1 uv run pytest tests/llm_translation/test_<file>.py
|
||||
```
|
||||
|
||||
@ -5,6 +5,7 @@
|
||||
# - Function-scoped fixture resets litellm globals to true defaults
|
||||
# - Module-scoped reload only in single-process mode
|
||||
|
||||
import asyncio
|
||||
import importlib
|
||||
import os
|
||||
import sys
|
||||
@ -14,9 +15,195 @@ import pytest
|
||||
sys.path.insert(
|
||||
0, os.path.abspath("../..")
|
||||
) # Adds the parent directory to the system path
|
||||
import litellm
|
||||
|
||||
import asyncio
|
||||
import litellm # noqa: E402
|
||||
|
||||
from tests._vcr_redis_persister import ( # noqa: E402
|
||||
filter_non_2xx_response,
|
||||
format_vcr_verdict,
|
||||
make_redis_persister,
|
||||
mark_test_outcome_for_cassette,
|
||||
patch_vcrpy_aiohttp_record_path,
|
||||
vcr_verbose_enabled,
|
||||
)
|
||||
|
||||
|
||||
_controller_pluginmanager = None
|
||||
_controller_terminal_reporter = None
|
||||
|
||||
|
||||
# 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_azure_o_series.py",
|
||||
"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"}
|
||||
)
|
||||
|
||||
# Tests that observe live cross-call provider state (e.g. prompt-cache
|
||||
# warm-up between two consecutive calls); replay can't reproduce that state.
|
||||
_VCR_INCOMPATIBLE_NODEID_SUFFIXES = frozenset(
|
||||
{
|
||||
"::test_prompt_caching",
|
||||
"TestBedrockInvokeNovaJson::test_json_response_pydantic_obj",
|
||||
"::test_bedrock_converse__streaming_passthrough",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _is_vcr_incompatible(nodeid: str) -> bool:
|
||||
return any(nodeid.endswith(suffix) for suffix in _VCR_INCOMPATIBLE_NODEID_SUFFIXES)
|
||||
|
||||
|
||||
_FILTERED_REQUEST_HEADERS = (
|
||||
"authorization",
|
||||
"x-api-key",
|
||||
"anthropic-api-key",
|
||||
"anthropic-version",
|
||||
"openai-api-key",
|
||||
"azure-api-key",
|
||||
"api-key",
|
||||
"cookie",
|
||||
"x-amz-security-token",
|
||||
"x-amz-date",
|
||||
"x-amz-content-sha256",
|
||||
"amz-sdk-invocation-id",
|
||||
"amz-sdk-request",
|
||||
"x-goog-api-key",
|
||||
"x-goog-user-project",
|
||||
)
|
||||
|
||||
_FILTERED_RESPONSE_HEADERS = (
|
||||
"set-cookie",
|
||||
"x-request-id",
|
||||
"request-id",
|
||||
"cf-ray",
|
||||
"anthropic-organization-id",
|
||||
"openai-organization",
|
||||
"x-amzn-requestid",
|
||||
"x-amzn-trace-id",
|
||||
"date",
|
||||
)
|
||||
|
||||
|
||||
def _scrub_response(response):
|
||||
if not isinstance(response, dict):
|
||||
return response
|
||||
headers = response.get("headers") or {}
|
||||
if isinstance(headers, dict):
|
||||
for header in list(headers):
|
||||
if header.lower() in _FILTERED_RESPONSE_HEADERS:
|
||||
headers.pop(header, None)
|
||||
return response
|
||||
|
||||
|
||||
def _before_record_response(response):
|
||||
return filter_non_2xx_response(_scrub_response(response))
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def vcr_config():
|
||||
return {
|
||||
"filter_headers": list(_FILTERED_REQUEST_HEADERS),
|
||||
"decode_compressed_response": True,
|
||||
"record_mode": "new_episodes",
|
||||
"allow_playback_repeats": True,
|
||||
"match_on": (
|
||||
"method",
|
||||
"scheme",
|
||||
"host",
|
||||
"port",
|
||||
"path",
|
||||
"query",
|
||||
"body",
|
||||
),
|
||||
"before_record_response": _before_record_response,
|
||||
}
|
||||
|
||||
|
||||
def _vcr_disabled() -> bool:
|
||||
if os.environ.get("LITELLM_VCR_DISABLE") == "1":
|
||||
return True
|
||||
return not os.environ.get("CASSETTE_REDIS_URL")
|
||||
|
||||
|
||||
def pytest_recording_configure(config, vcr):
|
||||
if _vcr_disabled():
|
||||
return
|
||||
vcr.register_persister(make_redis_persister())
|
||||
patch_vcrpy_aiohttp_record_path()
|
||||
|
||||
|
||||
@pytest.hookimpl(hookwrapper=True)
|
||||
def pytest_runtest_makereport(item, call):
|
||||
outcome = yield
|
||||
rep = outcome.get_result()
|
||||
setattr(item, f"rep_{rep.when}", rep)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _vcr_outcome_gate(request, vcr):
|
||||
yield
|
||||
cassette = vcr
|
||||
rep_call = getattr(request.node, "rep_call", 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:
|
||||
mark_test_outcome_for_cassette(cassette_path, test_passed)
|
||||
|
||||
if not vcr_verbose_enabled():
|
||||
return
|
||||
verdict = format_vcr_verdict(cassette)
|
||||
request.node.user_properties.append(("vcr_verdict", verdict))
|
||||
|
||||
|
||||
def pytest_configure(config):
|
||||
global _controller_pluginmanager
|
||||
if os.environ.get("PYTEST_XDIST_WORKER"):
|
||||
return
|
||||
_controller_pluginmanager = config.pluginmanager
|
||||
|
||||
|
||||
def _resolve_terminal_reporter():
|
||||
global _controller_terminal_reporter
|
||||
if _controller_terminal_reporter is not None:
|
||||
return _controller_terminal_reporter
|
||||
if _controller_pluginmanager is None:
|
||||
return None
|
||||
_controller_terminal_reporter = _controller_pluginmanager.getplugin(
|
||||
"terminalreporter"
|
||||
)
|
||||
return _controller_terminal_reporter
|
||||
|
||||
|
||||
def pytest_runtest_logreport(report):
|
||||
if report.when != "teardown":
|
||||
return
|
||||
if os.environ.get("PYTEST_XDIST_WORKER"):
|
||||
return
|
||||
if not vcr_verbose_enabled():
|
||||
return
|
||||
reporter = _resolve_terminal_reporter()
|
||||
if reporter is None:
|
||||
return
|
||||
verdict = next(
|
||||
(v for k, v in (report.user_properties or []) if k == "vcr_verdict"),
|
||||
None,
|
||||
)
|
||||
if not verdict:
|
||||
return
|
||||
reporter.write_line(f"{verdict} :: {report.nodeid}")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Capture TRUE defaults at conftest import time (before test modules pollute).
|
||||
@ -48,7 +235,6 @@ def event_loop():
|
||||
|
||||
@pytest.fixture(scope="function", autouse=True)
|
||||
def setup_and_teardown(event_loop): # Add event_loop as a dependency
|
||||
curr_dir = os.getcwd()
|
||||
sys.path.insert(0, os.path.abspath("../.."))
|
||||
|
||||
import litellm
|
||||
@ -97,15 +283,23 @@ def setup_and_teardown(event_loop): # Add event_loop as a dependency
|
||||
|
||||
|
||||
def pytest_collection_modifyitems(config, items):
|
||||
# Separate tests in 'test_amazing_proxy_custom_logger.py' and other tests
|
||||
if not _vcr_disabled():
|
||||
for item in items:
|
||||
filename = os.path.basename(str(item.fspath))
|
||||
if filename in _VCR_AUTO_MARKER_SKIP_FILES:
|
||||
continue
|
||||
if _is_vcr_incompatible(item.nodeid):
|
||||
continue
|
||||
if item.get_closest_marker("vcr") is not None:
|
||||
continue
|
||||
item.add_marker(pytest.mark.vcr)
|
||||
|
||||
custom_logger_tests = [
|
||||
item for item in items if "custom_logger" in item.parent.name
|
||||
]
|
||||
other_tests = [item for item in items if "custom_logger" not in item.parent.name]
|
||||
|
||||
# Sort tests based on their names
|
||||
custom_logger_tests.sort(key=lambda x: x.name)
|
||||
other_tests.sort(key=lambda x: x.name)
|
||||
|
||||
# Reorder the items list
|
||||
items[:] = custom_logger_tests + other_tests
|
||||
|
||||
@ -1885,3 +1885,42 @@ def test_metadata_filter_applies_to_azure_anthropic():
|
||||
headers={},
|
||||
)
|
||||
assert data.get("metadata") == {"user_id": "u2"}
|
||||
|
||||
|
||||
def test_anthropic_basic_completion_replay():
|
||||
response = litellm.completion(
|
||||
model="anthropic/claude-sonnet-4-5-20250929",
|
||||
messages=[{"role": "user", "content": "Hello!"}],
|
||||
)
|
||||
|
||||
assert response is not None
|
||||
content = response.choices[0].message.content
|
||||
assert isinstance(content, str) and content.strip(), content
|
||||
assert response.usage.prompt_tokens > 0
|
||||
assert response.usage.completion_tokens > 0
|
||||
assert response.choices[0].finish_reason in {"stop", "length"}
|
||||
|
||||
|
||||
def test_anthropic_streaming_completion_replay():
|
||||
stream = litellm.completion(
|
||||
model="anthropic/claude-sonnet-4-5-20250929",
|
||||
messages=[{"role": "user", "content": "Hello!"}],
|
||||
stream=True,
|
||||
)
|
||||
|
||||
collected_text = ""
|
||||
finish_reason = None
|
||||
chunk_count = 0
|
||||
for chunk in stream:
|
||||
chunk_count += 1
|
||||
if not chunk.choices:
|
||||
continue
|
||||
delta = chunk.choices[0].delta
|
||||
if delta and delta.content:
|
||||
collected_text += delta.content
|
||||
if chunk.choices[0].finish_reason:
|
||||
finish_reason = chunk.choices[0].finish_reason
|
||||
|
||||
assert chunk_count > 1, "expected multiple SSE chunks from streaming response"
|
||||
assert collected_text.strip(), collected_text
|
||||
assert finish_reason in {"stop", "length"}
|
||||
|
||||
243
tests/llm_translation/test_vcr_redis_persister.py
Normal file
243
tests/llm_translation/test_vcr_redis_persister.py
Normal file
@ -0,0 +1,243 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
import fakeredis
|
||||
import pytest
|
||||
from redis.exceptions import ConnectionError as RedisConnectionError
|
||||
from vcr.persisters.filesystem import CassetteNotFoundError
|
||||
from vcr.request import Request
|
||||
from vcr.serializers import yamlserializer
|
||||
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
|
||||
|
||||
from tests._vcr_redis_persister import ( # noqa: E402
|
||||
CASSETTE_TTL_SECONDS,
|
||||
MAX_EPISODES_PER_CASSETTE,
|
||||
filter_non_2xx_response,
|
||||
make_redis_persister,
|
||||
mark_test_outcome_for_cassette,
|
||||
redis_key_for,
|
||||
)
|
||||
|
||||
|
||||
def _sample_cassette_dict():
|
||||
request = Request(
|
||||
method="POST",
|
||||
uri="https://api.anthropic.com/v1/messages",
|
||||
body=b'{"model":"claude","messages":[{"role":"user","content":"hi"}]}',
|
||||
headers={"content-type": "application/json"},
|
||||
)
|
||||
response = {
|
||||
"status": {"code": 200, "message": "OK"},
|
||||
"headers": {"content-type": ["application/json"]},
|
||||
"body": {"string": b'{"id":"msg_1","type":"message"}'},
|
||||
}
|
||||
return {"requests": [request], "responses": [response]}
|
||||
|
||||
|
||||
def _persister_with_fake_redis():
|
||||
fake = fakeredis.FakeStrictRedis()
|
||||
return fake, make_redis_persister(client=fake)
|
||||
|
||||
|
||||
def test_save_then_load_roundtrips_cassette_content():
|
||||
_, persister = _persister_with_fake_redis()
|
||||
cassette_id = "tests/llm_translation/test_x/test_y"
|
||||
|
||||
persister.save_cassette(cassette_id, _sample_cassette_dict(), yamlserializer)
|
||||
requests, responses = persister.load_cassette(cassette_id, yamlserializer)
|
||||
|
||||
assert len(requests) == 1
|
||||
assert len(responses) == 1
|
||||
assert requests[0].method == "POST"
|
||||
assert requests[0].uri == "https://api.anthropic.com/v1/messages"
|
||||
assert responses[0]["status"]["code"] == 200
|
||||
assert responses[0]["body"]["string"] == b'{"id":"msg_1","type":"message"}'
|
||||
|
||||
|
||||
def test_saved_key_has_24h_ttl():
|
||||
fake, persister = _persister_with_fake_redis()
|
||||
cassette_id = "tests/llm_translation/test_x/test_ttl"
|
||||
|
||||
persister.save_cassette(cassette_id, _sample_cassette_dict(), yamlserializer)
|
||||
|
||||
ttl = fake.ttl(redis_key_for(cassette_id))
|
||||
assert CASSETTE_TTL_SECONDS - 5 <= ttl <= CASSETTE_TTL_SECONDS
|
||||
|
||||
|
||||
def test_load_missing_key_raises_cassette_not_found():
|
||||
_, persister = _persister_with_fake_redis()
|
||||
with pytest.raises(CassetteNotFoundError):
|
||||
persister.load_cassette("never/recorded", yamlserializer)
|
||||
|
||||
|
||||
def test_redis_key_normalizes_path_passed_by_pytest_recording():
|
||||
raw = "tests/llm_translation/cassettes/test_anthropic/test_streaming.yaml"
|
||||
assert (
|
||||
redis_key_for(raw)
|
||||
== "litellm:vcr:cassette:tests/llm_translation/test_anthropic/test_streaming"
|
||||
)
|
||||
|
||||
|
||||
class _FlakyRedis:
|
||||
def __init__(self, inner, fail_on: str):
|
||||
self._inner = inner
|
||||
self._fail_on = fail_on
|
||||
|
||||
def get(self, *args, **kwargs):
|
||||
if self._fail_on == "get":
|
||||
raise RedisConnectionError("simulated outage")
|
||||
return self._inner.get(*args, **kwargs)
|
||||
|
||||
def set(self, *args, **kwargs):
|
||||
if self._fail_on == "set":
|
||||
raise RedisConnectionError("simulated outage")
|
||||
return self._inner.set(*args, **kwargs)
|
||||
|
||||
|
||||
def test_save_swallows_connection_errors_so_teardown_does_not_fail():
|
||||
flaky = _FlakyRedis(fakeredis.FakeStrictRedis(), fail_on="set")
|
||||
persister = make_redis_persister(client=flaky)
|
||||
|
||||
persister.save_cassette(
|
||||
"tests/llm_translation/test_x/test_save_outage",
|
||||
_sample_cassette_dict(),
|
||||
yamlserializer,
|
||||
)
|
||||
|
||||
|
||||
def test_save_skipped_when_test_marked_failed_and_prior_cassette_preserved():
|
||||
fake, persister = _persister_with_fake_redis()
|
||||
cassette_id = "tests/llm_translation/test_x/test_flaky"
|
||||
key = redis_key_for(cassette_id)
|
||||
|
||||
good = _sample_cassette_dict()
|
||||
persister.save_cassette(cassette_id, good, yamlserializer)
|
||||
good_payload = fake.get(key)
|
||||
assert good_payload is not None
|
||||
|
||||
mark_test_outcome_for_cassette(cassette_id, passed=False)
|
||||
bad_response = {
|
||||
"status": {"code": 200, "message": "OK"},
|
||||
"headers": {},
|
||||
"body": {"string": b'{"id":"BAD","type":"message"}'},
|
||||
}
|
||||
bad = {"requests": good["requests"], "responses": [bad_response]}
|
||||
persister.save_cassette(cassette_id, bad, yamlserializer)
|
||||
|
||||
assert fake.get(key) == good_payload
|
||||
|
||||
|
||||
def test_save_proceeds_when_test_marked_passed():
|
||||
fake, persister = _persister_with_fake_redis()
|
||||
cassette_id = "tests/llm_translation/test_x/test_passed"
|
||||
key = redis_key_for(cassette_id)
|
||||
|
||||
mark_test_outcome_for_cassette(cassette_id, passed=True)
|
||||
persister.save_cassette(cassette_id, _sample_cassette_dict(), yamlserializer)
|
||||
|
||||
assert fake.get(key) is not None
|
||||
|
||||
|
||||
def test_save_refused_when_cassette_exceeds_max_episodes():
|
||||
fake, persister = _persister_with_fake_redis()
|
||||
cassette_id = "tests/llm_translation/test_x/test_runaway"
|
||||
key = redis_key_for(cassette_id)
|
||||
|
||||
persister.save_cassette(cassette_id, _sample_cassette_dict(), yamlserializer)
|
||||
seed_payload = fake.get(key)
|
||||
|
||||
request = Request(
|
||||
method="POST",
|
||||
uri="https://api.anthropic.com/v1/messages",
|
||||
body=b"x",
|
||||
headers={"content-type": "application/json"},
|
||||
)
|
||||
response = {
|
||||
"status": {"code": 200, "message": "OK"},
|
||||
"headers": {},
|
||||
"body": {"string": b"{}"},
|
||||
}
|
||||
bloated = {
|
||||
"requests": [request] * (MAX_EPISODES_PER_CASSETTE + 1),
|
||||
"responses": [response] * (MAX_EPISODES_PER_CASSETTE + 1),
|
||||
}
|
||||
persister.save_cassette(cassette_id, bloated, yamlserializer)
|
||||
|
||||
assert fake.get(key) == seed_payload
|
||||
|
||||
|
||||
def test_save_proceeds_at_max_episodes_threshold():
|
||||
fake, persister = _persister_with_fake_redis()
|
||||
cassette_id = "tests/llm_translation/test_x/test_at_threshold"
|
||||
key = redis_key_for(cassette_id)
|
||||
|
||||
request = Request(
|
||||
method="POST",
|
||||
uri="https://api.anthropic.com/v1/messages",
|
||||
body=b"x",
|
||||
headers={"content-type": "application/json"},
|
||||
)
|
||||
response = {
|
||||
"status": {"code": 200, "message": "OK"},
|
||||
"headers": {},
|
||||
"body": {"string": b"{}"},
|
||||
}
|
||||
at_threshold = {
|
||||
"requests": [request] * MAX_EPISODES_PER_CASSETTE,
|
||||
"responses": [response] * MAX_EPISODES_PER_CASSETTE,
|
||||
}
|
||||
persister.save_cassette(cassette_id, at_threshold, yamlserializer)
|
||||
|
||||
assert fake.get(key) is not None
|
||||
|
||||
|
||||
def test_save_proceeds_when_outcome_unknown():
|
||||
fake, persister = _persister_with_fake_redis()
|
||||
cassette_id = "tests/llm_translation/test_x/test_no_marker"
|
||||
key = redis_key_for(cassette_id)
|
||||
|
||||
persister.save_cassette(cassette_id, _sample_cassette_dict(), yamlserializer)
|
||||
|
||||
assert fake.get(key) is not None
|
||||
|
||||
|
||||
def test_load_treats_connection_errors_as_cassette_miss():
|
||||
flaky = _FlakyRedis(fakeredis.FakeStrictRedis(), fail_on="get")
|
||||
persister = make_redis_persister(client=flaky)
|
||||
|
||||
with pytest.raises(CassetteNotFoundError):
|
||||
persister.load_cassette(
|
||||
"tests/llm_translation/test_x/test_load_outage", yamlserializer
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("status_code", "expect_dropped"),
|
||||
[
|
||||
(200, False),
|
||||
(201, False),
|
||||
(204, False),
|
||||
(299, False),
|
||||
(300, True),
|
||||
(400, True),
|
||||
(401, True),
|
||||
(404, True),
|
||||
(429, True),
|
||||
(500, True),
|
||||
(502, True),
|
||||
(503, True),
|
||||
],
|
||||
)
|
||||
def test_only_2xx_responses_are_cached(status_code, expect_dropped):
|
||||
response = {
|
||||
"status": {"code": status_code, "message": "X"},
|
||||
"headers": {},
|
||||
"body": {"string": ""},
|
||||
}
|
||||
result = filter_non_2xx_response(response)
|
||||
assert (result is None) == expect_dropped
|
||||
if not expect_dropped:
|
||||
assert result is response
|
||||
32
uv.lock
generated
32
uv.lock
generated
@ -9,7 +9,7 @@ resolution-markers = [
|
||||
]
|
||||
|
||||
[options]
|
||||
exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for backwards compatibility when using relative exclude-newer values.
|
||||
exclude-newer = "2026-04-27T18:03:18.987479976Z"
|
||||
exclude-newer-span = "P3D"
|
||||
|
||||
[manifest]
|
||||
@ -3231,6 +3231,7 @@ dev = [
|
||||
{ name = "pytest-cov" },
|
||||
{ name = "pytest-mock" },
|
||||
{ name = "pytest-postgresql" },
|
||||
{ name = "pytest-recording" },
|
||||
{ name = "pytest-rerunfailures" },
|
||||
{ name = "pytest-timeout" },
|
||||
{ name = "pytest-xdist" },
|
||||
@ -3242,6 +3243,7 @@ dev = [
|
||||
{ name = "types-redis" },
|
||||
{ name = "types-requests" },
|
||||
{ name = "types-setuptools" },
|
||||
{ name = "vcrpy" },
|
||||
]
|
||||
healthcheck = [
|
||||
{ name = "httpx" },
|
||||
@ -3384,6 +3386,7 @@ dev = [
|
||||
{ name = "pytest-cov", specifier = "==5.0.0" },
|
||||
{ name = "pytest-mock", specifier = "==3.15.1" },
|
||||
{ name = "pytest-postgresql", specifier = "==7.0.2" },
|
||||
{ name = "pytest-recording", specifier = "==0.13.4" },
|
||||
{ name = "pytest-rerunfailures", specifier = "==15.1" },
|
||||
{ name = "pytest-timeout", specifier = "==2.4.0" },
|
||||
{ name = "pytest-xdist", specifier = "==3.8.0" },
|
||||
@ -3395,6 +3398,7 @@ dev = [
|
||||
{ name = "types-redis", specifier = "==4.6.0.20241004" },
|
||||
{ name = "types-requests", specifier = "==2.32.4.20260107" },
|
||||
{ name = "types-setuptools", specifier = "==75.8.0.20250225" },
|
||||
{ name = "vcrpy", specifier = "==8.1.1" },
|
||||
]
|
||||
healthcheck = [
|
||||
{ name = "httpx", specifier = "==0.28.1" },
|
||||
@ -5920,6 +5924,19 @@ wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/18/57/f2db5a80b10c3ac48ce41786cb9b14172f997509ee1b1055ab7db4238e5e/pytest_postgresql-7.0.2-py3-none-any.whl", hash = "sha256:0b0d31c51620a9c1d6be93286af354256bc58a47c379f56f4147b22da6e81fb5", size = 41447, upload-time = "2025-05-17T20:17:58.011Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pytest-recording"
|
||||
version = "0.13.4"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
dependencies = [
|
||||
{ name = "pytest" },
|
||||
{ name = "vcrpy" },
|
||||
]
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/32/9c/f4027c5f1693847b06d11caf4b4f6bb09f22c1581ada4663877ec166b8c6/pytest_recording-0.13.4.tar.gz", hash = "sha256:568d64b2a85992eec4ae0a419c855d5fd96782c5fb016784d86f18053792768c", size = 26576, upload-time = "2025-05-08T10:41:11.231Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/42/c2/ce34735972cc42d912173e79f200fe66530225190c06655c5632a9d88f1e/pytest_recording-0.13.4-py3-none-any.whl", hash = "sha256:ad49a434b51b1c4f78e85b1e6b74fdcc2a0a581ca16e52c798c6ace971f7f439", size = 13723, upload-time = "2025-05-08T10:41:09.684Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pytest-rerunfailures"
|
||||
version = "15.1"
|
||||
@ -7541,6 +7558,19 @@ wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/63/9a/0962b05b308494e3202d3f794a6e85abe471fe3cafdbcf95c2e8c713aabd/uvloop-0.21.0-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:a5c39f217ab3c663dc699c04cbd50c13813e31d917642d459fdcec07555cc553", size = 4660018, upload-time = "2024-10-14T23:38:10.888Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "vcrpy"
|
||||
version = "8.1.1"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
dependencies = [
|
||||
{ name = "pyyaml" },
|
||||
{ name = "wrapt" },
|
||||
]
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/b3/07/bcfd5ebd7cb308026ab78a353e091bd699593358be49197d39d004e5ad83/vcrpy-8.1.1.tar.gz", hash = "sha256:58e3053e33b423f3594031cb758c3f4d1df931307f1e67928e30cf352df7709f", size = 85770, upload-time = "2026-01-04T19:22:03.886Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/3a/d7/f79b05a5d728f8786876a7d75dfb0c5cae27e428081b2d60152fb52f155f/vcrpy-8.1.1-py3-none-any.whl", hash = "sha256:2d16f31ad56493efb6165182dd99767207031b0da3f68b18f975545ede8ac4b9", size = 42445, upload-time = "2026-01-04T19:22:02.532Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "waitress"
|
||||
version = "3.0.2"
|
||||
|
||||
Loading…
Reference in New Issue
Block a user