scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

tests: drop YAML cassettes, make Redis-backed VCR the default

Browse source 
Removes the YAML cassette feature entirely and replaces it with a
Redis-only flow. Every test in tests/llm_translation/ and
tests/llm_responses_api_testing/ is auto-marked @pytest.mark.vcr via
conftest.pytest_collection_modifyitems, so any provider call lands in
the Redis cache (litellm:vcr:cassette:<rel_path>, 24h TTL). First run
records, runs within the day replay, day rollover re-records and
surfaces upstream API drift within 24h.

VCR is on by default. Set LITELLM_VCR_DISABLE=1, or simply leave
REDIS_HOST unset, to opt out — both bypass the auto-marker entirely so
nothing about cassettes runs. record_mode is "once" so cache-miss
records and cache-hit replays.

The 8 existing respx-using files in tests/llm_translation are excluded
from the auto-marker (vcrpy and respx both patch the httpx transport;
applying both makes one silently win). The persister's own unit-test
file is also excluded so it doesn't recursively run inside a cassette.

The persister moved from tests/llm_translation/_vcr_redis_persister.py
to tests/_vcr_redis_persister.py so both conftests share it. The two
demo tests in test_anthropic_completion_vcr.py were ported into
test_anthropic_completion.py and the demo file was deleted.

Adds tests/_flush_vcr_cache.py + a Make target
(test-llm-translation-flush-vcr-cache) that scans
litellm:vcr:cassette:* and pipelines DELETEs, for the
"I want the next CI run to re-record now" workflow. Drops the now-dead
test-llm-translation-record target.

Provider keys are still required on cache-miss (which happens on first
run and once a day after that). Replay-mode runs need only Redis.
This commit is contained in:
mateo-berri 2026-04-30 21:40:58 +00:00
parent 33a051636d
commit c7d647b567
No known key found for this signature in database
13 changed files with 322 additions and 672 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
Makefile
View file

@ -186,16 +186,11 @@ test-llm-translation-single: install-test-deps
--junitxml=test-results/junit.xml \
-v --tb=short --maxfail=100 --timeout=300
# VCR cassette helpers --------------------------------------------------------
# Sweep-record every @pytest.mark.vcr test under tests/llm_translation in one
# shot. Provider credentials must be exported for the providers exercised
# (e.g. ANTHROPIC_API_KEY, OPENAI_API_KEY, AWS_*).
#
# Examples:
# ANTHROPIC_API_KEY=sk-ant-... make test-llm-translation-record
# ANTHROPIC_API_KEY=sk-ant-... make test-llm-translation-record \
# TARGET=test_anthropic_completion_vcr.py
TARGET ?= .
test-llm-translation-record: install-test-deps
$(UV_RUN) pytest tests/llm_translation/$(TARGET) \
-m vcr --record-mode=once -v --tb=short
# VCR cache helpers -----------------------------------------------------------
# Drop every Redis key under the ``litellm:vcr:cassette:*`` prefix. Use this
# when you want the next CI run (or local run) to re-record against live
# providers immediately instead of waiting for the 24h TTL to roll over.
# Reads REDIS_HOST / REDIS_PORT / REDIS_PASSWORD from the environment, the
# same vars CircleCI uses for its other Redis-backed jobs.
test-llm-translation-flush-vcr-cache:
$(UV_RUN) python tests/_flush_vcr_cache.py

53
tests/_flush_vcr_cache.py Normal file
View file

@ -0,0 +1,53 @@
"""Flush every VCR cassette stored in Redis.
Run via ``make test-llm-translation-flush-vcr-cache``. Use when you want the
next test run to re-record against live providers right now instead of
waiting for the 24h TTL to expire.
Reads ``REDIS_HOST``, ``REDIS_PORT``, ``REDIS_PASSWORD`` from the environment.
"""
from __future__ import annotations
import os
import sys
import redis
PREFIX = "litellm:vcr:cassette:"
SCAN_BATCH = 500
def _client() -> redis.Redis:
host = os.environ.get("REDIS_HOST")
if not host:
sys.exit("REDIS_HOST is not set; cannot flush VCR cache")
return redis.Redis(
host=host,
port=int(os.environ.get("REDIS_PORT", 6379)),
password=os.environ.get("REDIS_PASSWORD") or None,
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()

0
tests/llm_translation/_vcr_redis_persister.py → tests/_vcr_redis_persister.py
View file

109
tests/llm_responses_api_testing/conftest.py
View file

@ -1,5 +1,12 @@
# conftest.py
#
# Auto-applies ``@pytest.mark.vcr`` to every collected test (see
# ``pytest_collection_modifyitems``) so live provider calls land in the
# Redis-backed VCR cache. The persister, header scrubbing and 2xx-only
# filtering live in ``tests/_vcr_redis_persister.py``; the cache key and
# 24h TTL match the llm_translation conftest.
import asyncio
import importlib
import os
import sys
@ -9,9 +16,93 @@ 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,
make_redis_persister,
)
# Headers that must never be persisted to a cassette.
_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",
)
# Per-request response headers we strip so cassettes diff cleanly.
_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):
response = _scrub_response(response)
return filter_non_2xx_response(response)
@pytest.fixture(scope="module")
def vcr_config():
return {
"filter_headers": list(_FILTERED_REQUEST_HEADERS),
"decode_compressed_response": True,
"record_mode": "once",
"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("REDIS_HOST")
def pytest_recording_configure(config, vcr):
if _vcr_disabled():
return
vcr.register_persister(make_redis_persister())
@pytest.fixture(scope="session")
@ -61,15 +152,23 @@ def setup_and_teardown():
def pytest_collection_modifyitems(config, items):
# Separate tests in 'test_amazing_proxy_custom_logger.py' and other tests
# Auto-apply ``@pytest.mark.vcr`` so any provider call lands in the
# Redis cache. No respx files exist in this directory today; if any are
# added later, exclude them by filename here. Skip entirely when VCR
# is disabled (no REDIS_HOST or LITELLM_VCR_DISABLE=1).
if not _vcr_disabled():
for item in items:
if item.get_closest_marker("vcr") is not None:
continue
item.add_marker(pytest.mark.vcr)
# Preserve historical custom_logger ordering.
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

44
tests/llm_translation/Readme.md
View file

@ -2,28 +2,40 @@ 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.
## VCR-backed tests
## Redis-backed VCR cache
Tests decorated with `@pytest.mark.vcr` (typically in `*_vcr.py` files,
e.g. `test_anthropic_completion_vcr.py`) replay recorded HTTP traffic from
`cassettes/` via [`pytest-recording`](https://github.com/kiwicom/pytest-recording)
instead of calling the real provider. They run offline by default — no API
keys required, no per-PR cost.
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:<rel_path>`. 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.
To re-record every marked test in one sweep:
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
ANTHROPIC_API_KEY=sk-ant-... OPENAI_API_KEY=sk-... \
make test-llm-translation-record
make test-llm-translation-flush-vcr-cache
```
To scope to a single file:
### Disabling VCR
Skip the cache entirely (every call goes live, no recording):
```bash
ANTHROPIC_API_KEY=sk-ant-... \
make test-llm-translation-record TARGET=test_anthropic_completion_vcr.py
LITELLM_VCR_DISABLE=1 uv run pytest tests/llm_translation/test_<file>.py
```
See [`cassettes/README.md`](./cassettes/README.md) for the full workflow,
including how to add a new cassette-backed test and what to scrub from
recordings before committing.

115
tests/llm_translation/cassettes/README.md
View file

@ -1,115 +0,0 @@
# VCR cassettes for LLM translation tests
This directory holds [vcrpy](https://vcrpy.readthedocs.io/) cassettes used by
`tests/llm_translation/` to replay real provider HTTP traffic without hitting
the live API.
Why this exists is tracked in
[LIT-2683](https://linear.app/litellm-ai/issue/LIT-2683) and discussed in
`#sdlc` on Slack: e2e tests were repeatedly draining provider billing accounts
and producing flaky CI on outages. Recording the HTTP exchange once and
replaying it on subsequent runs gives us realistic provider responses
(streaming, headers, edge-case payloads) at zero per-PR cost.
## Layout
We use [`pytest-recording`](https://github.com/kiwicom/pytest-recording),
which auto-resolves the cassette path from the test location:
```
tests/llm_translation/
cassettes/
<test_module>/
<test_name>.yaml
test_<provider>_completion_vcr.py
conftest.py # provides the shared vcr_config fixture
```
For example, a test
`tests/llm_translation/test_anthropic_completion_vcr.py::test_basic` is backed by
`tests/llm_translation/cassettes/test_anthropic_completion_vcr/test_basic.yaml`.
## Adding a new cassette-backed test
1. Pick a small, deterministic call. Avoid prompts whose output depends on
wall-clock time, randomness, or live web data.
2. Write the test as you normally would and decorate it with
`@pytest.mark.vcr`. No imports beyond `pytest` are needed — the
`vcr_config` fixture in `conftest.py` is applied automatically.
3. Run the sweep recorder once with the credentials you need. Recording is
strictly opt-in via `--record-mode=once`; the default replay mode never
touches the network.
## Bulk re-record (the common path)
A single sweep replays every `@pytest.mark.vcr` test under
`tests/llm_translation`, hitting the live provider only for tests that don't
yet have a cassette:
```bash
ANTHROPIC_API_KEY=sk-ant-... \
OPENAI_API_KEY=sk-... \
AWS_ACCESS_KEY_ID=... AWS_SECRET_ACCESS_KEY=... \
make test-llm-translation-record
```
Or scope it to a single file:
```bash
ANTHROPIC_API_KEY=sk-ant-... \
make test-llm-translation-record TARGET=test_anthropic_completion_vcr.py
```
vcrpy's `once` record mode does **not** overwrite an existing cassette —
delete the file first if you're intentionally refreshing it:
```bash
rm tests/llm_translation/cassettes/test_anthropic_completion_vcr/test_basic.yaml
ANTHROPIC_API_KEY=sk-ant-... \
make test-llm-translation-record TARGET=test_anthropic_completion_vcr.py
```
To force a full refresh of every cassette in one shot:
```bash
rm -rf tests/llm_translation/cassettes/test_*
ANTHROPIC_API_KEY=... OPENAI_API_KEY=... AWS_* \
uv run pytest tests/llm_translation -m vcr --record-mode=all -v
```
## Refreshing the canned Anthropic fixtures (no API key)
The two Anthropic cassettes shipped with this directory are recorded against
an in-process mock so contributors can regenerate them without an
`ANTHROPIC_API_KEY`:
```bash
uv run python tests/llm_translation/cassettes/_record_anthropic_fixtures.py
```
For a full refresh against the real API, delete the cassettes first and use
the bulk-record sweep above.
## Cassette hygiene
After recording, **always inspect the YAML before committing**:
- The `vcr_config` fixture in `conftest.py` already filters the common
request headers (`Authorization`, `x-api-key`, `anthropic-api-key`, AWS
sigv4 headers, cookies, GCP keys, …) and per-request response headers
(`set-cookie`, `cf-ray`, request IDs, org IDs, dates).
- A request *body* might still contain a token if your test passed one
inline — scrub it manually.
- Quick sanity check: `grep -i 'sk-\|bearer\|api-key' cassettes/<dir>/*.yaml`
should be clean.
- Trim unhelpful response bodies if they're megabytes large but the
assertion only needs a few fields.
## Don't
- Don't commit cassettes with real API keys, OAuth tokens, or PII.
- Don't rely on cassettes for tests of *non-deterministic* behavior
(rate-limit retries, timeouts, model creativity). Mock those at the
LiteLLM layer instead.
- Don't manually edit cassette YAML beyond scrubbing — the format is
byte-sensitive (e.g. content-length headers must match the body).

268
tests/llm_translation/cassettes/_record_anthropic_fixtures.py
View file

@ -1,268 +0,0 @@
"""Helper script that records Anthropic-shaped cassettes against a local mock.
This is a *one-shot* utility, not a test. It exists so contributors can
regenerate the canned Anthropic cassettes shipped with this PR without
spending real provider credits and without needing an ``ANTHROPIC_API_KEY``.
Run it with::
uv run python tests/llm_translation/cassettes/_record_anthropic_fixtures.py
The script:
1. Spins up a tiny in-process HTTP server that returns canned Anthropic
``/v1/messages`` payloads (one non-streaming, one SSE streaming).
2. Records LiteLLM's real outbound HTTP through vcrpy.
3. Rewrites the cassette URL/Host so replay matches genuine
``https://api.anthropic.com/v1/messages`` traffic.
The cassettes are written to the per-test paths that ``pytest-recording``
expects (``cassettes/<test_module>/<test_name>.yaml``) so the existing tests
in ``test_anthropic_completion_vcr.py`` pick them up unchanged.
For a refresh against the *real* Anthropic API, use the
``--record-mode=once`` sweep described in
``tests/llm_translation/cassettes/README.md`` — that path needs a real
``ANTHROPIC_API_KEY``.
"""
from __future__ import annotations
import json
import os
import sys
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any, Iterable
import vcr # type: ignore[import-not-found]
import yaml # type: ignore[import-not-found]
REPO_ROOT = Path(__file__).resolve().parents[3]
sys.path.insert(0, str(REPO_ROOT))
import litellm # noqa: E402
CASSETTE_DIR = Path(__file__).parent / "test_anthropic_completion_vcr"
MOCK_HOST = "127.0.0.1"
NON_STREAM_PORT = 18765
STREAM_PORT = 18766
REAL_ANTHROPIC_HOST = "api.anthropic.com"
NON_STREAM_RESPONSE: dict[str, Any] = {
"id": "msg_01ABCDEFGHIJKLMNOPQRSTUV",
"type": "message",
"role": "assistant",
"model": "claude-sonnet-4-5-20250929",
"content": [{"type": "text", "text": "Hello! How can I help you today?"}],
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {
"input_tokens": 12,
"cache_creation_input_tokens": 0,
"cache_read_input_tokens": 0,
"output_tokens": 11,
},
}
STREAM_EVENTS: list[tuple[str, dict[str, Any]]] = [
(
"message_start",
{
"type": "message_start",
"message": {
"id": "msg_01STREAMABCDEFGH",
"type": "message",
"role": "assistant",
"model": "claude-sonnet-4-5-20250929",
"content": [],
"stop_reason": None,
"stop_sequence": None,
"usage": {"input_tokens": 14, "output_tokens": 1},
},
},
),
(
"content_block_start",
{
"type": "content_block_start",
"index": 0,
"content_block": {"type": "text", "text": ""},
},
),
(
"content_block_delta",
{
"type": "content_block_delta",
"index": 0,
"delta": {"type": "text_delta", "text": "Hello"},
},
),
(
"content_block_delta",
{
"type": "content_block_delta",
"index": 0,
"delta": {"type": "text_delta", "text": " from"},
},
),
(
"content_block_delta",
{
"type": "content_block_delta",
"index": 0,
"delta": {"type": "text_delta", "text": " LiteLLM!"},
},
),
("content_block_stop", {"type": "content_block_stop", "index": 0}),
(
"message_delta",
{
"type": "message_delta",
"delta": {"stop_reason": "end_turn", "stop_sequence": None},
"usage": {"output_tokens": 5},
},
),
("message_stop", {"type": "message_stop"}),
]
def _make_handler(mode: str) -> type[BaseHTTPRequestHandler]:
class Handler(BaseHTTPRequestHandler):
def log_message(self, *args: Any, **kwargs: Any) -> None: # silence
return
def do_POST(self) -> None: # noqa: N802
length = int(self.headers.get("Content-Length", "0"))
self.rfile.read(length)
if mode == "json":
body = json.dumps(NON_STREAM_RESPONSE).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.send_header("anthropic-ratelimit-requests-limit", "4000")
self.send_header("anthropic-ratelimit-requests-remaining", "3999")
self.end_headers()
self.wfile.write(body)
else:
self.send_response(200)
self.send_header("Content-Type", "text/event-stream")
self.send_header("Cache-Control", "no-cache")
self.end_headers()
for event_name, data in STREAM_EVENTS:
chunk = (
f"event: {event_name}\n" f"data: {json.dumps(data)}\n\n"
).encode("utf-8")
self.wfile.write(chunk)
self.wfile.flush()
return Handler
def _serve(port: int, mode: str) -> ThreadingHTTPServer:
srv = ThreadingHTTPServer((MOCK_HOST, port), _make_handler(mode))
threading.Thread(target=srv.serve_forever, daemon=True).start()
return srv
# Headers that vary every run (timestamps, server build) and must be stripped
# so the cassette is byte-stable across regenerations. Replay does not depend
# on them.
_NON_DETERMINISTIC_HEADERS = ("Date", "Server")
def _strip_nondeterministic_headers(path: Path) -> None:
"""Remove headers whose values change every run from the cassette.
Parses + rewrites the YAML rather than regex-substituting against a
fixed indentation, so this stays correct if vcrpy ever changes its
serialization style.
"""
cassette = yaml.safe_load(path.read_text())
for interaction in cassette.get("interactions") or []:
headers = (interaction.get("response") or {}).get("headers") or {}
for header in _NON_DETERMINISTIC_HEADERS:
headers.pop(header, None)
path.write_text(yaml.safe_dump(cassette, default_flow_style=False, sort_keys=False))
def _rewrite_cassette_to_real_host(path: Path, mock_host_port: str) -> None:
"""Replace mock host/port in the cassette with the real Anthropic host."""
text = path.read_text()
text = text.replace(f"http://{mock_host_port}", f"https://{REAL_ANTHROPIC_HOST}")
text = text.replace(mock_host_port, REAL_ANTHROPIC_HOST)
path.write_text(text)
_strip_nondeterministic_headers(path)
def _consume(iterable: Iterable[Any]) -> None:
for _ in iterable:
pass
def record_non_streaming() -> None:
cassette = CASSETTE_DIR / "test_anthropic_basic_completion_replay.yaml"
if cassette.exists():
cassette.unlink()
server = _serve(NON_STREAM_PORT, "json")
try:
my_vcr = vcr.VCR(
record_mode="all",
filter_headers=[
"authorization",
"x-api-key",
"anthropic-version",
],
)
with my_vcr.use_cassette(str(cassette)):
response = litellm.completion(
model="anthropic/claude-sonnet-4-5-20250929",
messages=[{"role": "user", "content": "Hello!"}],
api_base=f"http://{MOCK_HOST}:{NON_STREAM_PORT}",
api_key="sk-ant-recording",
)
assert response.choices[0].message.content
finally:
server.shutdown()
_rewrite_cassette_to_real_host(cassette, f"{MOCK_HOST}:{NON_STREAM_PORT}")
def record_streaming() -> None:
cassette = CASSETTE_DIR / "test_anthropic_streaming_completion_replay.yaml"
if cassette.exists():
cassette.unlink()
server = _serve(STREAM_PORT, "stream")
try:
my_vcr = vcr.VCR(
record_mode="all",
filter_headers=[
"authorization",
"x-api-key",
"anthropic-version",
],
)
with my_vcr.use_cassette(str(cassette)):
stream = litellm.completion(
model="anthropic/claude-sonnet-4-5-20250929",
messages=[{"role": "user", "content": "Hello!"}],
api_base=f"http://{MOCK_HOST}:{STREAM_PORT}",
api_key="sk-ant-recording",
stream=True,
)
_consume(stream)
finally:
server.shutdown()
_rewrite_cassette_to_real_host(cassette, f"{MOCK_HOST}:{STREAM_PORT}")
def main() -> None:
os.environ.setdefault("LITELLM_LOG", "WARNING")
CASSETTE_DIR.mkdir(parents=True, exist_ok=True)
record_non_streaming()
record_streaming()
print(f"Wrote cassettes to {CASSETTE_DIR}")
if __name__ == "__main__":
main()

41
tests/llm_translation/cassettes/test_anthropic_completion_vcr/test_anthropic_basic_completion_replay.yaml
View file

@ -1,41 +0,0 @@
interactions:
- request:
body: '{"model": "claude-sonnet-4-5-20250929", "messages": [{"role": "user", "content":
[{"type": "text", "text": "Hello!"}]}], "max_tokens": 64000}'
headers:
Accept-Encoding:
- gzip, deflate
Connection:
- keep-alive
Content-Length:
- '141'
Host:
- api.anthropic.com
User-Agent:
- litellm/1.84.0
accept:
- application/json
content-type:
- application/json
method: POST
uri: https://api.anthropic.com/v1/messages
response:
body:
string: '{"id": "msg_01ABCDEFGHIJKLMNOPQRSTUV", "type": "message", "role": "assistant",
"model": "claude-sonnet-4-5-20250929", "content": [{"type": "text", "text":
"Hello! How can I help you today?"}], "stop_reason": "end_turn", "stop_sequence":
null, "usage": {"input_tokens": 12, "cache_creation_input_tokens": 0, "cache_read_input_tokens":
0, "output_tokens": 11}}'
headers:
Content-Length:
- '358'
Content-Type:
- application/json
anthropic-ratelimit-requests-limit:
- '4000'
anthropic-ratelimit-requests-remaining:
- '3999'
status:
code: 200
message: OK
version: 1

81
tests/llm_translation/cassettes/test_anthropic_completion_vcr/test_anthropic_streaming_completion_replay.yaml
View file

@ -1,81 +0,0 @@
interactions:
- request:
body: '{"model": "claude-sonnet-4-5-20250929", "messages": [{"role": "user", "content":
[{"type": "text", "text": "Hello!"}]}], "max_tokens": 64000, "stream": true}'
headers:
Accept-Encoding:
- gzip, deflate
Connection:
- keep-alive
Content-Length:
- '157'
Host:
- api.anthropic.com
User-Agent:
- litellm/1.84.0
accept:
- application/json
content-type:
- application/json
method: POST
uri: https://api.anthropic.com/v1/messages
response:
body:
string: 'event: message_start
data: {"type": "message_start", "message": {"id": "msg_01STREAMABCDEFGH",
"type": "message", "role": "assistant", "model": "claude-sonnet-4-5-20250929",
"content": [], "stop_reason": null, "stop_sequence": null, "usage": {"input_tokens":
14, "output_tokens": 1}}}
event: content_block_start
data: {"type": "content_block_start", "index": 0, "content_block": {"type":
"text", "text": ""}}
event: content_block_delta
data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta",
"text": "Hello"}}
event: content_block_delta
data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta",
"text": " from"}}
event: content_block_delta
data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta",
"text": " LiteLLM!"}}
event: content_block_stop
data: {"type": "content_block_stop", "index": 0}
event: message_delta
data: {"type": "message_delta", "delta": {"stop_reason": "end_turn", "stop_sequence":
null}, "usage": {"output_tokens": 5}}
event: message_stop
data: {"type": "message_stop"}
'
headers:
Cache-Control:
- no-cache
Content-Type:
- text/event-stream
status:
code: 200
message: OK
version: 1

111
tests/llm_translation/conftest.py
View file

@ -4,7 +4,14 @@
# Mirrors the pattern in tests/local_testing/conftest.py:
# - Function-scoped fixture resets litellm globals to true defaults
# - Module-scoped reload only in single-process mode
#
# Also wires up the Redis-backed VCR cache. Every test in this directory is
# auto-marked with ``@pytest.mark.vcr`` (see ``pytest_collection_modifyitems``)
# unless its file appears in ``_RESPX_CONFLICTING_FILES`` — those use respx,
# which patches the same httpx transport vcrpy does. Cache key naming, TTL,
# and 2xx-only filtering live in ``tests/_vcr_redis_persister.py``.
import asyncio
import importlib
import os
import sys
@ -14,32 +21,48 @@ import pytest
sys.path.insert(
0, os.path.abspath("../..")
) # Adds the parent directory to the system path
# Make sibling helper modules under tests/llm_translation/ importable regardless
# of the directory pytest is invoked from.
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import litellm
import litellm # noqa: E402
import asyncio
from _vcr_redis_persister import ( # noqa: E402 (sibling module under conftest dir)
from tests._vcr_redis_persister import ( # noqa: E402
filter_non_2xx_response,
make_redis_persister,
)
# ---------------------------------------------------------------------------
# VCR cassette infrastructure (pytest-recording)
# VCR cassette infrastructure (pytest-recording + Redis)
# ---------------------------------------------------------------------------
# Tests marked with ``@pytest.mark.vcr`` replay HTTP traffic from a cassette
# under ``cassettes/<test_module>/<test_name>.yaml`` instead of hitting the
# live provider. Default record mode is ``none`` (replay only) so CI never
# accidentally calls a real LLM. To re-record every marked test in one sweep::
#
# ANTHROPIC_API_KEY=sk-ant-... \
# uv run pytest tests/llm_translation -m vcr --record-mode=once
#
# See ``tests/llm_translation/cassettes/README.md`` for the full workflow.
# All tests in tests/llm_translation/ are auto-marked with ``@pytest.mark.vcr``
# (excluding the respx-using files listed below). On cache miss vcrpy records
# the live response into Redis under ``litellm:vcr:cassette:<rel_path>`` with
# a 24h TTL; subsequent runs within that window replay without touching the
# network. Set ``LITELLM_VCR_DISABLE=1`` to skip VCR entirely (e.g. when
# debugging an upstream API change locally).
# Test files that use ``respx`` to patch httpx. vcrpy patches the same
# transport, so applying both to the same test will make one of them silently
# win and the other look like a no-op. Skip auto-marking these.
_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",
}
)
# The persister's own unit tests must not run inside a VCR cassette context —
# they call ``save_cassette`` / ``load_cassette`` directly against fakeredis
# and don't make HTTP calls, but auto-marking them would still wrap each
# test in a Redis lookup we don't want.
_VCR_AUTO_MARKER_SKIP_FILES = _RESPX_CONFLICTING_FILES | frozenset(
{"test_vcr_redis_persister.py"}
)
# Headers that must never be persisted to a cassette. Matched
# case-insensitively by vcrpy.
@ -47,8 +70,8 @@ _FILTERED_REQUEST_HEADERS = (
"authorization",
"x-api-key",
"anthropic-api-key",
# Strip ``anthropic-version`` so live-recorded and mock-recorded cassettes
# have the same shape (the local mock helper drops it too).
# Strip ``anthropic-version`` so cassettes have a stable shape across
# SDK versions that bump the header.
"anthropic-version",
"openai-api-key",
"azure-api-key",
@ -104,11 +127,17 @@ def _before_record_response(response):
def vcr_config():
"""Shared VCR config consumed by ``pytest-recording``.
Applied to every ``@pytest.mark.vcr`` test in this directory.
``record_mode="once"`` is what makes this a useful daily cache:
- cassette absent (cache miss) → record the live call into Redis,
- cassette present (cache hit) → replay only.
24h TTL on the Redis key means each new day's first run records against
live providers, surfacing API drift within a day instead of silently
serving stale responses forever.
"""
return {
"filter_headers": list(_FILTERED_REQUEST_HEADERS),
"decode_compressed_response": True,
"record_mode": "once",
# Match on full request shape so streaming vs non-streaming and
# different prompts produce distinct cassettes.
"match_on": (
@ -124,22 +153,24 @@ def vcr_config():
}
def pytest_recording_configure(config, vcr):
"""Swap vcrpy's default filesystem persister for a Redis-backed one.
def _vcr_disabled() -> bool:
"""VCR is disabled when explicitly opted out or when Redis isn't wired.
Opt-in via ``LITELLM_VCR_REDIS=1`` so local dev keeps the YAML-on-disk
behaviour and CI (which sets the flag) gets a 24h-TTL cache that
auto-refreshes against live providers without manual ``make`` runs.
No Redis means no cache to read from or write to — fall back to live
calls instead of silently writing YAML to disk (which we don't ship).
"""
if os.environ.get("LITELLM_VCR_REDIS") != "1":
if os.environ.get("LITELLM_VCR_DISABLE") == "1":
return True
return not os.environ.get("REDIS_HOST")
def pytest_recording_configure(config, vcr):
"""Register the Redis-backed cassette persister."""
if _vcr_disabled():
return
vcr.register_persister(make_redis_persister())
# pytest-recording's default cassette dir is
# ``<test_dir>/cassettes/<test_module>``. Keep that — it gives every test its
# own file and avoids name collisions across modules.
# ---------------------------------------------------------------------------
# Capture TRUE defaults at conftest import time (before test modules pollute).
# ---------------------------------------------------------------------------
@ -170,7 +201,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
@ -239,15 +269,28 @@ def _vcr_record_retries(setup_and_teardown, request):
def pytest_collection_modifyitems(config, items):
# Separate tests in 'test_amazing_proxy_custom_logger.py' and other tests
# 1. Auto-apply ``@pytest.mark.vcr`` to every collected test in this
# directory so any provider call lands in the Redis cache. Skip files
# that use respx (it patches the same transport vcrpy does) and the
# persister's own unit tests. Skip entirely if VCR is disabled (no
# REDIS_HOST or LITELLM_VCR_DISABLE=1) so dev runs without Redis
# don't go through cassette logic at all.
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 item.get_closest_marker("vcr") is not None:
continue
item.add_marker(pytest.mark.vcr)
# 2. Preserve the historical ordering of custom_logger tests vs the rest.
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

48
tests/llm_translation/test_anthropic_completion.py
View file

@ -1885,3 +1885,51 @@ def test_metadata_filter_applies_to_azure_anthropic():
headers={},
)
assert data.get("metadata") == {"user_id": "u2"}
def test_anthropic_basic_completion_replay():
"""Smoke-test that a vanilla Anthropic completion replays from a cassette.
Exercises the full LiteLLM transformation pipeline (request shaping +
response parsing) against a real-shape Anthropic payload. The cassette
is loaded from the Redis-backed VCR cache configured in conftest.py.
"""
response = litellm.completion(
model="anthropic/claude-sonnet-4-5-20250929",
messages=[{"role": "user", "content": "Hello!"}],
)
assert response is not None
assert response.choices[0].message.content == ("Hello! How can I help you today?")
assert response.usage.prompt_tokens == 12
assert response.usage.completion_tokens == 11
# Anthropic sets stop_reason="end_turn" → litellm normalises to "stop"
assert response.choices[0].finish_reason == "stop"
def test_anthropic_streaming_completion_replay():
"""Replay a streaming Anthropic completion from the VCR cache.
Exercises the SSE chunk parser and the public streaming surface — any
regression in the streaming transformation surfaces here because the
cassette captures every ``content_block_delta`` event Anthropic emits.
"""
stream = litellm.completion(
model="anthropic/claude-sonnet-4-5-20250929",
messages=[{"role": "user", "content": "Hello!"}],
stream=True,
)
collected_text = ""
finish_reason = None
for chunk in stream:
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 collected_text == "Hello from LiteLLM!"
assert finish_reason == "stop"

97
tests/llm_translation/test_anthropic_completion_vcr.py
View file

@ -1,97 +0,0 @@
"""
Cassette-replayed Anthropic completion tests.
These tests exercise the same end-to-end ``litellm.completion`` code paths as
``test_anthropic_completion.py`` but replay HTTP traffic from cassettes under
``cassettes/test_anthropic_completion_vcr/`` instead of calling
``api.anthropic.com``. CI runs them with no API key and zero cost.
Add a new test by writing it normally and decorating with ``@pytest.mark.vcr``.
The cassette path is resolved automatically from the test module + test name
by ``pytest-recording`` (see ``conftest.py``).
To re-record every marked test in one sweep::
ANTHROPIC_API_KEY=sk-ant-... \\
uv run pytest tests/llm_translation -m vcr --record-mode=once
See ``tests/llm_translation/cassettes/README.md`` for the full workflow.
"""
import os
import sys
import pytest
sys.path.insert(0, os.path.abspath("../.."))
import litellm # noqa: E402
# A non-secret placeholder API key. The vcr_config fixture in conftest.py
# filters Authorization / x-api-key headers from cassettes, so this value
# never lands on disk; it only stops the SDK from raising when
# ``ANTHROPIC_API_KEY`` is unset (the common CI case).
PLACEHOLDER_ANTHROPIC_API_KEY = "sk-ant-vcr-placeholder"
@pytest.fixture(autouse=True)
def _placeholder_anthropic_key(monkeypatch):
"""Ensure an API key is set so replay works offline.
If a real key is present (e.g. when re-recording with
``--record-mode=once``), we leave it untouched.
"""
if not os.environ.get("ANTHROPIC_API_KEY"):
monkeypatch.setenv("ANTHROPIC_API_KEY", PLACEHOLDER_ANTHROPIC_API_KEY)
@pytest.mark.vcr
def test_anthropic_basic_completion_replay():
"""Smoke-test that a vanilla Anthropic completion replays from a cassette.
This is the canonical example for the cassette-based testing pattern: no
API key required at runtime, deterministic output, and the full LiteLLM
transformation pipeline (request shaping + response parsing) runs against
a real-shape Anthropic payload.
"""
response = litellm.completion(
model="anthropic/claude-sonnet-4-5-20250929",
messages=[{"role": "user", "content": "Hello!"}],
)
assert response is not None
assert response.choices[0].message.content == ("Hello! How can I help you today?")
assert response.usage.prompt_tokens == 12
assert response.usage.completion_tokens == 11
# Anthropic sets stop_reason="end_turn" → litellm normalises to "stop"
assert response.choices[0].finish_reason == "stop"
@pytest.mark.vcr
def test_anthropic_streaming_completion_replay():
"""Replay a streaming Anthropic completion from a cassette.
Exercises the SSE chunk parser and the public streaming surface. The
underlying cassette captures every ``content_block_delta`` event Anthropic
emits, so any regression in the streaming transformation will surface
here.
"""
stream = litellm.completion(
model="anthropic/claude-sonnet-4-5-20250929",
messages=[{"role": "user", "content": "Hello!"}],
stream=True,
)
collected_text = ""
finish_reason = None
for chunk in stream:
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 collected_text == "Hello from LiteLLM!"
assert finish_reason == "stop"

6
tests/llm_translation/test_vcr_redis_persister.py
View file

@ -25,9 +25,11 @@ from vcr.persisters.filesystem import CassetteNotFoundError
from vcr.request import Request
from vcr.serializers import yamlserializer
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
# Make tests/ importable as a package so we can pull the shared persister
# without depending on pytest's CWD.
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
from _vcr_redis_persister import ( # noqa: E402
from tests._vcr_redis_persister import ( # noqa: E402
CASSETTE_TTL_SECONDS,
filter_non_2xx_response,
make_redis_persister,