ai-workspace-services/litellm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/litellm_internal_staging' into litellm_/peaceful-jang-c0e43b

Browse Source
This commit is contained in:
Yuneng Jiang 2026-05-14 14:12:08 -07:00
commit 5176e22737
No known key found for this signature in database
41 changed files with 2665 additions and 935 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

94
.github/actions/helm-oci-chart-releaser/action.yml vendored
View File

@ -1,94 +0,0 @@
name: Helm OCI Chart Releaser
description: Push Helm charts to OCI-based (Docker) registries
author: sergeyshaykhullin
branding:
color: yellow
icon: upload-cloud
inputs:
name:
required: true
description: Chart name
repository:
required: true
description: Chart repository name
tag:
required: true
description: Chart version
app_version:
required: true
description: App version
path:
required: false
description: Chart path (Default 'charts/{name}')
registry:
required: true
description: OCI registry
registry_username:
required: true
description: OCI registry username
registry_password:
required: true
description: OCI registry password
update_dependencies:
required: false
default: 'false'
description: Update chart dependencies before packaging (Default 'false')
outputs:
image:
value: ${{ steps.output.outputs.image }}
description: Chart image (Default '{registry}/{repository}/{image}:{tag}')
runs:
using: composite
steps:
- name: Helm | Setup
uses: azure/setup-helm@1a275c3b69536ee54be43f2070a358922e12c8d4 # v4.3.1
with:
version: v3.20.0
- name: Helm | Login
shell: bash
env:
REGISTRY_PASSWORD: ${{ inputs.registry_password }}
REGISTRY_USERNAME: ${{ inputs.registry_username }}
REGISTRY: ${{ inputs.registry }}
run: echo "$REGISTRY_PASSWORD" | helm registry login -u "$REGISTRY_USERNAME" --password-stdin "$REGISTRY"
- name: Helm | Dependency
if: inputs.update_dependencies == 'true'
shell: bash
env:
CHART_PATH: ${{ inputs.path == null && format('{0}/{1}', 'charts', inputs.name) || inputs.path }}
run: helm dependency update "$CHART_PATH"
- name: Helm | Package
shell: bash
env:
CHART_PATH: ${{ inputs.path == null && format('{0}/{1}', 'charts', inputs.name) || inputs.path }}
TAG: ${{ inputs.tag }}
APP_VERSION: ${{ inputs.app_version }}
run: helm package "$CHART_PATH" --version "$TAG" --app-version "$APP_VERSION"
- name: Helm | Push
shell: bash
env:
NAME: ${{ inputs.name }}
TAG: ${{ inputs.tag }}
REGISTRY: ${{ inputs.registry }}
REPOSITORY: ${{ inputs.repository }}
run: helm push "${NAME}-${TAG}.tgz" "oci://${REGISTRY}/${REPOSITORY}"
- name: Helm | Logout
shell: bash
env:
REGISTRY: ${{ inputs.registry }}
run: helm registry logout "$REGISTRY"
- name: Helm | Output
id: output
shell: bash
env:
REGISTRY: ${{ inputs.registry }}
REPOSITORY: ${{ inputs.repository }}
NAME: ${{ inputs.name }}
TAG: ${{ inputs.tag }}
run: echo "image=${REGISTRY}/${REPOSITORY}/${NAME}:${TAG}" >> $GITHUB_OUTPUT

35
.github/workflows/README.md vendored
View File

@ -1,35 +0,0 @@
# Simple PyPI Publishing
A GitHub workflow to manually publish LiteLLM packages to PyPI with a specified version.
## How to Use
1. Go to the **Actions** tab in the GitHub repository
2. Select **Simple PyPI Publish** from the workflow list
3. Click **Run workflow**
4. Enter the version to publish (e.g., `1.74.10`)
## What the Workflow Does
1. **Updates** the version in `pyproject.toml`
2. **Copies** the model prices backup file
3. **Builds** the Python package
4. **Publishes** to PyPI
## Prerequisites
Make sure the following secret is configured in the repository:
- `PYPI_PUBLISH_PASSWORD`: PyPI API token for authentication
## Example Usage
- Version: `1.74.11` → Publishes as v1.74.11
- Version: `1.74.10-hotfix1` → Publishes as v1.74.10-hotfix1
## Features
- ✅ Manual trigger with version input
- ✅ Automatic version updates in `pyproject.toml`
- ✅ Repository safety check (only runs on official repo)
- ✅ Clean package building and publishing
- ✅ Success confirmation with PyPI package link

92
.github/workflows/llm-translation-testing.yml vendored
View File

@ -1,92 +0,0 @@
name: LLM Translation Tests
on:
workflow_dispatch:
inputs:
release_candidate_tag:
description: "Release candidate tag/version"
required: true
type: string
push:
tags:
- "v*-rc*" # Triggers on release candidate tags like v1.0.0-rc1
permissions:
contents: read
jobs:
run-llm-translation-tests:
runs-on: ubuntu-latest
timeout-minutes: 90
steps:
- name: Checkout code
uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
with:
persist-credentials: false
ref: ${{ github.event.inputs.release_candidate_tag || github.ref }}
- name: Set up Python
uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
with:
python-version: "3.12"
- name: Set up uv
uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7
with:
version: "0.10.9"
enable-cache: false
- name: Restore uv dependencies cache
uses: actions/cache/restore@0057852bfaa89a56745cba8c7296529d2fc39830 # v4.3.0
with:
path: |
~/.cache/uv
.venv
key: ${{ runner.os }}-uv-${{ hashFiles('uv.lock') }}
restore-keys: |
${{ runner.os }}-uv-
- name: Install dependencies
run: |
uv sync --frozen
- name: Create test results directory
run: mkdir -p test-results
- name: Run LLM Translation Tests
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
COHERE_API_KEY: ${{ secrets.COHERE_API_KEY }}
GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}
AZURE_API_KEY: ${{ secrets.AZURE_API_KEY }}
AZURE_API_BASE: ${{ secrets.AZURE_API_BASE }}
AZURE_API_VERSION: ${{ secrets.AZURE_API_VERSION }}
RC_TAG: ${{ github.event.inputs.release_candidate_tag || github.ref_name }}
COMMIT_SHA: ${{ github.sha }}
run: |
python .github/workflows/run_llm_translation_tests.py \
--tag "$RC_TAG" \
--commit "$COMMIT_SHA" \
|| true # Continue even if tests fail
- name: Display test summary
if: always()
run: |
if [ -f "test-results/llm_translation_report.md" ]; then
echo "Test report generated successfully!"
echo "Artifact will contain:"
echo "- test-results/junit.xml (JUnit XML results)"
echo "- test-results/llm_translation_report.md (Beautiful markdown report)"
else
echo "Warning: Test report was not generated"
fi
- name: Upload test artifacts
uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
if: always()
with:
name: LLM-Translation-Artifact-${{ github.event.inputs.release_candidate_tag || github.ref_name }}
path: test-results/
retention-days: 30

153
.github/workflows/publish_to_pypi.yml vendored
View File

@ -1,153 +0,0 @@
name: Publish to PyPI
on:
workflow_dispatch:
jobs:
preflight-checks:
name: Preflight Checks
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
# No environment — read-only checks, no approval needed
outputs:
needs_publish: ${{ steps.check-litellm.outputs.needs_publish }}
version: ${{ steps.check-litellm.outputs.version }}
steps:
- name: Checkout repo
uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
- name: Set up Python
uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
with:
python-version: "3.12"
- name: Set up uv
uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7
with:
version: "0.10.9"
enable-cache: false
- name: Check litellm version on PyPI
id: check-litellm
run: |
VERSION=$(python - <<'PY'
import tomllib
with open("pyproject.toml", "rb") as f:
print(tomllib.load(f)["project"]["version"])
PY
)
echo "version=$VERSION" >> "$GITHUB_OUTPUT"
echo "Checking if litellm $VERSION exists on PyPI..."
HTTP_STATUS=$(curl -s -o /dev/null -w "%{http_code}" "https://pypi.org/pypi/litellm/$VERSION/json")
if [ "$HTTP_STATUS" = "200" ]; then
echo "litellm $VERSION already exists on PyPI. Skipping publish."
echo "needs_publish=false" >> "$GITHUB_OUTPUT"
else
echo "litellm $VERSION not found on PyPI. Publish needed."
echo "needs_publish=true" >> "$GITHUB_OUTPUT"
fi
- name: Sanity check proxy-extras version
run: |
# Read pinned version from project optional dependencies
PYPROJECT_VERSION=$(python3 - <<'PY'
import sys
import tomllib
with open("pyproject.toml", "rb") as f:
proxy_requirements = tomllib.load(f)["project"]["optional-dependencies"]["proxy"]
version = None
for requirement in proxy_requirements:
normalized = requirement.split(";", 1)[0].strip()
if not normalized.startswith("litellm-proxy-extras"):
continue
parts = normalized.split("==", 1)
if len(parts) == 2 and parts[0].strip() == "litellm-proxy-extras":
candidate = parts[1].strip()
if candidate:
version = candidate
break
if version is None:
print(
"::error::Could not find an exact litellm-proxy-extras pin in project.optional-dependencies.proxy",
file=sys.stderr,
)
sys.exit(1)
print(version)
PY
)
echo "pyproject.toml pins litellm-proxy-extras version: $PYPROJECT_VERSION"
# Check that the pinned version exists on PyPI
echo "Checking if litellm-proxy-extras $PYPROJECT_VERSION exists on PyPI..."
HTTP_STATUS=$(curl -s -o /dev/null -w "%{http_code}" "https://pypi.org/pypi/litellm-proxy-extras/$PYPROJECT_VERSION/json")
if [ "$HTTP_STATUS" != "200" ]; then
echo "::error::litellm-proxy-extras $PYPROJECT_VERSION is not published on PyPI yet. Publish it before releasing litellm."
exit 1
fi
echo "litellm-proxy-extras $PYPROJECT_VERSION exists on PyPI. Sanity check passed."
publish-litellm:
name: Publish litellm to PyPI
needs: preflight-checks
if: needs.preflight-checks.outputs.needs_publish == 'true'
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
id-token: write
contents: read
environment: pypi-publish
steps:
- name: Checkout repo
uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
- name: Set up Python
uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
with:
python-version: "3.12"
- name: Set up uv
uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7
with:
version: "0.10.9"
enable-cache: false
- name: Copy model prices backup
run: cp model_prices_and_context_window.json litellm/model_prices_and_context_window_backup.json
- name: Build package
run: |
rm -rf build dist
uv build
- name: Verify build artifacts
env:
EXPECTED_VERSION: ${{ needs.preflight-checks.outputs.version }}
run: |
echo "Contents of dist/:"
ls -la dist/
# Ensure we have both sdist and wheel
ls dist/*.tar.gz
ls dist/*.whl
# Verify built version matches expected
ls dist/ | grep -q "litellm-${EXPECTED_VERSION}" || {
echo "::error::Built artifacts do not match expected version $EXPECTED_VERSION"
ls dist/
exit 1
}
- name: Validate package metadata
run: |
uv tool run --from 'twine==6.2.0' twine check dist/*
- name: Publish to PyPI
uses: pypa/gh-action-pypi-publish@ed0c53931b1dc9bd32cbe73a98c7f6766f8a527e # v1.13.0

28
.github/workflows/read_pyproject_version.yml vendored
View File

@ -1,28 +0,0 @@
name: Read Version from pyproject.toml
on:
push:
branches:
- main # Change this to the default branch of your repository
permissions:
contents: read
jobs:
read-version:
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
with:
persist-credentials: false
- name: Read version from pyproject.toml
id: read-version
run: |
version=$(grep -m1 '^version' pyproject.toml | sed 's/version = "\(.*\)"/\1/')
printf "LITELLM_VERSION=%s" "$version" >> $GITHUB_ENV
- name: Display version
run: echo "Current version is $LITELLM_VERSION"

27
.github/workflows/results_stats.csv vendored
View File

@ -1,27 +0,0 @@
Date,"Ben
Ashley",Tom Brooks,Jimmy Cooney,"Sue
Daniels",Berlinda Fong,Terry Jones,Angelina Little,Linda Smith
10/1,FALSE,TRUE,TRUE,TRUE,TRUE,TRUE,FALSE,TRUE
10/2,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE
10/3,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE
10/4,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE
10/5,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE
10/6,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE
10/7,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE
10/8,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE
10/9,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE
10/10,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE
10/11,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE
10/12,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE
10/13,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE
10/14,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE
10/15,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE
10/16,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE
10/17,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE
10/18,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE
10/19,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE
10/20,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE
10/21,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE
10/22,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE
10/23,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE
Total,0,1,1,1,1,1,0,1
1 Date Ben Ashley Tom Brooks Jimmy Cooney Sue Daniels Berlinda Fong Terry Jones Angelina Little Linda Smith
2 10/1 FALSE TRUE TRUE TRUE TRUE TRUE FALSE TRUE
3 10/2 FALSE FALSE FALSE FALSE FALSE FALSE FALSE FALSE
4 10/3 FALSE FALSE FALSE FALSE FALSE FALSE FALSE FALSE
5 10/4 FALSE FALSE FALSE FALSE FALSE FALSE FALSE FALSE
6 10/5 FALSE FALSE FALSE FALSE FALSE FALSE FALSE FALSE
7 10/6 FALSE FALSE FALSE FALSE FALSE FALSE FALSE FALSE
8 10/7 FALSE FALSE FALSE FALSE FALSE FALSE FALSE FALSE
9 10/8 FALSE FALSE FALSE FALSE FALSE FALSE FALSE FALSE
10 10/9 FALSE FALSE FALSE FALSE FALSE FALSE FALSE FALSE
11 10/10 FALSE FALSE FALSE FALSE FALSE FALSE FALSE FALSE
12 10/11 FALSE FALSE FALSE FALSE FALSE FALSE FALSE FALSE
13 10/12 FALSE FALSE FALSE FALSE FALSE FALSE FALSE FALSE
14 10/13 FALSE FALSE FALSE FALSE FALSE FALSE FALSE FALSE
15 10/14 FALSE FALSE FALSE FALSE FALSE FALSE FALSE FALSE
16 10/15 FALSE FALSE FALSE FALSE FALSE FALSE FALSE FALSE
17 10/16 FALSE FALSE FALSE FALSE FALSE FALSE FALSE FALSE
18 10/17 FALSE FALSE FALSE FALSE FALSE FALSE FALSE FALSE
19 10/18 FALSE FALSE FALSE FALSE FALSE FALSE FALSE FALSE
20 10/19 FALSE FALSE FALSE FALSE FALSE FALSE FALSE FALSE
21 10/20 FALSE FALSE FALSE FALSE FALSE FALSE FALSE FALSE
22 10/21 FALSE FALSE FALSE FALSE FALSE FALSE FALSE FALSE
23 10/22 FALSE FALSE FALSE FALSE FALSE FALSE FALSE FALSE
24 10/23 FALSE FALSE FALSE FALSE FALSE FALSE FALSE FALSE
25 Total 0 1 1 1 1 1 0 1

229
.github/workflows/run_observatory_tests.yml vendored
View File

@ -1,229 +0,0 @@
name: Run Observatory Tests
on:
workflow_dispatch:
inputs:
tag:
description: "Docker image tag to test (e.g. v1.61.0.rc1)"
required: true
type: string
commit_hash:
description: "Commit hash (defaults to HEAD of current branch)"
required: false
type: string
workflow_call:
inputs:
tag:
description: "Docker image tag to test"
required: true
type: string
commit_hash:
description: "Commit hash of the release"
required: true
type: string
permissions:
contents: read
env:
LITELLM_MASTER_KEY: ${{ secrets.LITELLM_MASTER_KEY_STAGING }}
jobs:
observatory-tests:
runs-on: ubuntu-latest
timeout-minutes: 30
steps:
- name: Checkout repository
uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
with:
persist-credentials: false
- name: Validate tag input
env:
TAG: ${{ inputs.tag }}
run: |
if [[ ! "$TAG" =~ ^v[0-9]+\.[0-9]+\.[0-9]+ ]]; then
echo "Invalid tag format: $TAG (expected vX.Y.Z...)"
exit 1
fi
- name: Start LiteLLM container
env:
TAG: ${{ inputs.tag }}
AZURE_API_KEY: ${{ secrets.AZURE_API_KEY }}
AZURE_API_BASE: ${{ secrets.AZURE_API_BASE }}
WORKSPACE: ${{ github.workspace }}
run: |
docker run -d \
--name litellm-rc \
-p 4000:4000 \
-v "${WORKSPACE}/.github/observatory/litellm_config.yaml:/app/config.yaml" \
-e LITELLM_MASTER_KEY="${LITELLM_MASTER_KEY}" \
-e AZURE_API_KEY="${AZURE_API_KEY}" \
-e AZURE_API_BASE="${AZURE_API_BASE}" \
"litellm/litellm:${TAG}" \
--config /app/config.yaml --port 4000
- name: Wait for LiteLLM health check
run: |
echo "Waiting for LiteLLM to be ready..."
for i in $(seq 1 30); do
if curl -s -f http://localhost:4000/health/liveliness > /dev/null 2>&1; then
echo "LiteLLM is healthy"
exit 0
fi
echo "Attempt $i/30 - not ready yet, waiting 10s..."
sleep 10
done
echo "LiteLLM failed to start within 5 minutes"
docker logs litellm-rc
exit 1
- name: Start cloudflared tunnel
run: |
# Install cloudflared (pinned version + checksum)
curl -sL https://github.com/cloudflare/cloudflared/releases/download/2025.2.1/cloudflared-linux-amd64 -o /usr/local/bin/cloudflared
echo "afdfadd1ef552e66bffc35246fe30a9bd578356d2d386de95585ccfc432472b8 /usr/local/bin/cloudflared" | sha256sum -c -
chmod +x /usr/local/bin/cloudflared
# Start a quick tunnel (no account needed) and capture the URL
cloudflared tunnel --url http://localhost:4000 --no-autoupdate > /tmp/cloudflared.log 2>&1 &
CLOUDFLARED_PID=$!
echo "CLOUDFLARED_PID=$CLOUDFLARED_PID" >> $GITHUB_ENV
# Wait for tunnel URL to appear in logs
echo "Waiting for tunnel URL..."
for i in $(seq 1 30); do
TUNNEL_URL=$(grep -oP 'https://[a-z0-9-]+\.trycloudflare\.com' /tmp/cloudflared.log | head -1 || true)
if [ -n "$TUNNEL_URL" ]; then
echo "Tunnel URL: $TUNNEL_URL"
echo "TUNNEL_URL=$TUNNEL_URL" >> $GITHUB_ENV
exit 0
fi
sleep 2
done
echo "Failed to get tunnel URL"
cat /tmp/cloudflared.log
exit 1
- name: Verify tunnel connectivity
run: |
echo "Testing tunnel at ${TUNNEL_URL}..."
# Quick tunnels need time for DNS propagation; retry to avoid
# transient NXDOMAIN (curl exit code 6) on first attempt.
for i in $(seq 1 10); do
if curl -sf "${TUNNEL_URL}/health/liveliness" > /dev/null 2>&1; then
echo "Tunnel is working (attempt $i)"
exit 0
fi
echo "Attempt $i/10 - tunnel not routable yet, waiting 5s..."
sleep 5
done
echo "Tunnel failed to become reachable after 50s"
cat /tmp/cloudflared.log
exit 1
- name: Trigger observatory test run
id: trigger
env:
OBSERVATORY_URL: ${{ secrets.OBSERVATORY_URL }}
OBSERVATORY_API_KEY: ${{ secrets.OBSERVATORY_API_KEY }}
run: |
PAYLOAD=$(jq -n \
--arg url "${TUNNEL_URL}" \
--arg key "${LITELLM_MASTER_KEY}" \
'{
deployment_url: $url,
api_key: $key,
test_suite: "TestOAIAzureRelease",
models: ["gpt-4o-mini", "gpt-4o"]
}')
RESPONSE=$(curl -s -w "\n%{http_code}" -X POST "${OBSERVATORY_URL}/run-test" \
-H "Content-Type: application/json" \
-H "X-LiteLLM-Observatory-API-Key: ${OBSERVATORY_API_KEY}" \
-d "$PAYLOAD")
HTTP_CODE=$(echo "$RESPONSE" | tail -1)
BODY=$(echo "$RESPONSE" | head -n -1)
echo "Response ($HTTP_CODE): $BODY"
if [ "$HTTP_CODE" -ge 400 ]; then
echo "Failed to trigger test run"
exit 1
fi
# Extract request_id for polling this specific run
REQUEST_ID=$(echo "$BODY" | jq -r '.results.request_id')
if [ -z "$REQUEST_ID" ] || [ "$REQUEST_ID" = "null" ]; then
echo "Failed to extract request_id from response"
exit 1
fi
echo "Request ID: $REQUEST_ID"
echo "request_id=$REQUEST_ID" >> $GITHUB_OUTPUT
- name: Poll for test completion
id: poll
env:
OBSERVATORY_URL: ${{ secrets.OBSERVATORY_URL }}
OBSERVATORY_API_KEY: ${{ secrets.OBSERVATORY_API_KEY }}
REQUEST_ID: ${{ steps.trigger.outputs.request_id }}
run: |
TIMEOUT=900 # 15 minutes
INTERVAL=30
ELAPSED=0
while [ $ELAPSED -lt $TIMEOUT ]; do
STATUS=$(curl -s "${OBSERVATORY_URL}/run-status/${REQUEST_ID}" \
-H "X-LiteLLM-Observatory-API-Key: ${OBSERVATORY_API_KEY}")
RUN_STATUS=$(echo "$STATUS" | jq -r '.status')
echo "Run status (${ELAPSED}s elapsed): $RUN_STATUS"
if [ "$RUN_STATUS" = "completed" ] || [ "$RUN_STATUS" = "failed" ]; then
echo "Test finished with status: $RUN_STATUS"
echo "$STATUS" > /tmp/observatory_result.json
exit 0
fi
sleep $INTERVAL
ELAPSED=$((ELAPSED + INTERVAL))
done
echo "Timed out waiting for test to complete after ${TIMEOUT}s"
exit 1
- name: Verify test results
run: |
RESULT=$(cat /tmp/observatory_result.json)
echo "Full result: $RESULT"
STATUS=$(echo "$RESULT" | jq -r '.status')
TEST_PASSED=$(echo "$RESULT" | jq -r '.result.test_passed // false')
FAILURE_RATE=$(echo "$RESULT" | jq -r '.result.failure_rate // "N/A"')
ERROR=$(echo "$RESULT" | jq -r '.error // empty')
echo "Status: $STATUS"
echo "Test passed: $TEST_PASSED"
echo "Failure rate: $FAILURE_RATE"
if [ -n "$ERROR" ]; then
echo "Error: $ERROR"
fi
if [ "$STATUS" = "failed" ]; then
echo "Test run failed"
exit 1
fi
if [ "$TEST_PASSED" != "true" ]; then
echo "Tests did not pass (failure rate: $FAILURE_RATE)"
exit 1
fi
echo "All tests passed!"
- name: Print LiteLLM logs on failure
if: failure()
run: |
docker logs litellm-rc 2>/dev/null || true
cat /tmp/cloudflared.log 2>/dev/null || true
- name: Cleanup
if: always()
run: |
kill "$CLOUDFLARED_PID" 2>/dev/null || true
docker rm -f litellm-rc 2>/dev/null || true

48
.github/workflows/scan_duplicate_issues.yml vendored
View File

@ -1,48 +0,0 @@
name: Scan Duplicate Issues (One-Time)
on:
workflow_dispatch:
inputs:
threshold:
description: "Similarity threshold (0-1)"
required: false
default: "0.85"
close:
description: "Actually close duplicates (false = dry run)"
required: false
type: boolean
default: false
jobs:
scan:
runs-on: ubuntu-latest
permissions:
issues: write
contents: read
steps:
- name: Checkout scripts
uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
with:
sparse-checkout: .github/scripts
persist-credentials: false
- name: Set up Python
uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
with:
python-version: "3.12"
- name: Scan for duplicate issues
env:
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
INPUT_THRESHOLD: ${{ inputs.threshold }}
INPUT_CLOSE: ${{ inputs.close }}
run: |
CLOSE_FLAG=""
if [ "$INPUT_CLOSE" = "true" ]; then
CLOSE_FLAG="--close"
fi
python3 .github/scripts/close_duplicate_issues.py \
--scan \
--repo ${{ github.repository }} \
--threshold "$INPUT_THRESHOLD" \
$CLOSE_FLAG

45
.github/workflows/test-litellm.yml vendored
View File

@ -1,45 +0,0 @@
name: LiteLLM Mock Tests (folder - tests/test_litellm)
# DEPRECATED: This workflow is replaced by test-litellm-matrix.yml which runs
# the same tests in parallel across 10 jobs for faster CI times.
# Kept for manual debugging only.
on:
workflow_dispatch: # Manual trigger only
# pull_request:
# branches: [ main ]
permissions:
contents: read
jobs:
test:
runs-on: ubuntu-latest
timeout-minutes: 25
steps:
- uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
with:
persist-credentials: false
- name: Thank You Message
run: |
echo "### 🙏 Thank you for contributing to LiteLLM!" >> $GITHUB_STEP_SUMMARY
echo "Your PR is being tested now. We appreciate your help in making LiteLLM better!" >> $GITHUB_STEP_SUMMARY
- name: Set up Python
uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
with:
python-version: "3.12"
- name: Set up uv
uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7
with:
version: "0.10.9"
- name: Install dependencies
run: |
uv lock --check
uv sync --frozen --group ci --group proxy-dev --extra google --extra proxy --extra semantic-router
- name: Run tests
run: |
uv run --no-sync pytest tests/test_litellm --tb=short -vv --maxfail=10 -n 4 --durations=50

54
.github/workflows/update_release.py vendored
View File

@ -1,54 +0,0 @@
import os
import requests
from datetime import datetime
# GitHub API endpoints
GITHUB_API_URL = "https://api.github.com"
REPO_OWNER = "BerriAI"
REPO_NAME = "litellm"
# GitHub personal access token (required for uploading release assets)
GITHUB_ACCESS_TOKEN = os.environ.get("GITHUB_ACCESS_TOKEN")
# Headers for GitHub API requests
headers = {
"Accept": "application/vnd.github+json",
"Authorization": f"Bearer {GITHUB_ACCESS_TOKEN}",
"X-GitHub-Api-Version": "2022-11-28",
}
# Get the latest release
releases_url = f"{GITHUB_API_URL}/repos/{REPO_OWNER}/{REPO_NAME}/releases/latest"
response = requests.get(releases_url, headers=headers)
latest_release = response.json()
print("Latest release:", latest_release)
# Upload an asset to the latest release
upload_url = latest_release["upload_url"].split("{?")[0]
asset_name = "results_stats.csv"
asset_path = os.path.join(os.getcwd(), asset_name)
print("upload_url:", upload_url)
with open(asset_path, "rb") as asset_file:
asset_data = asset_file.read()
upload_payload = {
"name": asset_name,
"label": "Load test results",
"created_at": datetime.utcnow().isoformat() + "Z",
}
upload_headers = headers.copy()
upload_headers["Content-Type"] = "application/octet-stream"
upload_response = requests.post(
upload_url,
headers=upload_headers,
data=asset_data,
params=upload_payload,
)
if upload_response.status_code == 201:
print(f"Asset '{asset_name}' uploaded successfully to the latest release.")
else:
print(f"Failed to upload asset. Response: {upload_response.text}")

21
litellm/integrations/custom_logger.py
View File

@ -697,6 +697,27 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac
"""
return AgenticLoopPlan(run_agentic_loop=False)
async def async_post_agentic_loop_response_hook(
self,
response: Any,
plan: AgenticLoopPlan,
kwargs: Dict,
) -> Any:
"""
Post-process the response returned by the agentic-loop follow-up call.
Called after BaseLLMHTTPHandler executes ``AgenticLoopPlan.request_patch``
and receives the final response from the provider. Lets callbacks shape
what the client sees without bypassing the loop's safety / observability
machinery (depth tracking, fingerprinting, etc.).
Use ``plan.metadata`` to carry whatever the build step decided to expose
for post-processing (e.g. native tool_result blocks to inject).
Default returns ``response`` unchanged.
"""
return response
async def async_should_run_chat_completion_agentic_loop(
self,
response: Any,

247
litellm/integrations/websearch_interception/handler.py
View File

@ -19,12 +19,14 @@ from litellm.integrations.custom_logger import CustomLogger
from litellm.integrations.websearch_interception.tools import (
get_litellm_web_search_tool,
get_litellm_web_search_tool_openai,
is_anthropic_native_web_search_tool,
is_web_search_tool,
is_web_search_tool_chat_completion,
)
from litellm.integrations.websearch_interception.transformation import (
WebSearchTransformation,
)
from litellm.llms.base_llm.search.transformation import SearchResponse
from litellm.types.integrations.websearch_interception import (
WebSearchInterceptionConfig,
)
@ -36,6 +38,16 @@ from litellm.types.llms.openai import AllMessageValues
from litellm.types.utils import LlmProviders
from litellm.utils import ProviderConfigManager
# Key used to flag, on per-request kwargs, that the originating client sent
# an Anthropic-native ``web_search_*`` tool — meaning the final response
# should include ``web_search_tool_result`` content blocks so the client
# (e.g. Claude Desktop's citations panel) can render sources.
WEBSEARCH_EMIT_NATIVE_BLOCKS_KEY = "_websearch_interception_emit_native_blocks"
# Key on ``AgenticLoopPlan.metadata`` carrying the list of pre-built
# ``web_search_tool_result`` blocks to inject into the final response.
WEBSEARCH_NATIVE_BLOCKS_METADATA_KEY = "websearch_native_blocks"
class WebSearchInterceptionLogger(CustomLogger):
"""
@ -152,22 +164,55 @@ class WebSearchInterceptionLogger(CustomLogger):
f"(provider={provider_str}, query='{query}')"
)
# Execute search
# Native clients (Claude Desktop / Cowork / Anthropic SDK) make a
# standalone /v1/messages sub-request just for the search, and they
# expect the response in native shape with server_tool_use +
# web_search_tool_result content blocks so the citations panel can
# render. The agentic-loop post-hook never fires on this path because
# there is no model call — emit the native blocks here instead.
native_tool = next(
(t for t in tools if is_anthropic_native_web_search_tool(t)),
None,
)
# Execute search — keep the structured SearchResponse so the native
# block can carry per-result url/title/page_age.
try:
search_result_text = await self._execute_search(query)
search_result_text, structured = await self._execute_search(query)
except Exception as e:
verbose_logger.error(
f"WebSearchInterception: Short-circuit search failed: {e}"
)
search_result_text = f"Search failed: {e}"
search_result_text, structured = f"Search failed: {e}", None
content: List[Dict[str, Any]] = []
if native_tool is not None:
tool_use_id = f"srvtoolu_{uuid.uuid4().hex}"
tool_name = native_tool.get("name") or "web_search"
content.append(
{
"type": "server_tool_use",
"id": tool_use_id,
"name": tool_name,
"input": {"query": query},
}
)
content.append(
WebSearchTransformation.build_web_search_tool_result_block(
tool_use_id=tool_use_id,
search_response=structured,
)
)
# Keep the text block so non-native short-circuit callers (Claude Code,
# github_copilot, etc.) see the same payload they always have.
content.append({"type": "text", "text": search_result_text})
# Build synthetic Anthropic response
response: Dict[str, Any] = {
"id": f"msg_{str(uuid.uuid4())}",
"type": "message",
"role": "assistant",
"model": model,
"content": [{"type": "text", "text": search_result_text}],
"content": content,
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {"input_tokens": 0, "output_tokens": 0},
@ -175,7 +220,8 @@ class WebSearchInterceptionLogger(CustomLogger):
verbose_logger.debug(
"WebSearchInterception: Short-circuit search completed, "
f"returning synthetic response ({len(search_result_text)} chars)"
f"returning synthetic response ({len(search_result_text)} chars, "
f"native_blocks={native_tool is not None})"
)
return response
@ -219,6 +265,14 @@ class WebSearchInterceptionLogger(CustomLogger):
"WebSearchInterception: Converting native web_search tools to LiteLLM standard"
)
# If the client sent an Anthropic-native web_search_* tool, mark the
# request so the agentic loop emits native web_search_tool_result
# blocks in the final response (matches async_pre_request_hook). This
# deployment hook fires before async_pre_request_hook on some paths,
# so flagging here ensures the signal isn't lost regardless of order.
if any(is_anthropic_native_web_search_tool(t) for t in tools):
kwargs[WEBSEARCH_EMIT_NATIVE_BLOCKS_KEY] = True
# Convert native/custom web_search tools to LiteLLM standard
converted_tools = []
for tool in tools:
@ -342,6 +396,14 @@ class WebSearchInterceptionLogger(CustomLogger):
f"WebSearchInterception: Pre-request hook triggered for provider={custom_llm_provider}"
)
# If the client sent an Anthropic-native web_search_* tool, mark the
# request so the agentic loop emits native web_search_tool_result
# blocks in the final response (for citations panels, etc.). The flag
# is read by async_build_agentic_loop_plan; the leading underscore
# prefix ensures it is stripped before the follow-up call kwargs.
if any(is_anthropic_native_web_search_tool(t) for t in tools):
kwargs[WEBSEARCH_EMIT_NATIVE_BLOCKS_KEY] = True
# Convert native web search tools to LiteLLM standard
converted_tools = []
for tool in tools:
@ -591,7 +653,7 @@ class WebSearchInterceptionLogger(CustomLogger):
) -> AgenticLoopPlan:
tool_calls = tools["tool_calls"]
thinking_blocks = tools.get("thinking_blocks", [])
request_patch = await self._build_anthropic_request_patch(
request_patch, structured_results = await self._build_anthropic_request_patch(
model=model,
messages=messages,
tool_calls=tool_calls,
@ -600,12 +662,92 @@ class WebSearchInterceptionLogger(CustomLogger):
logging_obj=logging_obj,
kwargs=kwargs,
)
metadata: Dict[str, Any] = {
"tool_type": "websearch",
"response_format": "anthropic",
}
# If the client request originally carried a native web_search_* tool,
# pre-build the Anthropic-native ``web_search_tool_result`` blocks now
# (while we still have the structured SearchResponse list) and stash
# them on plan metadata for the post-hook to inject.
if kwargs.get(WEBSEARCH_EMIT_NATIVE_BLOCKS_KEY):
metadata[WEBSEARCH_NATIVE_BLOCKS_METADATA_KEY] = (
self._build_native_result_blocks(
tool_calls=tool_calls,
structured_results=structured_results,
)
)
return AgenticLoopPlan(
run_agentic_loop=True,
request_patch=request_patch,
metadata={"tool_type": "websearch", "response_format": "anthropic"},
metadata=metadata,
)
async def async_post_agentic_loop_response_hook(
self,
response: Any,
plan: AgenticLoopPlan,
kwargs: Dict,
) -> Any:
"""
Inject Anthropic-native ``web_search_tool_result`` blocks into the
final response when the originating client used a native
``web_search_*`` tool.
See ``WebSearchTransformation.build_web_search_tool_result_block`` for
the block shape. The blocks are prepended to ``response.content`` so
Anthropic-native clients (Claude Desktop, the Anthropic SDK) can
render citations / sources alongside the model's textual reply.
"""
native_blocks = plan.metadata.get(WEBSEARCH_NATIVE_BLOCKS_METADATA_KEY)
if not native_blocks:
return response
return self._inject_native_blocks(response, native_blocks)
@staticmethod
def _build_native_result_blocks(
tool_calls: List[Dict],
structured_results: List[Optional[SearchResponse]],
) -> List[Dict[str, Any]]:
"""Build one ``web_search_tool_result`` block per tool_call."""
blocks: List[Dict[str, Any]] = []
for i, tool_call in enumerate(tool_calls):
tool_use_id = tool_call.get("id") or ""
structured = structured_results[i] if i < len(structured_results) else None
blocks.append(
WebSearchTransformation.build_web_search_tool_result_block(
tool_use_id=tool_use_id,
search_response=structured,
)
)
return blocks
@staticmethod
def _inject_native_blocks(
response: Any, native_blocks: List[Dict[str, Any]]
) -> Any:
"""Prepend native blocks to response content, dict or object form."""
if not native_blocks:
return response
if isinstance(response, dict):
existing = response.get("content") or []
response["content"] = list(native_blocks) + list(existing)
return response
existing = getattr(response, "content", None) or []
try:
response.content = list(native_blocks) + list(existing)
except (AttributeError, TypeError):
# Object refused write — fall through and leave the response
# untouched rather than crash the request.
verbose_logger.debug(
"WebSearchInterception: could not inject native blocks into "
f"response of type {type(response).__name__}"
)
return response
async def async_run_chat_completion_agentic_loop(
self,
tools: Dict,
@ -733,7 +875,7 @@ class WebSearchInterceptionLogger(CustomLogger):
kwargs: Dict,
) -> Any:
"""Legacy path: execute search + build patch + run follow-up call."""
request_patch = await self._build_anthropic_request_patch(
request_patch, structured_results = await self._build_anthropic_request_patch(
model=model,
messages=messages,
tool_calls=tool_calls,
@ -755,7 +897,7 @@ class WebSearchInterceptionLogger(CustomLogger):
if max_tokens is None:
max_tokens = cast(int, kwargs.get("max_tokens", 1024))
return await anthropic_messages.acreate(
response = await anthropic_messages.acreate(
max_tokens=max_tokens,
messages=request_patch.messages,
model=request_patch.model or model,
@ -763,6 +905,18 @@ class WebSearchInterceptionLogger(CustomLogger):
**request_patch.kwargs,
)
# Legacy path: the new path goes through the typed plan + core
# dispatcher which runs the post-hook automatically. Mirror the
# native-block injection here so both paths behave identically.
if kwargs.get(WEBSEARCH_EMIT_NATIVE_BLOCKS_KEY):
native_blocks = self._build_native_result_blocks(
tool_calls=tool_calls,
structured_results=structured_results,
)
response = self._inject_native_blocks(response, native_blocks)
return response
async def _build_anthropic_request_patch(
self,
model: str,
@ -772,8 +926,16 @@ class WebSearchInterceptionLogger(CustomLogger):
anthropic_messages_optional_request_params: Dict,
logging_obj: Any,
kwargs: Dict,
) -> AgenticLoopRequestPatch:
"""Execute litellm.search() and build follow-up request patch."""
) -> Tuple[AgenticLoopRequestPatch, List[Optional[SearchResponse]]]:
"""
Execute litellm.search() and build follow-up request patch.
Returns the patch alongside the parallel list of structured
``SearchResponse`` objects (one per tool_call, ``None`` when the
search failed or the tool_call had no query). The caller uses these
to optionally build Anthropic-native ``web_search_tool_result``
content blocks for the final response.
"""
# Extract search queries from tool_use blocks
search_tasks = []
@ -797,23 +959,38 @@ class WebSearchInterceptionLogger(CustomLogger):
)
search_results = await asyncio.gather(*search_tasks, return_exceptions=True)
# Handle any exceptions in search results
# Split the gathered (text, structured) tuples into two parallel lists.
# The text list feeds the follow-up model call; the structured list
# is returned to the caller for native-block emission.
final_search_results: List[str] = []
structured_results: List[Optional[SearchResponse]] = []
for i, result in enumerate(search_results):
if isinstance(result, Exception):
verbose_logger.error(
f"WebSearchInterception: Search {i} failed with error: {str(result)}"
)
final_search_results.append(f"Search failed: {str(result)}")
elif isinstance(result, str):
# Explicitly cast to str for type checker
final_search_results.append(cast(str, result))
structured_results.append(None)
elif isinstance(result, tuple) and len(result) == 2:
text_value, structured_value = result
final_search_results.append(
cast(str, text_value)
if isinstance(text_value, str)
else str(text_value)
)
structured_results.append(
structured_value
if isinstance(structured_value, SearchResponse)
else None
)
else:
# Should never happen, but handle for type safety
# Defensive: legacy callers / unexpected shape — preserve text,
# drop structure.
verbose_logger.debug(
f"WebSearchInterception: Unexpected result type {type(result)} at index {i}"
)
final_search_results.append(str(result))
structured_results.append(None)
# Build assistant and user messages using transformation
assistant_message, user_message = WebSearchTransformation.transform_response(
@ -859,16 +1036,26 @@ class WebSearchInterceptionLogger(CustomLogger):
len(follow_up_messages),
len(final_search_results),
)
return AgenticLoopRequestPatch(
patch = AgenticLoopRequestPatch(
model=full_model_name,
messages=follow_up_messages,
max_tokens=max_tokens,
optional_params=optional_params_without_max_tokens,
kwargs=kwargs_for_followup,
)
return patch, structured_results
async def _execute_search(self, query: str) -> str:
"""Execute a single web search using router's search tools"""
async def _execute_search(self, query: str) -> Tuple[str, Optional[SearchResponse]]:
"""
Execute a single web search using router's search tools.
Returns both the formatted text (fed back to the model in the follow-up
call) and the structured ``SearchResponse`` (preserved so callers can
build Anthropic-native ``web_search_tool_result`` blocks for clients
that requested a native ``web_search_*`` tool). The structured value
is None on the failure path so callers can still emit an empty result
block rather than dropping the search entirely.
"""
try:
# Import router from proxy_server
try:
@ -934,7 +1121,7 @@ class WebSearchInterceptionLogger(CustomLogger):
verbose_logger.debug(
f"WebSearchInterception: Search completed for '{query}', got {len(search_result_text)} chars"
)
return search_result_text
return search_result_text, result
except Exception as e:
verbose_logger.error(
f"WebSearchInterception: Search failed for '{query}': {str(e)}"
@ -1015,7 +1202,8 @@ class WebSearchInterceptionLogger(CustomLogger):
)
search_results = await asyncio.gather(*search_tasks, return_exceptions=True)
# Handle any exceptions in search results
# Chat-completion path only needs text — OpenAI tool_result format
# has no equivalent of Anthropic's web_search_tool_result block.
final_search_results: List[str] = []
for i, result in enumerate(search_results):
if isinstance(result, Exception):
@ -1023,8 +1211,13 @@ class WebSearchInterceptionLogger(CustomLogger):
f"WebSearchInterception: Search {i} failed with error: {str(result)}"
)
final_search_results.append(f"Search failed: {str(result)}")
elif isinstance(result, str):
final_search_results.append(cast(str, result))
elif isinstance(result, tuple) and len(result) == 2:
text_value, _ = result
final_search_results.append(
cast(str, text_value)
if isinstance(text_value, str)
else str(text_value)
)
else:
verbose_logger.debug(
f"WebSearchInterception: Unexpected result type {type(result)} at index {i}"
@ -1112,9 +1305,11 @@ class WebSearchInterceptionLogger(CustomLogger):
kwargs=kwargs_for_followup,
)
async def _create_empty_search_result(self) -> str:
async def _create_empty_search_result(
self,
) -> Tuple[str, Optional[SearchResponse]]:
"""Create an empty search result for tool calls without queries"""
return "No search query provided"
return "No search query provided", None
@staticmethod
def initialize_from_proxy_config(

47
litellm/integrations/websearch_interception/tools.py
View File

@ -126,6 +126,27 @@ def is_web_search_tool_chat_completion(tool: Dict[str, Any]) -> bool:
return False
def is_anthropic_native_web_search_tool(tool: Dict[str, Any]) -> bool:
"""
Check if a tool is an Anthropic-native ``web_search_*`` tool.
Native clients (Anthropic SDK, Claude Desktop, Anthropic Console) send
tools like ``{"type": "web_search_20250305", "name": "web_search"}`` and
expect the response to contain ``web_search_tool_result`` content blocks
so that citations can be rendered. This helper identifies that contract
so the agentic loop can emit native-format blocks for those clients
without affecting clients that send the LiteLLM standard tool.
Returns False for the LiteLLM standard tool (``litellm_web_search``),
the OpenAI-shaped variant, the bare ``WebSearch`` legacy name, and the
bare ``web_search`` name (Claude Code style).
"""
tool_type = tool.get("type", "")
if not isinstance(tool_type, str):
return False
return tool_type.startswith("web_search_") and tool_type != "function"
def is_web_search_tool(tool: Dict[str, Any]) -> bool:
"""
Check if a tool is a web search tool (native or LiteLLM standard).
@ -135,7 +156,22 @@ def is_web_search_tool(tool: Dict[str, Any]) -> bool:
- OpenAI format: type == "function" with function.name == "litellm_web_search"
- Anthropic native: type starts with "web_search_" (e.g., "web_search_20250305")
- Claude Code: name == "web_search" with a type field
- Custom: name == "WebSearch" (legacy format)
- Custom: name == "WebSearch" (legacy interception marker — only matched
when input_schema is absent; see note below)
Note on the legacy ``WebSearch`` name:
Clients like Claude Desktop / Cowork ship a *client-side* tool called
``WebSearch`` (a fully-formed Anthropic client tool with its own
``input_schema``) that they handle themselves. Treating that as our
interception marker hijacks it server-side and the client's own tool
handler never fires — which means Cowork's separate native
``web_search_20250305`` sub-request (where citation data actually
flows) never gets made.
Real Anthropic client tools always carry an ``input_schema`` (the API
rejects them otherwise), so a bare ``{name: "WebSearch"}`` with no
schema is the only thing that could be a legacy interception marker.
Gate the match on schema absence to keep both groups working.
Args:
tool: Tool dictionary to check
@ -152,6 +188,10 @@ def is_web_search_tool(tool: Dict[str, Any]) -> bool:
True
>>> is_web_search_tool({"name": "calculator"})
False
>>> is_web_search_tool({"name": "WebSearch"}) # legacy interception marker
True
>>> is_web_search_tool({"name": "WebSearch", "input_schema": {"type": "object"}}) # Cowork client tool
False
"""
tool_name = tool.get("name", "")
tool_type = tool.get("type", "")
@ -175,8 +215,9 @@ def is_web_search_tool(tool: Dict[str, Any]) -> bool:
if tool_name == "web_search" and tool_type:
return True
# Check for legacy WebSearch format
if tool_name == "WebSearch":
# Legacy "WebSearch" interception marker — only when no schema is
# present, so real client-side WebSearch tools (Cowork) pass through.
if tool_name == "WebSearch" and "input_schema" not in tool:
return True
return False

66
litellm/integrations/websearch_interception/transformation.py
View File

@ -100,11 +100,14 @@ class WebSearchTransformation:
block_id = getattr(block, "id", None)
block_input = getattr(block, "input", {})
# Check for LiteLLM standard or legacy web search tools
# Handles: litellm_web_search, WebSearch, web_search
# Detect tool_use blocks that came from interception. After
# pre-request conversion the model always sees
# ``litellm_web_search``; the bare ``web_search`` entry handles
# callers that bypass our pre-request hooks (e.g. direct
# litellm.acompletion). "WebSearch" is intentionally omitted —
# see is_web_search_tool for the Cowork rationale.
if block_type == "tool_use" and block_name in (
LITELLM_WEB_SEARCH_TOOL_NAME,
"WebSearch",
"web_search",
):
# Convert to dict for easier handling
@ -190,10 +193,12 @@ class WebSearchTransformation:
getattr(function, "arguments", None) if function else None
)
# Check for LiteLLM standard or legacy web search tools
# Detect function-style web search tool_calls. ``WebSearch`` is
# intentionally omitted — see is_web_search_tool for the Cowork
# rationale (clients ship their own client-side ``WebSearch`` and
# we must not hijack it).
if tool_type == "function" and function_name in (
LITELLM_WEB_SEARCH_TOOL_NAME,
"WebSearch",
"web_search",
):
# Parse arguments (might be JSON string)
@ -350,6 +355,57 @@ class WebSearchTransformation:
return assistant_message, tool_messages
@staticmethod
def build_web_search_tool_result_block(
tool_use_id: str,
search_response: Optional[SearchResponse],
) -> Dict[str, Any]:
"""
Build an Anthropic-native ``web_search_tool_result`` content block.
Native Anthropic clients (Claude Desktop, the Anthropic SDK, the
Anthropic Console) expect search-tool results to be returned as
structured ``web_search_tool_result`` blocks so that citations and
source links can be rendered. The agentic loop currently feeds the
model a flat text blob in the follow-up call (which is correct — the
model needs readable evidence). This helper produces the *additional*
block that should accompany the model's text reply when the original
request used a native ``web_search_*`` tool.
Spec reference:
https://docs.anthropic.com/en/api/web-search-tool
Args:
tool_use_id: The ``tool_use_id`` the model emitted on the first
turn. Must match exactly so the client can pair the result
with its tool_use block.
search_response: Structured ``SearchResponse`` from
``litellm.asearch()``. If None or empty, the block is still
emitted with an empty result list (signals "search ran, no
results" rather than "search did not run").
"""
items: List[Dict[str, Any]] = []
if search_response is not None:
results = getattr(search_response, "results", None) or []
for r in results:
url = getattr(r, "url", "") or ""
title = getattr(r, "title", "") or ""
page_age = getattr(r, "date", None) or getattr(r, "last_updated", None)
items.append(
{
"type": "web_search_result",
"url": url,
"title": title,
"page_age": page_age,
"encrypted_content": "",
}
)
return {
"type": "web_search_tool_result",
"tool_use_id": tool_use_id,
"content": items,
}
@staticmethod
def format_search_response(result: SearchResponse) -> str:
"""

21
litellm/llms/custom_httpx/llm_http_handler.py
View File

@ -4634,6 +4634,7 @@ class BaseLLMHTTPHandler:
fingerprints: List[str],
fingerprint: str,
stream: bool = False,
callback: Optional[Any] = None,
) -> Any:
from litellm.anthropic_interface import messages as anthropic_messages
@ -4675,7 +4676,7 @@ class BaseLLMHTTPHandler:
kwargs_for_followup["max_agentic_loops"] = max_loops
kwargs_for_followup["_agentic_loop_fingerprints"] = fingerprints + [fingerprint]
return await anthropic_messages.acreate(
response = await anthropic_messages.acreate(
**{
"max_tokens": max_tokens,
"messages": patch.messages,
@ -4686,6 +4687,23 @@ class BaseLLMHTTPHandler:
}
)
if callback is not None:
try:
response = await callback.async_post_agentic_loop_response_hook(
response=response, plan=plan, kwargs=kwargs
)
except Exception as e:
_call_id = getattr(logging_obj, "litellm_call_id", "unknown")
verbose_logger.exception(
"LiteLLM.AgenticHookError: Exception in "
"async_post_agentic_loop_response_hook [call_id=%s model=%s]: %s",
_call_id,
model,
str(e),
)
return response
async def _execute_chat_completion_agentic_plan(
self,
plan: AgenticLoopPlan,
@ -4869,6 +4887,7 @@ class BaseLLMHTTPHandler:
fingerprints=fingerprints,
fingerprint=fingerprint,
stream=stream,
callback=callback,
)
except Exception as e:
_call_id = getattr(logging_obj, "litellm_call_id", "unknown")

779
tests/_vcr_conftest_common.py
View File

@ -5,25 +5,28 @@ See ``tests/llm_translation/Readme.md`` for the full design and
from __future__ import annotations
import ast
import atexit
import hashlib
import json
import os
import re
import socket
import sys
from collections import defaultdict
from typing import Iterable
import pytest
from tests._vcr_redis_persister import (
MAX_EPISODES_PER_CASSETTE,
VCR_VERBOSE_ENV,
cassette_cache_capacity_snapshot,
cassette_cache_health,
filter_non_2xx_response,
format_vcr_verdict,
make_redis_persister,
mark_test_outcome_for_cassette,
patch_vcrpy_aiohttp_record_path,
vcr_verbose_enabled,
)
CASSETTE_CACHE_HIGH_WATER_FRACTION = 0.85
@ -231,6 +234,29 @@ def _iter_header_values(headers, name: str):
yield value
_AWS_SIGV4_CREDENTIAL_RE = re.compile(
r"AWS4-HMAC-SHA256\s+Credential=([^/\s,]+)/", re.IGNORECASE
)
def _stable_key_value(header_name: str, raw: str) -> str:
"""Return a *stable* identifier for a credential header.
For Bearer / API-key headers the entire value is stable across calls,
so we hash it as-is. For AWS SigV4 ``Authorization`` headers, only
the access-key portion of ``Credential=AKIA.../<DATE>/...`` is stable
— date, region, signed headers, and signature all rotate per request,
so hashing the full value would push every Bedrock request into a new
cassette episode. Extract just the access-key id when present.
"""
if header_name.lower() != "authorization":
return raw
match = _AWS_SIGV4_CREDENTIAL_RE.search(raw)
if match:
return f"aws-sigv4:{match.group(1)}"
return raw
def _compute_key_fingerprint(request) -> str:
headers = getattr(request, "headers", None)
parts: list[str] = []
@ -242,7 +268,8 @@ def _compute_key_fingerprint(request) -> str:
text = text.strip()
if not text:
continue
parts.append(f"{header_name}={text}")
stable = _stable_key_value(header_name, text)
parts.append(f"{header_name}={stable}")
if not parts:
return "no-key"
digest = hashlib.sha256("\n".join(parts).encode("utf-8")).hexdigest()
@ -470,6 +497,235 @@ def register_persister_if_enabled(vcr) -> None:
_atexit_banner_registered = True
VCR_SKIP_REASON_USER_ATTR = "vcr_skip_reason"
# Marker reasons recorded per-item / per-test for the session summary.
SKIP_REASON_RESPX = "respx_conflict"
SKIP_REASON_RESPX_MODULE = "respx_conflict_module"
SKIP_REASON_INCOMPATIBLE = "incompatible"
SKIP_REASON_FILE_OPT_OUT = "file_opt_out"
SKIP_REASON_DISABLED = "disabled"
SKIP_REASON_PRE_MARKED = "already_marked"
# Hostnames we consider an "expensive live call" if a non-VCR-marked test
# happens to hit them. Localhost/redis/databases are explicitly excluded.
_LIVE_CALL_HOST_SUFFIXES = (
".openai.com",
".anthropic.com",
".vertexai.googleapis.com",
".aiplatform.googleapis.com",
".googleapis.com",
".x.ai",
".cohere.ai",
".cohere.com",
".voyageai.com",
".perplexity.ai",
".mistral.ai",
".groq.com",
".huggingface.co",
".azure.com",
".tavily.com",
".serper.dev",
".searchapi.io",
".firecrawl.dev",
".exa.ai",
)
_LIVE_CALL_LOCAL_PREFIXES = (
"127.",
"localhost",
"::1",
"0.0.0.0",
"10.",
"172.16.",
"172.17.",
"172.18.",
"172.19.",
"172.20.",
"172.21.",
"172.22.",
"172.23.",
"172.24.",
"172.25.",
"172.26.",
"172.27.",
"172.28.",
"172.29.",
"172.30.",
"172.31.",
"192.168.",
)
class _RespxUsageVisitor(ast.NodeVisitor):
"""AST visitor that flags real respx wiring in a test module.
Substring scans of the source text are unreliable: a comment like
``# Previously used respx.mock`` or a docstring referencing respx
would falsely flag the module. We only count:
* ``@pytest.mark.respx`` / ``@respx.mock`` decorators
* ``with respx.mock(): ...`` context managers
* ``respx.mock(...)`` / ``respx.mock`` attribute access
* function parameters / fixture arguments named ``respx_mock``
"""
def __init__(self) -> None:
self.uses_respx = False
def _decorator_is_respx(self, dec: ast.expr) -> bool:
# ``@respx.mock`` (Attribute) or ``@respx.mock(...)`` (Call wrapping Attribute)
if isinstance(dec, ast.Call):
dec = dec.func
if isinstance(dec, ast.Attribute):
return (
isinstance(dec.value, ast.Name)
and dec.value.id == "respx"
and dec.attr == "mock"
)
return False
def _is_pytest_mark_respx(self, dec: ast.expr) -> bool:
# ``@pytest.mark.respx`` or ``@pytest.mark.respx(...)``.
if isinstance(dec, ast.Call):
dec = dec.func
if (
isinstance(dec, ast.Attribute)
and dec.attr == "respx"
and isinstance(dec.value, ast.Attribute)
and dec.value.attr == "mark"
and isinstance(dec.value.value, ast.Name)
and dec.value.value.id == "pytest"
):
return True
return False
def _check_decorators(self, decs: list[ast.expr]) -> None:
for d in decs:
if self._decorator_is_respx(d) or self._is_pytest_mark_respx(d):
self.uses_respx = True
def _check_args(self, args: ast.arguments) -> None:
# ``def test_foo(respx_mock): ...`` — pytest supplies the fixture
# whenever the parameter name appears, regardless of marker.
all_args = (
list(args.args)
+ list(args.kwonlyargs)
+ (list(args.posonlyargs) if hasattr(args, "posonlyargs") else [])
)
for a in all_args:
if a.arg == "respx_mock":
self.uses_respx = True
return
def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
self._check_decorators(node.decorator_list)
self._check_args(node.args)
self.generic_visit(node)
def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
self._check_decorators(node.decorator_list)
self._check_args(node.args)
self.generic_visit(node)
def visit_ClassDef(self, node: ast.ClassDef) -> None:
self._check_decorators(node.decorator_list)
self.generic_visit(node)
def _is_respx_mock_attr(self, node: ast.expr) -> bool:
return (
isinstance(node, ast.Attribute)
and isinstance(node.value, ast.Name)
and node.value.id == "respx"
and node.attr == "mock"
)
def visit_With(self, node: ast.With) -> None:
for item in node.items:
ctx = item.context_expr
if isinstance(ctx, ast.Call):
ctx = ctx.func
if self._is_respx_mock_attr(ctx):
self.uses_respx = True
self.generic_visit(node)
def visit_AsyncWith(self, node: ast.AsyncWith) -> None:
for item in node.items:
ctx = item.context_expr
if isinstance(ctx, ast.Call):
ctx = ctx.func
if self._is_respx_mock_attr(ctx):
self.uses_respx = True
self.generic_visit(node)
def visit_Call(self, node: ast.Call) -> None:
# ``respx.mock(...)`` invocation outside a ``with``/decorator —
# e.g. ``mock = respx.mock()`` at module scope.
if self._is_respx_mock_attr(node.func):
self.uses_respx = True
self.generic_visit(node)
def _module_uses_respx(item) -> bool:
"""Return True if the test's *module* actually wires up respx.
Uses an ``ast`` walk (not substring matching) so comments and
docstrings that mention respx don't count as real usage. A bare
``from respx import MockRouter`` import with no other respx
references therefore won't flag the module — that's exactly the
dead-import case this PR is trying to surface.
"""
module = getattr(item, "module", None)
src_file = getattr(module, "__file__", None) or str(getattr(item, "path", "") or "")
if not src_file or not os.path.isfile(src_file):
return False
try:
with open(src_file, encoding="utf-8") as f:
src = f.read()
except OSError:
return False
try:
tree = ast.parse(src, filename=src_file)
except SyntaxError:
# If the test file itself is broken, fall back to "no respx" —
# the test will fail collection on its own and we don't want
# the auto-marker to mask that with a misleading skip reason.
return False
visitor = _RespxUsageVisitor()
visitor.visit(tree)
return visitor.uses_respx
def _item_uses_respx(item) -> bool:
"""Return True if *this specific item* will trigger respx.
Two signals: the ``respx`` pytest marker, and the ``respx_mock``
fixture appearing in the item's resolved fixture chain. Either alone
causes vcrpy + respx to fight over the httpx transport.
"""
if item.get_closest_marker("respx") is not None:
return True
fixturenames = getattr(item, "fixturenames", None) or ()
if "respx_mock" in fixturenames:
return True
return False
# Cache the source-scan result so we don't reread each module per item.
_RESPX_MODULE_CACHE: dict[str, bool] = {}
def _module_path_uses_respx(item) -> bool:
src_file = str(getattr(item, "path", "") or "")
if not src_file:
return False
cached = _RESPX_MODULE_CACHE.get(src_file)
if cached is not None:
return cached
result = _module_uses_respx(item)
_RESPX_MODULE_CACHE[src_file] = result
return result
def apply_vcr_auto_marker_to_items(
items,
*,
@ -478,26 +734,349 @@ def apply_vcr_auto_marker_to_items(
) -> None:
"""Auto-apply ``pytest.mark.vcr`` to collected items.
``skip_files`` are basenames to leave un-marked (e.g. respx-using
files, since respx and vcrpy both patch the httpx transport).
``skip_nodeid_suffixes`` are node-id suffixes for individual tests
that depend on live cross-call provider state.
Skip semantics (in priority order):
1. ``vcr_disabled()`` — global env-var off-switch (``LITELLM_VCR_DISABLE=1``
or no ``CASSETTE_REDIS_URL``).
2. Item already carries ``@pytest.mark.vcr`` — leave it alone.
3. Item triggers respx (per-item marker / fixture) — vcrpy and respx
both patch the httpx transport so applying both makes one silently
no-op. We tag the item ``vcr_skip_reason=respx_conflict``.
4. Module wires up respx anywhere — even tests in the file that don't
themselves use respx still inherit the patched transport when
respx fixtures activate at session level. Tagged
``respx_conflict_module``.
5. ``skip_files`` / ``skip_nodeid_suffixes`` opt-out lists from the
caller — used for tests that observe live cross-call provider state
(e.g. prompt-cache warmup) which deterministic replay can't model.
Tagged ``incompatible``.
Each skipped item gets a ``vcr_skip_reason`` attribute so the
session-end summary can show why it isn't cached.
"""
if vcr_disabled():
for item in items:
setattr(item, VCR_SKIP_REASON_USER_ATTR, SKIP_REASON_DISABLED)
return
skip_files = frozenset(skip_files)
skip_nodeid_suffixes = tuple(skip_nodeid_suffixes)
for item in items:
if item.get_closest_marker("vcr") is not None:
setattr(item, VCR_SKIP_REASON_USER_ATTR, SKIP_REASON_PRE_MARKED)
continue
if _item_uses_respx(item):
setattr(item, VCR_SKIP_REASON_USER_ATTR, SKIP_REASON_RESPX)
continue
filename = os.path.basename(str(item.path))
if filename in skip_files:
# Trust the caller's opt-out, but split by reason: if the
# module actually uses respx, label the conflict precisely so
# the summary surfaces dead respx imports vs. real conflicts.
if _module_path_uses_respx(item):
setattr(item, VCR_SKIP_REASON_USER_ATTR, SKIP_REASON_RESPX_MODULE)
else:
setattr(item, VCR_SKIP_REASON_USER_ATTR, SKIP_REASON_FILE_OPT_OUT)
continue
if any(item.nodeid.endswith(suffix) for suffix in skip_nodeid_suffixes):
continue
if item.get_closest_marker("vcr") is not None:
setattr(item, VCR_SKIP_REASON_USER_ATTR, SKIP_REASON_INCOMPATIBLE)
continue
item.add_marker(pytest.mark.vcr)
# ---------------------------------------------------------------------------
# Per-test stats accumulator + verdict classification.
#
# The session-end summary needs richer signal than the line-level verdict:
# - which tests overflowed ``MAX_EPISODES_PER_CASSETTE`` (cassette refused
# to save → live calls every CI run);
# - which tests fired live HTTP at a real LLM endpoint while VCR was not
# active for them (genuine wasted spend, not just "test mocked elsewhere");
# - skip-reason buckets so we can tell respx-conflict from
# incompatible-by-design from "module imports respx but never uses it".
# ---------------------------------------------------------------------------
# Verdict tags used in the per-test logline AND in the session summary
# breakdown.
VERDICT_HIT = "VCR HIT"
VERDICT_MISS_RECORDED = "VCR MISS:RECORDED"
VERDICT_MISS_OVERFLOW = "VCR MISS:OVERFLOW"
VERDICT_MISS_NOT_PERSISTED = "VCR MISS:NOT_PERSISTED"
VERDICT_PARTIAL = "VCR PARTIAL"
VERDICT_NOOP_NO_TRAFFIC = "VCR NOOP"
VERDICT_UNMARKED_LIVE_CALL = "VCR UNMARKED:LIVE_CALL"
VERDICT_UNMARKED_NO_TRAFFIC = "VCR UNMARKED:NO_TRAFFIC"
VERDICT_DISABLED = "VCR DISABLED"
# Per-session stats. Cleared by ``_reset_session_stats`` for unit tests.
_session_stats = {
"verdict_counts": defaultdict(int),
"overflow_tests": [], # list of nodeids
"unmarked_live_call_tests": [], # list of (nodeid, hosts)
"skip_reason_counts": defaultdict(int),
"skip_reason_examples": defaultdict(list),
}
def _reset_session_stats() -> None:
_session_stats["verdict_counts"].clear()
_session_stats["overflow_tests"].clear()
_session_stats["unmarked_live_call_tests"].clear()
_session_stats["skip_reason_counts"].clear()
_session_stats["skip_reason_examples"].clear()
# user_properties keys used to ship structured outcome data from xdist workers
# back to the controller. ``vcr_verdict`` is the human-readable line that
# ``VerboseReporterState.maybe_emit_verdict`` writes next to each test;
# ``vcr_outcome`` + ``vcr_recorded_by`` are the structured payload that
# ``aggregate_report_outcome`` folds into the controller's ``_session_stats``
# so the session-end summary actually has data in xdist mode.
_USER_PROP_VERDICT_LINE = "vcr_verdict"
_USER_PROP_OUTCOME = "vcr_outcome"
_USER_PROP_RECORDED_BY = "vcr_recorded_by"
def _emit_outcome_payload(
node,
verdict: str,
*,
skip_reason: str | None = None,
live_call_hosts: Iterable[str] | None = None,
) -> None:
"""Stash a structured VCR outcome on a pytest node so the xdist
controller can fold it into ``_session_stats``.
On a worker, ``record_vcr_outcome`` has already updated the worker-local
``_session_stats`` — but in xdist mode that state lives in the worker
process and never reaches the controller's ``pytest_terminal_summary``.
We use the report's ``user_properties`` channel (which xdist round-trips
back to the controller) to ship the outcome, and
``aggregate_report_outcome`` rebuilds the controller's stats from there.
The recorder tags ``vcr_recorded_by`` with ``PYTEST_XDIST_WORKER`` so
the controller can distinguish "recorded in this same main process —
already counted" from "recorded in a worker — needs aggregation here".
"""
node.user_properties.append(
(
_USER_PROP_OUTCOME,
{
"verdict": verdict,
"skip_reason": skip_reason,
"live_call_hosts": list(live_call_hosts) if live_call_hosts else [],
},
)
)
node.user_properties.append(
(_USER_PROP_RECORDED_BY, os.environ.get("PYTEST_XDIST_WORKER", ""))
)
def aggregate_report_outcome(report) -> None:
"""Fold a worker-produced VCR outcome into the controller's session stats.
No-op outside the xdist controller path:
* On a worker, ``_session_stats`` was already updated in-process by
``record_vcr_outcome`` — and the worker doesn't render the summary
anyway, so there's nothing for us to aggregate.
* In single-process mode, ``vcr_recorded_by`` is the empty string,
which means the same process that ran the test is now handling the
report — ``_session_stats`` already has the entry, double-counting
would be a bug.
* Only when ``vcr_recorded_by`` is a non-empty worker id (``"gw0"``
etc.) do we know the controller's ``_session_stats`` is missing this
test and needs the outcome folded in.
"""
if os.environ.get("PYTEST_XDIST_WORKER"):
return
if report.when != "teardown":
return
recorded_by = next(
(v for k, v in (report.user_properties or []) if k == _USER_PROP_RECORDED_BY),
None,
)
if not recorded_by:
return
outcome = next(
(v for k, v in (report.user_properties or []) if k == _USER_PROP_OUTCOME),
None,
)
if not outcome:
return
verdict = outcome.get("verdict")
if not verdict:
return
nodeid = report.nodeid
_session_stats["verdict_counts"][verdict] += 1
if verdict == VERDICT_MISS_OVERFLOW:
_session_stats["overflow_tests"].append(nodeid)
elif verdict == VERDICT_UNMARKED_LIVE_CALL:
_session_stats["unmarked_live_call_tests"].append(
(nodeid, list(outcome.get("live_call_hosts") or []))
)
skip_reason = outcome.get("skip_reason")
if skip_reason:
_session_stats["skip_reason_counts"][skip_reason] += 1
examples = _session_stats["skip_reason_examples"][skip_reason]
if len(examples) < 5:
examples.append(nodeid)
def session_stats_snapshot() -> dict:
"""Read-only copy of the per-session VCR stats. Used by the summary."""
return {
"verdict_counts": dict(_session_stats["verdict_counts"]),
"overflow_tests": list(_session_stats["overflow_tests"]),
"unmarked_live_call_tests": list(_session_stats["unmarked_live_call_tests"]),
"skip_reason_counts": dict(_session_stats["skip_reason_counts"]),
"skip_reason_examples": {
k: list(v) for k, v in _session_stats["skip_reason_examples"].items()
},
}
def _classify_marked_test(cassette) -> str:
"""Map cassette state → verdict tag for tests that *were* VCR-marked."""
played = getattr(cassette, "play_count", 0) or 0
dirty = getattr(cassette, "dirty", False)
total = len(cassette) if hasattr(cassette, "__len__") else 0
# "OVERFLOW" mirrors ``_RedisPersister.save_cassette``'s
# ``> MAX_EPISODES_PER_CASSETTE`` guard. Cassettes that hit this
# threshold are refused for save, so the test re-records live every
# run. Only flag when ``dirty=True`` — if a cassette grew past the
# cap historically but this run replayed it without adding new
# episodes, the persister never tries to save (no recording
# happened), so the cache state is stable and the next run will
# replay too. Flagging that case as OVERFLOW would tag healthy
# cached tests as cost leaks.
if total > MAX_EPISODES_PER_CASSETTE and dirty:
return VERDICT_MISS_OVERFLOW
if played == 0 and not dirty:
return VERDICT_NOOP_NO_TRAFFIC
if played > 0 and not dirty:
return VERDICT_HIT
if played == 0 and dirty:
return VERDICT_MISS_RECORDED
return VERDICT_PARTIAL
def _format_verdict_line(verdict: str, cassette, extra: str = "") -> str:
if cassette is None:
return f"[{verdict}]{(' ' + extra) if extra else ''}"
played = getattr(cassette, "play_count", 0) or 0
total = len(cassette) if hasattr(cassette, "__len__") else 0
base = f"[{verdict}] played={played} entries={total}"
if extra:
base = f"{base} {extra}"
return base
# ---------------------------------------------------------------------------
# Live-call detection for tests that bypass VCR.
#
# When a test isn't VCR-marked (respx_conflict, incompatible, or just
# plain unmarked), we wrap its socket calls inside the autouse
# ``_vcr_outcome_gate`` fixture so we can flag any outbound TCP connection
# to a known LLM provider. This converts "likely live call" into
# "confirmed: this test connected to host X".
# ---------------------------------------------------------------------------
_LIVE_CALL_BUFFER_KEY = "vcr_live_call_hosts"
def _is_live_call_host(host: str) -> bool:
if not host:
return False
host = host.lower()
if any(host.startswith(p) for p in _LIVE_CALL_LOCAL_PREFIXES):
return False
if any(host.endswith(suffix) for suffix in _LIVE_CALL_HOST_SUFFIXES):
return True
# AWS Bedrock endpoints are ``bedrock-runtime[-fips].{region}.amazonaws.com``
# (region between ``bedrock-runtime`` and ``amazonaws.com``), so plain
# suffix matching can't catch them.
if host.endswith(".amazonaws.com") and host.split(".", 1)[0].startswith(
"bedrock-runtime"
):
return True
return False
class _LiveCallProbe:
"""Context manager that monkeypatches ``socket.create_connection`` and
``socket.socket.connect`` for the lifetime of a test, recording any
outbound TCP connection to a known LLM host.
We don't intercept HTTP at the application layer because that would
fight with vcrpy/respx in tests that *do* mock httpx — the socket
layer is below both, so this probe is safe regardless of what's
patched above it. We also don't raise: the goal is observability, not
a hard gate.
"""
def __init__(self) -> None:
self.hosts: list[str] = []
self._orig_create_connection = None
self._orig_socket_connect = None
def __enter__(self):
self._orig_create_connection = socket.create_connection
self._orig_socket_connect = socket.socket.connect
def _wrapped_create_connection(address, *args, **kwargs):
try:
host = address[0] if isinstance(address, tuple) else None
if host and _is_live_call_host(host) and host not in self.hosts:
self.hosts.append(host)
except Exception:
pass
return self._orig_create_connection(address, *args, **kwargs)
def _wrapped_socket_connect(sock_self, address):
try:
host = address[0] if isinstance(address, tuple) else None
if host and _is_live_call_host(host) and host not in self.hosts:
self.hosts.append(host)
except Exception:
pass
return self._orig_socket_connect(sock_self, address)
socket.create_connection = _wrapped_create_connection
socket.socket.connect = _wrapped_socket_connect
return self
def __exit__(self, *exc):
if self._orig_create_connection is not None:
socket.create_connection = self._orig_create_connection
if self._orig_socket_connect is not None:
socket.socket.connect = self._orig_socket_connect
return False
def vcr_outcome_logging_enabled() -> bool:
"""Verdict logging is on whenever VCR itself is active.
The old ``LITELLM_VCR_VERBOSE=1`` gate kept logs quiet by default, but
that hides the very signal we need to know whether a paid test ran
against a real provider. CI logs already drop a one-line verdict per
test; that's what makes the cost analysis tractable. Set
``LITELLM_VCR_VERBOSE=0`` if you really want the legacy quiet mode.
"""
if vcr_disabled():
return False
if os.environ.get(VCR_VERBOSE_ENV) == "0":
return False
return True
def record_vcr_outcome(request, vcr) -> None:
"""Call from the post-yield section of an autouse fixture per test."""
cassette = vcr
@ -507,10 +1086,78 @@ def record_vcr_outcome(request, vcr) -> None:
if cassette_path:
mark_test_outcome_for_cassette(cassette_path, test_passed)
if not vcr_verbose_enabled():
nodeid = request.node.nodeid
if cassette is not None:
verdict = _classify_marked_test(cassette)
# Track overflow tests even when verbose logging is off — the
# session summary shows them either way.
if verdict == VERDICT_MISS_OVERFLOW:
_session_stats["overflow_tests"].append(nodeid)
if not test_passed and verdict == VERDICT_MISS_RECORDED:
verdict = VERDICT_MISS_NOT_PERSISTED
_session_stats["verdict_counts"][verdict] += 1
_emit_outcome_payload(request.node, verdict)
if vcr_outcome_logging_enabled():
line = _format_verdict_line(verdict, cassette)
request.node.user_properties.append((_USER_PROP_VERDICT_LINE, line))
return
verdict = format_vcr_verdict(cassette)
request.node.user_properties.append(("vcr_verdict", verdict))
# Cassette is None ⇒ test wasn't VCR-marked. Honor the skip reason
# we tagged at collection time, and pull live-call hosts captured by
# the socket probe (if any).
skip_reason = getattr(
request.node, VCR_SKIP_REASON_USER_ATTR, SKIP_REASON_FILE_OPT_OUT
)
_session_stats["skip_reason_counts"][skip_reason] += 1
hosts = getattr(request.node, _LIVE_CALL_BUFFER_KEY, []) or []
if hosts:
verdict = VERDICT_UNMARKED_LIVE_CALL
_session_stats["unmarked_live_call_tests"].append((nodeid, list(hosts)))
extra = f"reason={skip_reason} hosts={','.join(hosts)}"
else:
verdict = VERDICT_UNMARKED_NO_TRAFFIC
extra = f"reason={skip_reason}"
_session_stats["verdict_counts"][verdict] += 1
examples = _session_stats["skip_reason_examples"][skip_reason]
if len(examples) < 5:
examples.append(nodeid)
_emit_outcome_payload(
request.node,
verdict,
skip_reason=skip_reason,
live_call_hosts=hosts,
)
if vcr_outcome_logging_enabled():
request.node.user_properties.append(
(_USER_PROP_VERDICT_LINE, _format_verdict_line(verdict, None, extra))
)
def install_live_call_probe(request, vcr) -> None:
"""Activate the live-call socket probe for non-VCR-marked tests.
Call this from inside the per-test autouse ``_vcr_outcome_gate``
fixture *before* the ``yield``. When ``vcr`` is ``None`` (test isn't
VCR-marked) we patch ``socket.connect`` for the duration of the test
and stash any LLM-host connections on ``request.node`` so
``record_vcr_outcome`` can include them in the verdict line.
Tests that *are* VCR-marked don't get the probe — vcrpy itself
intercepts above the socket layer, so any "outbound" socket would be
a recording cycle, not real spend.
"""
if vcr is not None or vcr_disabled():
return None
probe = _LiveCallProbe()
probe.__enter__()
setattr(request.node, _LIVE_CALL_BUFFER_KEY, probe.hosts)
request.addfinalizer(lambda: probe.__exit__(None, None, None))
return probe
def _format_capacity_line(snapshot: dict) -> str:
@ -525,6 +1172,99 @@ def _format_capacity_line(snapshot: dict) -> str:
)
def emit_vcr_classification_summary(terminalreporter) -> None:
"""Render the per-classification summary at session end.
Output sections (only included when non-empty):
* **Verdict counts** — full breakdown of HIT / MISS:RECORDED /
MISS:OVERFLOW / MISS:NOT_PERSISTED / PARTIAL / NOOP /
UNMARKED:LIVE_CALL / UNMARKED:NO_TRAFFIC. The OVERFLOW and
UNMARKED:LIVE_CALL counts are the cost-leak signals.
* **Cassette overflow** (>``MAX_EPISODES_PER_CASSETTE``) — these tests
fire live every CI run because the persister refuses to save them.
Usually means the request body is non-deterministic (file handle
consumed, AWS SigV4 timestamp, random UUID).
* **Unmarked tests with live API calls** — confirmed live HTTP traffic
to a known LLM host while VCR was *not* active for the test. This
is the "convert likely → confirmed" signal: each entry is real
money the cache would otherwise prevent.
* **Skip-reason breakdown** — how many tests opted out of VCR and
why (respx_conflict, respx_conflict_module, file_opt_out,
incompatible). Bare ``file_opt_out`` entries with zero respx usage
in the module are dead skip-list rows worth pruning.
"""
if vcr_disabled():
return
if os.environ.get("PYTEST_XDIST_WORKER"):
return
snapshot = session_stats_snapshot()
counts = snapshot["verdict_counts"]
if not counts:
return
terminalreporter.write_sep("=", "VCR CACHE CLASSIFICATION SUMMARY", bold=True)
for verdict in (
VERDICT_HIT,
VERDICT_PARTIAL,
VERDICT_MISS_RECORDED,
VERDICT_MISS_OVERFLOW,
VERDICT_MISS_NOT_PERSISTED,
VERDICT_NOOP_NO_TRAFFIC,
VERDICT_UNMARKED_NO_TRAFFIC,
VERDICT_UNMARKED_LIVE_CALL,
):
n = counts.get(verdict, 0)
if not n:
continue
terminalreporter.write_line(f" [{verdict}] {n}")
overflow = snapshot["overflow_tests"]
if overflow:
terminalreporter.write_sep(
"-",
f"CASSETTE OVERFLOW (>{MAX_EPISODES_PER_CASSETTE} episodes, save refused)",
red=True,
bold=True,
)
terminalreporter.write_line(
" These tests will hit the live provider on every CI run "
"because the persister won't save cassettes that grew past "
"the limit. Stabilize the request body (file handle consumed, "
"SigV4 timestamp, UUID, or boundary leak)."
)
for nodeid in overflow:
terminalreporter.write_line(f" - {nodeid}")
live_calls = snapshot["unmarked_live_call_tests"]
if live_calls:
terminalreporter.write_sep(
"-",
"UNMARKED TESTS WITH LIVE API CALLS",
red=True,
bold=True,
)
terminalreporter.write_line(
" These tests connected to a real LLM provider host while "
"they were NOT VCR-marked. Either add @pytest.mark.vcr "
"explicitly, mock with respx, or move them off the "
"respx_conflict / incompatible skip list."
)
for nodeid, hosts in live_calls:
terminalreporter.write_line(f" - {nodeid} → {','.join(hosts)}")
reasons = snapshot["skip_reason_counts"]
if reasons:
terminalreporter.write_sep("-", "SKIP-REASON BREAKDOWN", bold=True)
for reason, n in sorted(reasons.items(), key=lambda kv: -kv[1]):
examples = snapshot["skip_reason_examples"].get(reason, [])
terminalreporter.write_line(f" {reason}: {n}")
for ex in examples:
terminalreporter.write_line(f" - {ex}")
terminalreporter.write_sep("=", bold=True)
def emit_cassette_cache_session_banner(terminalreporter) -> None:
"""Call from ``pytest_terminal_summary``. No-op on xdist workers."""
if vcr_disabled():
@ -596,17 +1336,28 @@ class VerboseReporterState:
return self.terminal_reporter
def maybe_emit_verdict(self, report) -> None:
# Aggregate xdist-worker stats into the controller's session counters
# first — this path is independent of verbose logging because the
# structured outcome payload is always attached when VCR is active,
# and ``aggregate_report_outcome`` no-ops outside the xdist-controller
# case on its own.
aggregate_report_outcome(report)
if report.when != "teardown":
return
if os.environ.get("PYTEST_XDIST_WORKER"):
return
if not vcr_verbose_enabled():
if not vcr_outcome_logging_enabled():
return
reporter = self.resolve_terminal_reporter()
if reporter is None:
return
verdict = next(
(v for k, v in (report.user_properties or []) if k == "vcr_verdict"),
(
v
for k, v in (report.user_properties or [])
if k == _USER_PROP_VERDICT_LINE
),
None,
)
if not verdict:

9
tests/audio_tests/conftest.py
View File

@ -8,6 +8,9 @@ sys.path.insert(0, os.path.abspath("../.."))
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
@ -34,6 +37,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@ -48,3 +52,8 @@ def pytest_runtest_logreport(report):
def pytest_collection_modifyitems(config, items):
apply_vcr_auto_marker_to_items(items)
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)

9
tests/guardrails_tests/conftest.py
View File

@ -19,6 +19,9 @@ import litellm
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
@ -45,6 +48,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@ -151,3 +155,8 @@ def pytest_collection_modifyitems(config, items):
# Reorder the items list
items[:] = custom_logger_tests + other_tests
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)

9
tests/image_gen_tests/conftest.py
View File

@ -12,6 +12,9 @@ import litellm # noqa: E402,F401
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
@ -48,6 +51,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@ -62,3 +66,8 @@ def pytest_runtest_logreport(report):
def pytest_collection_modifyitems(config, items):
apply_vcr_auto_marker_to_items(items)
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)

76
tests/image_gen_tests/test_image_edits.py
View File

@ -102,22 +102,47 @@ class BaseLLMImageEditTest(ABC):
# Get the current directory of the file being run
pwd = os.path.dirname(os.path.realpath(__file__))
TEST_IMAGES = [
open(os.path.join(pwd, "ishaan_github.png"), "rb"),
open(os.path.join(pwd, "litellm_site.png"), "rb"),
]
SINGLE_TEST_IMAGE = open(os.path.join(pwd, "ishaan_github.png"), "rb")
# Image fixtures must be regenerated per access — module-level
# ``open(...)`` handles get consumed after a single multipart upload, leaving
# subsequent tests in the same process to send empty bodies. That non-determinism
# (a) blows the recorded cassette past ``MAX_EPISODES_PER_CASSETTE`` so the
# persister refuses to save (see ``tests/_vcr_redis_persister.py``), and
# (b) re-bills the live image edit endpoint on every CI run.
def _read_image_bytes(filename: str) -> bytes:
with open(os.path.join(pwd, filename), "rb") as f:
return f.read()
_ISHAAN_GITHUB_BYTES = _read_image_bytes("ishaan_github.png")
_LITELLM_SITE_BYTES = _read_image_bytes("litellm_site.png")
def _make_test_images() -> list:
"""Return a fresh pair of image streams seeded with the fixture bytes.
Use this everywhere you'd previously have used the module-level
``TEST_IMAGES``. Each call returns brand new ``BytesIO`` objects whose
file pointers start at 0, so multipart uploads encode the full image
bytes on every test invocation. Parametrized and ``flaky``-retried
test methods call ``get_base_image_edit_call_args`` once per
invocation, so a fresh stream per call is sufficient — the factory
must not auto-rewind on EOF or the SDK's multipart writer will read
the same bytes forever (worker OOM).
"""
return [
BytesIO(_ISHAAN_GITHUB_BYTES),
BytesIO(_LITELLM_SITE_BYTES),
]
def _make_single_test_image() -> BytesIO:
return BytesIO(_ISHAAN_GITHUB_BYTES)
def get_test_images_as_bytesio():
"""Helper function to get test images as BytesIO objects"""
bytesio_images = []
for image_path in ["ishaan_github.png", "litellm_site.png"]:
with open(os.path.join(pwd, image_path), "rb") as f:
image_bytes = f.read()
bytesio_images.append(BytesIO(image_bytes))
return bytesio_images
return _make_test_images()
class TestOpenAIImageEditGPTImage1(BaseLLMImageEditTest):
@ -129,7 +154,7 @@ class TestOpenAIImageEditGPTImage1(BaseLLMImageEditTest):
"""Return base call args for OpenAI image edit"""
return {
"model": "gpt-image-1",
"image": TEST_IMAGES,
"image": _make_test_images(),
}
@ -143,7 +168,7 @@ class TestAzureAIFlux2ImageEdit(BaseLLMImageEditTest):
"""Return base call args for Azure AI FLUX 2 image edit"""
return {
"model": "azure_ai/flux.2-pro",
"image": SINGLE_TEST_IMAGE,
"image": _make_single_test_image(),
"api_base": os.getenv("AZURE_AI_API_BASE"),
"api_key": os.getenv("AZURE_AI_API_KEY"),
"api_version": "preview",
@ -171,7 +196,7 @@ async def test_openai_image_edit_litellm_router():
result = await router.aimage_edit(
prompt=prompt,
model="gpt-image-1",
image=TEST_IMAGES,
image=_make_test_images(),
)
print("result from image edit", result)
@ -275,7 +300,7 @@ async def test_azure_image_edit_litellm_sdk():
api_base=test_api_base,
api_key=test_api_key,
api_version=test_api_version,
image=TEST_IMAGES,
image=_make_test_images(),
)
# Verify the request was made correctly
@ -389,7 +414,7 @@ async def test_openai_image_edit_cost_tracking():
result = await aimage_edit(
prompt=prompt,
model="openai/gpt-image-1",
image=TEST_IMAGES,
image=_make_test_images(),
)
# Verify the request was made correctly
@ -480,7 +505,7 @@ async def test_azure_image_edit_cost_tracking():
prompt=prompt,
model="azure/CUSTOM_AZURE_DEPLOYMENT_NAME",
base_model="azure/gpt-image-1",
image=TEST_IMAGES,
image=_make_test_images(),
)
# Verify the request was made correctly
@ -528,7 +553,6 @@ async def test_recraft_image_edit_api():
import requests
litellm._turn_on_debug()
global TEST_IMAGES
try:
prompt = """
Create a studio ghibli style image that combines all the reference images. Make sure the person looks like a CTO.
@ -536,7 +560,7 @@ async def test_recraft_image_edit_api():
result = await aimage_edit(
prompt=prompt,
model="recraft/recraftv3",
image=TEST_IMAGES,
image=_make_test_images(),
)
print("result from image edit", result)
@ -634,13 +658,13 @@ async def test_multiple_vs_single_image_edit(sync_mode):
single_result = image_edit(
prompt=prompt,
model="gpt-image-1",
image=SINGLE_TEST_IMAGE,
image=_make_single_test_image(),
)
else:
single_result = await aimage_edit(
prompt=prompt,
model="gpt-image-1",
image=SINGLE_TEST_IMAGE,
image=_make_single_test_image(),
)
print("Single image result:", single_result)
@ -651,13 +675,13 @@ async def test_multiple_vs_single_image_edit(sync_mode):
multiple_result = image_edit(
prompt=prompt,
model="gpt-image-1",
image=TEST_IMAGES,
image=_make_test_images(),
)
else:
multiple_result = await aimage_edit(
prompt=prompt,
model="gpt-image-1",
image=TEST_IMAGES,
image=_make_test_images(),
)
print("Multiple images result:", multiple_result)
@ -688,7 +712,7 @@ async def test_multiple_image_edit_with_different_formats():
# Test with mixed BytesIO and file objects
mixed_images = [
SINGLE_TEST_IMAGE, # File object
_make_single_test_image(), # File object
get_test_images_as_bytesio()[1], # BytesIO object
]
@ -752,14 +776,14 @@ async def test_image_edit_array_handling():
result1 = await aimage_edit(
prompt=prompt,
model="gpt-image-1",
image=SINGLE_TEST_IMAGE,
image=_make_single_test_image(),
)
# Test 2: Multiple images (already a list)
result2 = await aimage_edit(
prompt=prompt,
model="gpt-image-1",
image=TEST_IMAGES,
image=_make_test_images(),
)
# Both valid calls should succeed

9
tests/litellm_utils_tests/conftest.py
View File

@ -15,6 +15,9 @@ import litellm # noqa: E402,F401
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
@ -76,6 +79,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@ -107,3 +111,8 @@ def pytest_collection_modifyitems(config, items):
# Reorder the items list
items[:] = custom_logger_tests + other_tests
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)

9
tests/llm_responses_api_testing/conftest.py
View File

@ -16,6 +16,9 @@ import litellm # noqa: E402
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
@ -42,6 +45,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@ -107,3 +111,8 @@ def pytest_collection_modifyitems(config, items):
other_tests.sort(key=lambda x: x.name)
items[:] = custom_logger_tests + other_tests
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)

31
tests/llm_translation/conftest.py
View File

@ -21,27 +21,20 @@ import litellm # noqa: E402
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
)
# vcrpy and respx both patch the httpx transport — applying both makes one
# silently win, so respx-using files opt out of the auto-marker.
_RESPX_CONFLICTING_FILES = frozenset(
{
"test_gpt4o_audio.py",
"test_nvidia_nim.py",
"test_openai.py",
"test_openai_o1.py",
"test_prompt_caching.py",
"test_text_completion_unit_tests.py",
"test_xai.py",
}
)
_VCR_AUTO_MARKER_SKIP_FILES = _RESPX_CONFLICTING_FILES | frozenset(
{"test_vcr_redis_persister.py"}
)
# Per-item respx detection (``apply_vcr_auto_marker_to_items``) handles
# the vast majority of respx-vs-vcrpy conflicts automatically. The only
# entry below is the persister's own unit-test file, which exercises
# ``save_cassette`` / ``load_cassette`` against fakeredis and must not
# itself run under a live cassette context.
_VCR_AUTO_MARKER_SKIP_FILES = frozenset({"test_vcr_redis_persister.py"})
# Tests that observe live cross-call provider state (e.g. prompt-cache
# warm-up between two consecutive calls); replay can't reproduce that state.
@ -73,6 +66,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@ -85,6 +79,11 @@ def pytest_runtest_logreport(report):
_verbose_state.maybe_emit_verdict(report)
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)
# ---------------------------------------------------------------------------
# Capture TRUE defaults at conftest import time (before test modules pollute).
# ---------------------------------------------------------------------------

1
tests/llm_translation/test_gpt4o_audio.py
View File

@ -11,7 +11,6 @@ sys.path.insert(
import httpx
import pytest
from respx import MockRouter
import litellm
from litellm import Choices, Message, ModelResponse

1
tests/llm_translation/test_nvidia_nim.py
View File

@ -11,7 +11,6 @@ sys.path.insert(
import httpx
import pytest
from respx import MockRouter
from unittest.mock import patch, MagicMock, AsyncMock
import litellm

1
tests/llm_translation/test_openai.py
View File

@ -12,7 +12,6 @@ sys.path.insert(
import httpx
import pytest
from respx import MockRouter
import litellm
from litellm import Choices, Message, ModelResponse

1
tests/llm_translation/test_openai_o1.py
View File

@ -11,7 +11,6 @@ sys.path.insert(
import httpx
import pytest
from respx import MockRouter
import litellm
from litellm import Choices, Message, ModelResponse

1
tests/llm_translation/test_prompt_caching.py
View File

@ -11,7 +11,6 @@ sys.path.insert(
import httpx
import pytest
from respx import MockRouter
import litellm
from litellm import Choices, Message, ModelResponse

804
tests/llm_translation/test_vcr_classification.py Normal file
View File

@ -0,0 +1,804 @@
"""Unit tests for the VCR classification + observability layer.
Covers:
- per-item respx detection (module scan, marker, fixture)
- skip-reason tagging in ``apply_vcr_auto_marker_to_items``
- verdict classification (HIT / MISS:RECORDED / MISS:OVERFLOW / MISS:NOT_PERSISTED /
PARTIAL / NOOP / UNMARKED:LIVE_CALL / UNMARKED:NO_TRAFFIC)
- AWS SigV4 fingerprint stability
- session-end summary rendering
- live-call host classification
"""
from __future__ import annotations
import os
import sys
from types import SimpleNamespace
from typing import Optional
import pytest
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
from tests._vcr_conftest_common import ( # noqa: E402
SKIP_REASON_FILE_OPT_OUT,
SKIP_REASON_INCOMPATIBLE,
SKIP_REASON_PRE_MARKED,
SKIP_REASON_RESPX,
SKIP_REASON_RESPX_MODULE,
VCR_SKIP_REASON_USER_ATTR,
VERDICT_HIT,
VERDICT_MISS_NOT_PERSISTED,
VERDICT_MISS_OVERFLOW,
VERDICT_MISS_RECORDED,
VERDICT_NOOP_NO_TRAFFIC,
VERDICT_PARTIAL,
VERDICT_UNMARKED_LIVE_CALL,
VERDICT_UNMARKED_NO_TRAFFIC,
_RESPX_MODULE_CACHE,
_classify_marked_test,
_compute_key_fingerprint,
_is_live_call_host,
_reset_session_stats,
_stable_key_value,
aggregate_report_outcome,
apply_vcr_auto_marker_to_items,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
session_stats_snapshot,
)
# ---------------------------------------------------------------------------
# Test doubles
# ---------------------------------------------------------------------------
class _StubItem:
"""Pytest item double sufficient for the auto-marker logic."""
def __init__(
self,
nodeid: str,
path: str,
*,
markers: Optional[list[str]] = None,
fixturenames: Optional[list[str]] = None,
module=None,
) -> None:
self.nodeid = nodeid
self.path = path
self._markers = list(markers or [])
self.fixturenames = list(fixturenames or [])
self.module = module
self.user_properties: list = []
def get_closest_marker(self, name: str):
return name if name in self._markers else None
def add_marker(self, marker):
# ``pytest.mark.vcr`` is a MarkDecorator; rely on its ``name``.
name = getattr(marker, "name", str(marker))
self._markers.append(name)
@pytest.fixture
def vcr_enabled(monkeypatch):
monkeypatch.setenv("CASSETTE_REDIS_URL", "redis://stub")
monkeypatch.delenv("LITELLM_VCR_DISABLE", raising=False)
monkeypatch.delenv("PYTEST_XDIST_WORKER", raising=False)
@pytest.fixture(autouse=True)
def _reset_module_caches():
_reset_session_stats()
_RESPX_MODULE_CACHE.clear()
yield
_reset_session_stats()
_RESPX_MODULE_CACHE.clear()
# ---------------------------------------------------------------------------
# AWS SigV4 fingerprint stability — the Bedrock cassette overflow root cause
# ---------------------------------------------------------------------------
def test_should_extract_only_aws_access_key_from_sigv4_authorization():
"""Two Bedrock requests with the same access key but different
timestamps and signatures must produce the same fingerprint, otherwise
every CI run pushes a new episode into the cassette."""
auth_today = (
"AWS4-HMAC-SHA256 Credential=AKIAEXAMPLE12345/20260512/us-east-1/"
"bedrock/aws4_request, SignedHeaders=host;x-amz-date, "
"Signature=AAAAAAAA"
)
auth_tomorrow = (
"AWS4-HMAC-SHA256 Credential=AKIAEXAMPLE12345/20260513/us-east-1/"
"bedrock/aws4_request, SignedHeaders=host;x-amz-date, "
"Signature=BBBBBBBB"
)
today = _stable_key_value("Authorization", auth_today)
tomorrow = _stable_key_value("Authorization", auth_tomorrow)
assert today == tomorrow == "aws-sigv4:AKIAEXAMPLE12345"
def test_should_keep_bearer_authorization_unchanged():
"""OpenAI ``Bearer <key>`` headers are stable as-is — keep them."""
out = _stable_key_value("Authorization", "Bearer sk-1234")
assert out == "Bearer sk-1234"
def test_should_produce_stable_fingerprint_across_sigv4_signatures():
"""``_compute_key_fingerprint`` should not change when only the SigV4
signature/timestamp rotates."""
req_a = SimpleNamespace(
headers={
"authorization": (
"AWS4-HMAC-SHA256 Credential=AKIA1/20260101/us-east-1/"
"bedrock/aws4_request, SignedHeaders=host, Signature=AAA"
)
}
)
req_b = SimpleNamespace(
headers={
"authorization": (
"AWS4-HMAC-SHA256 Credential=AKIA1/20260512/us-east-1/"
"bedrock/aws4_request, SignedHeaders=host;x-amz-date, "
"Signature=ZZZ"
)
}
)
assert _compute_key_fingerprint(req_a) == _compute_key_fingerprint(req_b)
def test_should_distinguish_different_aws_access_keys():
"""Two different access keys must produce different fingerprints so
cassettes recorded under one identity never serve another."""
req_a = SimpleNamespace(
headers={
"authorization": "AWS4-HMAC-SHA256 Credential=AKIAONE/x/y/z/aws4_request, Signature=A"
}
)
req_b = SimpleNamespace(
headers={
"authorization": "AWS4-HMAC-SHA256 Credential=AKIATWO/x/y/z/aws4_request, Signature=A"
}
)
assert _compute_key_fingerprint(req_a) != _compute_key_fingerprint(req_b)
# ---------------------------------------------------------------------------
# Live-call host classification
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"host,expected",
[
("api.openai.com", True),
("api.anthropic.com", True),
("bedrock-runtime.us-east-1.amazonaws.com", True),
("bedrock-runtime-fips.us-east-1.amazonaws.com", True),
("api.us-east-1.bedrock-runtime.amazonaws.com", False),
("foo.bar.openai.com", True),
("127.0.0.1", False),
("localhost", False),
("10.0.0.1", False),
("172.16.0.1", False),
("redis.example.com", False),
("", False),
],
)
def test_should_classify_live_call_hosts(host, expected):
assert _is_live_call_host(host) is expected
# ---------------------------------------------------------------------------
# Verdict classification
# ---------------------------------------------------------------------------
def _cassette(played: int, dirty: bool, total: int):
class _Sized:
def __init__(self, n):
self.n = n
self.play_count = played
self.dirty = dirty
def __len__(self):
return self.n
return _Sized(total)
def test_should_classify_pure_replay_as_hit():
assert (
_classify_marked_test(_cassette(played=3, dirty=False, total=3)) == VERDICT_HIT
)
def test_should_classify_no_traffic_as_noop():
assert (
_classify_marked_test(_cassette(played=0, dirty=False, total=0))
== VERDICT_NOOP_NO_TRAFFIC
)
def test_should_classify_pure_record_as_miss_recorded():
assert (
_classify_marked_test(_cassette(played=0, dirty=True, total=1))
== VERDICT_MISS_RECORDED
)
def test_should_classify_mixed_replay_and_record_as_partial():
assert (
_classify_marked_test(_cassette(played=2, dirty=True, total=4))
== VERDICT_PARTIAL
)
def test_should_classify_overflow_only_when_dirty_episodes_were_recorded():
"""Cassettes that exceed ``MAX_EPISODES_PER_CASSETTE`` (50) are
refused for save — but only when ``dirty=True`` (new episodes were
actually recorded that the persister would refuse). Replaying an
already-large cassette with no new traffic is healthy: the persister
never tries to save, so the cache state is stable and the next run
will replay too."""
assert (
_classify_marked_test(_cassette(played=0, dirty=True, total=51))
== VERDICT_MISS_OVERFLOW
)
assert (
_classify_marked_test(_cassette(played=10, dirty=True, total=52))
== VERDICT_MISS_OVERFLOW
)
def test_should_classify_large_cassette_with_no_new_episodes_as_hit():
"""``total > 50`` + ``dirty=False`` means everything was replayed
from cache; no save attempt happens, so this is a healthy HIT, not
OVERFLOW."""
assert (
_classify_marked_test(_cassette(played=51, dirty=False, total=51))
== VERDICT_HIT
)
assert (
_classify_marked_test(_cassette(played=60, dirty=False, total=60))
== VERDICT_HIT
)
# ---------------------------------------------------------------------------
# apply_vcr_auto_marker_to_items: skip-reason tagging
# ---------------------------------------------------------------------------
def _make_module_with_source(tmp_path, src: str, name: str):
p = tmp_path / f"{name}.py"
p.write_text(src)
mod = SimpleNamespace(__file__=str(p))
return mod, str(p)
def test_should_apply_vcr_marker_to_clean_test(vcr_enabled, tmp_path):
mod, p = _make_module_with_source(tmp_path, "def test_x(): pass\n", "clean")
item = _StubItem("clean.py::test_x", p, module=mod)
apply_vcr_auto_marker_to_items([item])
assert item.get_closest_marker("vcr") == "vcr"
def test_should_skip_per_item_when_respx_marker_present(vcr_enabled, tmp_path):
mod, p = _make_module_with_source(tmp_path, "def test_x(): pass\n", "respx_marker")
item = _StubItem("respx_marker.py::test_x", p, markers=["respx"], module=mod)
apply_vcr_auto_marker_to_items([item])
assert item.get_closest_marker("vcr") is None
assert getattr(item, VCR_SKIP_REASON_USER_ATTR) == SKIP_REASON_RESPX
def test_should_skip_per_item_when_respx_mock_fixture_present(vcr_enabled, tmp_path):
mod, p = _make_module_with_source(tmp_path, "def test_x(): pass\n", "respx_fixture")
item = _StubItem(
"respx_fixture.py::test_x", p, fixturenames=["respx_mock"], module=mod
)
apply_vcr_auto_marker_to_items([item])
assert item.get_closest_marker("vcr") is None
assert getattr(item, VCR_SKIP_REASON_USER_ATTR) == SKIP_REASON_RESPX
def test_should_tag_pre_marked_items_so_summary_can_show_them(vcr_enabled, tmp_path):
mod, p = _make_module_with_source(tmp_path, "def test_x(): pass\n", "premarked")
item = _StubItem("premarked.py::test_x", p, markers=["vcr"], module=mod)
apply_vcr_auto_marker_to_items([item])
assert getattr(item, VCR_SKIP_REASON_USER_ATTR) == SKIP_REASON_PRE_MARKED
def test_should_tag_skip_files_with_respx_module_when_module_actually_uses_respx(
vcr_enabled, tmp_path
):
"""A file in ``skip_files`` whose module *does* call respx should be
labeled as a real conflict (respx_conflict_module), not a dead opt-out."""
mod, p = _make_module_with_source(
tmp_path,
"import respx\n@pytest.mark.respx\ndef test_x(): pass\n",
"real_respx",
)
item = _StubItem("real_respx.py::test_x", p, module=mod)
apply_vcr_auto_marker_to_items([item], skip_files={"real_respx.py"})
assert getattr(item, VCR_SKIP_REASON_USER_ATTR) == SKIP_REASON_RESPX_MODULE
def test_should_tag_skip_files_with_file_opt_out_when_module_does_not_use_respx(
vcr_enabled, tmp_path
):
"""A file in ``skip_files`` whose module never wires up respx is a
dead skip-list entry — surface it so we can prune."""
mod, p = _make_module_with_source(
tmp_path,
"from respx import MockRouter # dead import\ndef test_x(): pass\n",
"dead_skip",
)
item = _StubItem("dead_skip.py::test_x", p, module=mod)
apply_vcr_auto_marker_to_items([item], skip_files={"dead_skip.py"})
assert getattr(item, VCR_SKIP_REASON_USER_ATTR) == SKIP_REASON_FILE_OPT_OUT
def test_should_not_flag_respx_mentioned_in_comment_or_docstring(vcr_enabled, tmp_path):
"""Substring scans of source text false-positive on
``# Previously used respx.mock`` and similar — defeats the dead
skip-list pruning goal. AST-based detection ignores comments and
string literals."""
src = (
'"""Module docstring mentions respx.mock and @pytest.mark.respx and respx_mock."""\n'
"# Previously tried respx.mock but switched to vcrpy\n"
"# Old code did `with respx.mock(): ...`\n"
"x = '@respx.mock' # string literal, not a real decorator\n"
"def test_x():\n"
" pass\n"
)
mod, p = _make_module_with_source(tmp_path, src, "comment_respx")
item = _StubItem("comment_respx.py::test_x", p, module=mod)
apply_vcr_auto_marker_to_items([item], skip_files={"comment_respx.py"})
assert getattr(item, VCR_SKIP_REASON_USER_ATTR) == SKIP_REASON_FILE_OPT_OUT
def test_should_flag_real_respx_mark_decorator_via_ast(vcr_enabled, tmp_path):
src = "import pytest\n" "@pytest.mark.respx\n" "def test_x(respx_mock): pass\n"
mod, p = _make_module_with_source(tmp_path, src, "real_respx_mark")
item = _StubItem("real_respx_mark.py::test_x", p, module=mod)
apply_vcr_auto_marker_to_items([item], skip_files={"real_respx_mark.py"})
assert getattr(item, VCR_SKIP_REASON_USER_ATTR) == SKIP_REASON_RESPX_MODULE
def test_should_flag_real_respx_with_block_via_ast(vcr_enabled, tmp_path):
src = "import respx\n" "def test_x():\n" " with respx.mock():\n" " pass\n"
mod, p = _make_module_with_source(tmp_path, src, "real_respx_with")
item = _StubItem("real_respx_with.py::test_x", p, module=mod)
apply_vcr_auto_marker_to_items([item], skip_files={"real_respx_with.py"})
assert getattr(item, VCR_SKIP_REASON_USER_ATTR) == SKIP_REASON_RESPX_MODULE
def test_should_flag_respx_mock_call_at_module_scope_via_ast(vcr_enabled, tmp_path):
src = "import respx\nmock = respx.mock()\ndef test_x(): pass\n"
mod, p = _make_module_with_source(tmp_path, src, "real_respx_call")
item = _StubItem("real_respx_call.py::test_x", p, module=mod)
apply_vcr_auto_marker_to_items([item], skip_files={"real_respx_call.py"})
assert getattr(item, VCR_SKIP_REASON_USER_ATTR) == SKIP_REASON_RESPX_MODULE
def test_should_tag_nodeid_suffix_skips_as_incompatible(vcr_enabled, tmp_path):
mod, p = _make_module_with_source(tmp_path, "def test_x(): pass\n", "incompat")
item = _StubItem("incompat.py::test_prompt_caching", p, module=mod)
apply_vcr_auto_marker_to_items(
[item], skip_nodeid_suffixes=("::test_prompt_caching",)
)
assert getattr(item, VCR_SKIP_REASON_USER_ATTR) == SKIP_REASON_INCOMPATIBLE
# ---------------------------------------------------------------------------
# Session-end summary
# ---------------------------------------------------------------------------
class _FakeReporter:
def __init__(self):
self.lines: list[str] = []
def write_sep(self, sep, title="", **kwargs):
self.lines.append(f"=== {title}" if title else "===")
def write_line(self, line):
self.lines.append(line)
@property
def output(self):
return "\n".join(self.lines)
def test_should_render_overflow_section_when_any_test_overflowed(vcr_enabled):
"""The OVERFLOW section is the cost-leak signal: if it's empty, no
cassettes are silently being refused; if it's not empty, those tests
re-bill on every run."""
request = SimpleNamespace(
node=SimpleNamespace(
nodeid="t::overflow",
user_properties=[],
rep_call=SimpleNamespace(passed=True),
)
)
cassette = _cassette(played=0, dirty=True, total=51)
cassette._path = None # avoid mark_test_outcome side-effects
record_vcr_outcome(request, cassette)
reporter = _FakeReporter()
emit_vcr_classification_summary(reporter)
assert "VCR CACHE CLASSIFICATION SUMMARY" in reporter.output
assert "VCR MISS:OVERFLOW" in reporter.output
assert "CASSETTE OVERFLOW" in reporter.output
assert "t::overflow" in reporter.output
def test_should_render_unmarked_live_call_section_with_hosts(vcr_enabled):
request_node = SimpleNamespace(
nodeid="t::leak",
user_properties=[],
rep_call=SimpleNamespace(passed=True),
)
setattr(request_node, VCR_SKIP_REASON_USER_ATTR, SKIP_REASON_RESPX)
setattr(request_node, "vcr_live_call_hosts", ["api.openai.com"])
request = SimpleNamespace(node=request_node)
record_vcr_outcome(request, None)
snap = session_stats_snapshot()
assert snap["unmarked_live_call_tests"] == [("t::leak", ["api.openai.com"])]
assert snap["verdict_counts"][VERDICT_UNMARKED_LIVE_CALL] == 1
reporter = _FakeReporter()
emit_vcr_classification_summary(reporter)
assert "UNMARKED TESTS WITH LIVE API CALLS" in reporter.output
assert "api.openai.com" in reporter.output
assert "t::leak" in reporter.output
def test_should_record_unmarked_no_traffic_when_test_skipped_vcr_but_did_not_call_out(
vcr_enabled,
):
request_node = SimpleNamespace(
nodeid="t::clean_skip",
user_properties=[],
rep_call=SimpleNamespace(passed=True),
)
setattr(request_node, VCR_SKIP_REASON_USER_ATTR, SKIP_REASON_INCOMPATIBLE)
request = SimpleNamespace(node=request_node)
record_vcr_outcome(request, None)
snap = session_stats_snapshot()
assert snap["verdict_counts"][VERDICT_UNMARKED_NO_TRAFFIC] == 1
assert snap["skip_reason_counts"][SKIP_REASON_INCOMPATIBLE] == 1
def test_should_demote_miss_recorded_to_not_persisted_when_test_failed(vcr_enabled):
"""If a test failed, ``save_cassette`` skips persisting — that means
the next CI run will hit live again. The verdict must reflect that."""
request = SimpleNamespace(
node=SimpleNamespace(
nodeid="t::failed",
user_properties=[],
rep_call=SimpleNamespace(passed=False),
)
)
cassette = _cassette(played=0, dirty=True, total=1)
cassette._path = None
record_vcr_outcome(request, cassette)
snap = session_stats_snapshot()
assert snap["verdict_counts"].get(VERDICT_MISS_NOT_PERSISTED) == 1
def test_should_emit_no_summary_when_no_tests_observed(vcr_enabled):
reporter = _FakeReporter()
emit_vcr_classification_summary(reporter)
assert reporter.output == ""
# ---------------------------------------------------------------------------
# xdist controller aggregation
#
# _session_stats lives in module-global memory. Under xdist that memory is
# per-worker, so the controller's pytest_terminal_summary would render an
# empty summary without these aggregation hooks. The tests below simulate
# the controller receiving teardown reports produced by workers.
# ---------------------------------------------------------------------------
def _worker_report(nodeid: str, user_properties, *, when: str = "teardown"):
"""Stand-in for a pytest TestReport delivered to the xdist controller.
Only the attributes ``aggregate_report_outcome`` reads (``nodeid``,
``when``, ``user_properties``) are populated.
"""
return SimpleNamespace(
nodeid=nodeid,
when=when,
user_properties=list(user_properties),
)
def _outcome_from_worker(
verdict: str,
*,
worker_id: str = "gw0",
skip_reason=None,
live_call_hosts=None,
):
"""Build the ``user_properties`` list a worker-side ``record_vcr_outcome``
would attach. ``worker_id=""`` simulates the single-process case where
the same process that ran the test is handling the report."""
return [
(
"vcr_outcome",
{
"verdict": verdict,
"skip_reason": skip_reason,
"live_call_hosts": list(live_call_hosts) if live_call_hosts else [],
},
),
("vcr_recorded_by", worker_id),
]
def test_controller_aggregates_hit_outcome_from_worker_report(vcr_enabled):
"""An xdist controller starts with an empty _session_stats; a teardown
report carrying a worker-produced ``vcr_outcome`` must populate the
controller's verdict counts so the session summary has data to render."""
report = _worker_report(
"t::hit",
_outcome_from_worker(VERDICT_HIT),
)
aggregate_report_outcome(report)
snap = session_stats_snapshot()
assert snap["verdict_counts"][VERDICT_HIT] == 1
def test_controller_records_overflow_nodeid_from_worker_report(vcr_enabled):
"""OVERFLOW outcomes from workers must also populate
``overflow_tests`` (the named-list the summary surfaces)."""
report = _worker_report(
"t::bedrock_overflow",
_outcome_from_worker(VERDICT_MISS_OVERFLOW),
)
aggregate_report_outcome(report)
snap = session_stats_snapshot()
assert snap["verdict_counts"][VERDICT_MISS_OVERFLOW] == 1
assert snap["overflow_tests"] == ["t::bedrock_overflow"]
def test_controller_records_live_call_hosts_from_worker_report(vcr_enabled):
"""LIVE_CALL outcomes must round-trip the destination hosts so the
summary's 'UNMARKED TESTS WITH LIVE API CALLS' section has the same
detail it would in single-process mode."""
report = _worker_report(
"t::prompt_caching",
_outcome_from_worker(
VERDICT_UNMARKED_LIVE_CALL,
skip_reason=SKIP_REASON_INCOMPATIBLE,
live_call_hosts=["api.anthropic.com", "api.x.ai"],
),
)
aggregate_report_outcome(report)
snap = session_stats_snapshot()
assert snap["verdict_counts"][VERDICT_UNMARKED_LIVE_CALL] == 1
assert snap["unmarked_live_call_tests"] == [
("t::prompt_caching", ["api.anthropic.com", "api.x.ai"])
]
assert snap["skip_reason_counts"][SKIP_REASON_INCOMPATIBLE] == 1
assert "t::prompt_caching" in snap["skip_reason_examples"][SKIP_REASON_INCOMPATIBLE]
def test_controller_does_not_double_count_single_process_reports(vcr_enabled):
"""In single-process mode, ``record_vcr_outcome`` updates
``_session_stats`` in the same process that later handles the report.
The aggregator must detect this (via empty ``vcr_recorded_by``) and
skip — otherwise every verdict would be counted twice."""
report = _worker_report(
"t::single_proc",
_outcome_from_worker(VERDICT_HIT, worker_id=""),
)
aggregate_report_outcome(report)
snap = session_stats_snapshot()
assert snap["verdict_counts"] == {}
def test_controller_ignores_reports_without_vcr_outcome(vcr_enabled):
"""Tests outside the VCR plumbing (e.g. when VCR is disabled, or unit
tests that never went through ``_vcr_outcome_gate``) produce reports
with no ``vcr_outcome`` user property. The aggregator must no-op."""
report = _worker_report("t::unrelated", [("other", "value")])
aggregate_report_outcome(report)
snap = session_stats_snapshot()
assert snap["verdict_counts"] == {}
def test_controller_ignores_non_teardown_phases(vcr_enabled):
"""Only the teardown report carries the final outcome; setup/call
reports must not contribute to the counts."""
for phase in ("setup", "call"):
report = _worker_report(
"t::phase",
_outcome_from_worker(VERDICT_HIT),
when=phase,
)
aggregate_report_outcome(report)
snap = session_stats_snapshot()
assert snap["verdict_counts"] == {}
def test_controller_no_ops_when_running_inside_xdist_worker(vcr_enabled, monkeypatch):
"""Workers update their own ``_session_stats`` directly via
``record_vcr_outcome`` — re-aggregating from the report would
double-count their own work. The aggregator must bail when
``PYTEST_XDIST_WORKER`` is set."""
monkeypatch.setenv("PYTEST_XDIST_WORKER", "gw3")
report = _worker_report(
"t::on_worker",
_outcome_from_worker(VERDICT_HIT, worker_id="gw3"),
)
aggregate_report_outcome(report)
snap = session_stats_snapshot()
assert snap["verdict_counts"] == {}
def test_controller_aggregated_outcomes_drive_session_summary(vcr_enabled):
"""End-to-end: with only worker-produced reports (no in-process
``record_vcr_outcome``), the session-end summary must still render
the OVERFLOW + LIVE_CALL sections that prove the cost-leak signal
survived the xdist worker→controller hop."""
aggregate_report_outcome(
_worker_report(
"t::overflow_via_worker",
_outcome_from_worker(VERDICT_MISS_OVERFLOW),
)
)
aggregate_report_outcome(
_worker_report(
"t::live_call_via_worker",
_outcome_from_worker(
VERDICT_UNMARKED_LIVE_CALL,
skip_reason=SKIP_REASON_RESPX,
live_call_hosts=["api.openai.com"],
),
)
)
reporter = _FakeReporter()
emit_vcr_classification_summary(reporter)
assert "VCR CACHE CLASSIFICATION SUMMARY" in reporter.output
assert "CASSETTE OVERFLOW" in reporter.output
assert "t::overflow_via_worker" in reporter.output
assert "UNMARKED TESTS WITH LIVE API CALLS" in reporter.output
assert "api.openai.com" in reporter.output
assert "t::live_call_via_worker" in reporter.output
def test_record_vcr_outcome_emits_structured_payload_for_marked_tests(
vcr_enabled,
):
"""``record_vcr_outcome`` must always stash the structured outcome on
``user_properties`` (independent of verbose logging) so the controller
has something to aggregate from in xdist mode."""
request = SimpleNamespace(
node=SimpleNamespace(
nodeid="t::marked",
user_properties=[],
rep_call=SimpleNamespace(passed=True),
)
)
cassette = _cassette(played=1, dirty=False, total=1)
cassette._path = None
record_vcr_outcome(request, cassette)
outcomes = [v for k, v in request.node.user_properties if k == "vcr_outcome"]
recorded_by = [v for k, v in request.node.user_properties if k == "vcr_recorded_by"]
assert outcomes == [
{"verdict": VERDICT_HIT, "skip_reason": None, "live_call_hosts": []}
]
# No PYTEST_XDIST_WORKER set in the vcr_enabled fixture, so the
# recording-process tag is the empty string (single-process mode).
assert recorded_by == [""]
def test_record_vcr_outcome_emits_structured_payload_for_unmarked_live_call(
vcr_enabled,
):
"""The unmarked-LIVE_CALL path must ship the hosts list and the
skip-reason so the controller can rebuild both."""
request_node = SimpleNamespace(
nodeid="t::leak",
user_properties=[],
rep_call=SimpleNamespace(passed=True),
)
setattr(request_node, VCR_SKIP_REASON_USER_ATTR, SKIP_REASON_RESPX)
setattr(request_node, "vcr_live_call_hosts", ["api.openai.com"])
request = SimpleNamespace(node=request_node)
record_vcr_outcome(request, None)
outcomes = [v for k, v in request.node.user_properties if k == "vcr_outcome"]
assert outcomes == [
{
"verdict": VERDICT_UNMARKED_LIVE_CALL,
"skip_reason": SKIP_REASON_RESPX,
"live_call_hosts": ["api.openai.com"],
}
]
# ---------------------------------------------------------------------------
# Live-call probe
# ---------------------------------------------------------------------------
def test_should_skip_live_probe_when_vcr_active(vcr_enabled):
"""When the test *is* VCR-marked (cassette truthy), we don't install
the probe — vcrpy intercepts above the socket layer, so any
'connection' would be vcrpy's own bookkeeping and not real spend."""
request = SimpleNamespace(node=SimpleNamespace(), addfinalizer=lambda fn: None)
fake_cassette = SimpleNamespace(play_count=0, dirty=False)
probe = install_live_call_probe(request, fake_cassette)
assert probe is None
def test_live_call_probe_records_known_llm_hosts(vcr_enabled, monkeypatch):
"""The probe should record outbound TCP connections to known LLM
provider hosts (and ignore localhost / RFC1918 / unknown hosts)."""
finalizers = []
class _Node:
pass
request = SimpleNamespace(
node=_Node(), addfinalizer=lambda fn: finalizers.append(fn)
)
probe = install_live_call_probe(request, None)
assert probe is not None
import socket
# Manually invoke the patched function — we don't actually open a
# connection because that would hit the network. The probe records
# at the *call site* before delegating, and the original
# ``socket.create_connection`` will then fail; we swallow that.
try:
socket.create_connection(("api.openai.com", 443), timeout=0.001)
except Exception:
pass
try:
socket.create_connection(("127.0.0.1", 6379), timeout=0.001)
except Exception:
pass
# Restore via finalizers before asserting so the rest of the test
# session is unaffected.
for fn in finalizers:
fn()
hosts = getattr(request.node, "vcr_live_call_hosts", [])
assert "api.openai.com" in hosts
assert "127.0.0.1" not in hosts

1
tests/llm_translation/test_xai.py
View File

@ -11,7 +11,6 @@ sys.path.insert(
import httpx
import pytest
from respx import MockRouter
import litellm
from litellm import Choices, Message, ModelResponse, EmbeddingResponse, Usage

27
tests/local_testing/conftest.py
View File

@ -25,20 +25,21 @@ import litellm
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
)
# vcrpy and respx both patch the httpx transport — applying both makes one
# silently win, so respx-using files opt out of the auto-marker.
_RESPX_CONFLICTING_FILES = frozenset(
{
"test_router.py",
"test_amazing_vertex_completion.py",
"test_azure_openai.py",
}
)
# Per-item respx detection (``apply_vcr_auto_marker_to_items``) auto-skips
# tests whose ``@pytest.mark.respx`` marker or ``respx_mock`` fixture
# would conflict with vcrpy's transport patch. We no longer maintain a
# file-level ``_RESPX_CONFLICTING_FILES`` list here — the previous
# entries (``test_router.py``) had only a stale ``from respx import
# MockRouter`` import with no actual respx wiring, so file-level
# blacklisting was masking valid cache opportunities.
# Files where VCR replay breaks the test:
# - ``test_assistants.py``: polls fresh per-session run IDs that no cassette
@ -76,6 +77,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@ -88,6 +90,11 @@ def pytest_runtest_logreport(report):
_verbose_state.maybe_emit_verdict(report)
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)
# ---------------------------------------------------------------------------
# Capture TRUE defaults at conftest import time. This runs before any test
# module's top-level code (e.g. `litellm.num_retries = 3`) executes, so
@ -215,7 +222,7 @@ def setup_and_teardown():
def pytest_collection_modifyitems(config, items):
apply_vcr_auto_marker_to_items(
items,
skip_files=_RESPX_CONFLICTING_FILES | _VCR_INCOMPATIBLE_FILES,
skip_files=_VCR_INCOMPATIBLE_FILES,
skip_nodeid_suffixes=_VCR_INCOMPATIBLE_NODEID_SUFFIXES,
)

1
tests/local_testing/test_router.py
View File

@ -20,7 +20,6 @@ import os
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor
from unittest.mock import AsyncMock, MagicMock, patch
from respx import MockRouter
import httpx
from dotenv import load_dotenv
from pydantic import BaseModel

9
tests/logging_callback_tests/conftest.py
View File

@ -22,6 +22,9 @@ import litellm
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
@ -69,6 +72,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@ -220,3 +224,8 @@ def pytest_collection_modifyitems(config, items):
# Reorder the items list
items[:] = custom_logger_tests + other_tests
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)

9
tests/ocr_tests/conftest.py
View File

@ -15,6 +15,9 @@ sys.path.insert(0, os.path.abspath("../.."))
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
@ -41,6 +44,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@ -55,3 +59,8 @@ def pytest_runtest_logreport(report):
def pytest_collection_modifyitems(config, items):
apply_vcr_auto_marker_to_items(items)
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)

25
tests/pass_through_unit_tests/conftest.py
View File

@ -8,11 +8,26 @@ sys.path.insert(0, os.path.abspath("../.."))
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
)
# Tests that observe live cross-call provider state — typically a
# warm-up call followed by an assertion that the *second* call sees the
# upstream's prompt-cache (Anthropic / Bedrock prompt-caching). VCR's
# deterministic replay can't model this: both calls match the same
# cassette episode, so the second call returns the first call's
# pre-warmup response. Opt these out so they run live (no caching).
_VCR_INCOMPATIBLE_NODEID_SUFFIXES = (
"::test_prompt_caching_returns_cache_read_tokens_on_second_call",
"::test_prompt_caching_streaming_second_call_returns_cache_read",
)
_verbose_state = VerboseReporterState()
@ -34,6 +49,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@ -47,4 +63,11 @@ def pytest_runtest_logreport(report):
def pytest_collection_modifyitems(config, items):
apply_vcr_auto_marker_to_items(items)
apply_vcr_auto_marker_to_items(
items, skip_nodeid_suffixes=_VCR_INCOMPATIBLE_NODEID_SUFFIXES
)
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)

9
tests/router_unit_tests/conftest.py
View File

@ -15,6 +15,9 @@ import litellm # noqa: E402,F401
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
@ -87,6 +90,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@ -114,3 +118,8 @@ def pytest_collection_modifyitems(config, items):
# Reorder the items list
items[:] = custom_logger_tests + other_tests
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)

9
tests/search_tests/conftest.py
View File

@ -16,6 +16,9 @@ sys.path.insert(0, os.path.abspath("../.."))
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
@ -42,6 +45,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@ -56,3 +60,8 @@ def pytest_runtest_logreport(report):
def pytest_collection_modifyitems(config, items):
apply_vcr_auto_marker_to_items(items)
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)

484
tests/test_litellm/integrations/websearch_interception/test_websearch_native_blocks.py Normal file
View File

@ -0,0 +1,484 @@
"""
Tests for Anthropic-native ``web_search_tool_result`` block emission.
Covers the path that lets Claude Desktop / Anthropic SDK clients render
citations when their request used a native ``web_search_*`` tool against a
provider (e.g. Bedrock) that can't run web search natively.
"""
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from litellm.integrations.websearch_interception.handler import (
WEBSEARCH_EMIT_NATIVE_BLOCKS_KEY,
WEBSEARCH_NATIVE_BLOCKS_METADATA_KEY,
WebSearchInterceptionLogger,
)
from litellm.integrations.websearch_interception.tools import (
is_anthropic_native_web_search_tool,
is_web_search_tool,
)
from litellm.integrations.websearch_interception.transformation import (
WebSearchTransformation,
)
from litellm.llms.base_llm.search.transformation import SearchResponse, SearchResult
from litellm.types.integrations.custom_logger import (
AgenticLoopPlan,
AgenticLoopRequestPatch,
)
def _make_search_response() -> SearchResponse:
return SearchResponse(
results=[
SearchResult(
title="LiteLLM Docs",
url="https://docs.litellm.ai/",
snippet="Unified interface for LLMs.",
date="2025-01-15",
),
SearchResult(
title="Bedrock Pricing",
url="https://aws.amazon.com/bedrock/pricing/",
snippet="Pay-per-use pricing model.",
date=None,
),
]
)
class TestIsAnthropicNativeWebSearchTool:
"""The detector must match native tools without catching look-alikes."""
def test_matches_web_search_20250305(self):
assert is_anthropic_native_web_search_tool(
{"type": "web_search_20250305", "name": "web_search", "max_uses": 5}
)
def test_matches_future_dated_variant(self):
assert is_anthropic_native_web_search_tool(
{"type": "web_search_20260101", "name": "web_search"}
)
def test_rejects_litellm_standard(self):
assert not is_anthropic_native_web_search_tool(
{"name": "litellm_web_search", "input_schema": {}}
)
def test_rejects_openai_function_shape(self):
assert not is_anthropic_native_web_search_tool(
{"type": "function", "function": {"name": "litellm_web_search"}}
)
def test_rejects_claude_desktop_builtin(self):
# Claude Desktop's builtin client-side ``WebSearch`` tool must not be
# misidentified — that's the collision PR #25242 introduced.
assert not is_anthropic_native_web_search_tool({"name": "WebSearch"})
def test_rejects_unrelated_tool(self):
assert not is_anthropic_native_web_search_tool(
{"type": "function", "function": {"name": "calculator"}}
)
def test_handles_missing_type(self):
assert not is_anthropic_native_web_search_tool({"name": "web_search"})
class TestLegacyWebSearchNameGate:
"""The bare ``WebSearch`` name is a legacy interception marker. Real
client-side ``WebSearch`` tools (Cowork, Claude Desktop) carry an
``input_schema`` and must pass through untouched — otherwise the proxy
hijacks them server-side and the client's own tool handler never fires,
which means the separate ``web_search_20250305`` sub-request (where
citations actually flow) is never made."""
def test_bare_legacy_name_still_matched(self):
# Caller deliberately uses the bare-name interception marker —
# back-compat for anyone relying on the old shape.
assert is_web_search_tool({"name": "WebSearch"})
def test_real_client_tool_passes_through(self):
# Cowork's client-side WebSearch tool ships with input_schema.
cowork_tool = {
"name": "WebSearch",
"input_schema": {
"type": "object",
"properties": {"query": {"type": "string"}},
"required": ["query"],
},
}
assert not is_web_search_tool(cowork_tool)
def test_real_client_tool_with_description_passes_through(self):
# description-only client tools (no schema) are not valid Anthropic
# tools; only the schema-bearing shape is the disambiguator. This
# case stays matched on the assumption it's a legacy marker.
assert is_web_search_tool({"name": "WebSearch", "description": "search"})
class TestBuildWebSearchToolResultBlock:
"""The block-builder must produce the Anthropic-native shape exactly."""
def test_shape_with_results(self):
block = WebSearchTransformation.build_web_search_tool_result_block(
tool_use_id="toolu_abc",
search_response=_make_search_response(),
)
assert block["type"] == "web_search_tool_result"
assert block["tool_use_id"] == "toolu_abc"
assert len(block["content"]) == 2
first = block["content"][0]
assert first["type"] == "web_search_result"
assert first["url"] == "https://docs.litellm.ai/"
assert first["title"] == "LiteLLM Docs"
assert first["page_age"] == "2025-01-15"
assert first["encrypted_content"] == ""
def test_handles_none_search_response(self):
block = WebSearchTransformation.build_web_search_tool_result_block(
tool_use_id="toolu_abc",
search_response=None,
)
assert block["type"] == "web_search_tool_result"
assert block["tool_use_id"] == "toolu_abc"
assert block["content"] == []
def test_handles_empty_results(self):
block = WebSearchTransformation.build_web_search_tool_result_block(
tool_use_id="toolu_xyz",
search_response=SearchResponse(results=[]),
)
assert block["content"] == []
class TestPreRequestHookFlagsNativeTools:
"""The pre-request hook must mark the request when a native tool is used."""
@pytest.mark.asyncio
async def test_native_tool_sets_flag(self):
logger = WebSearchInterceptionLogger(enabled_providers=["bedrock"])
kwargs = {
"tools": [
{"type": "web_search_20250305", "name": "web_search", "max_uses": 5}
],
"litellm_params": {"custom_llm_provider": "bedrock"},
}
out = await logger.async_pre_request_hook(
model="bedrock/claude", messages=[], kwargs=kwargs
)
assert out is not None
assert out.get(WEBSEARCH_EMIT_NATIVE_BLOCKS_KEY) is True
@pytest.mark.asyncio
async def test_litellm_standard_tool_does_not_set_flag(self):
logger = WebSearchInterceptionLogger(enabled_providers=["bedrock"])
kwargs = {
"tools": [{"name": "litellm_web_search", "input_schema": {}}],
"litellm_params": {"custom_llm_provider": "bedrock"},
}
out = await logger.async_pre_request_hook(
model="bedrock/claude", messages=[], kwargs=kwargs
)
assert out is not None
assert WEBSEARCH_EMIT_NATIVE_BLOCKS_KEY not in out
class TestBuildPlanAttachesBlocks:
"""async_build_agentic_loop_plan must put pre-built blocks on metadata."""
@pytest.mark.asyncio
async def test_metadata_carries_blocks_when_flag_set(self):
logger = WebSearchInterceptionLogger(enabled_providers=["bedrock"])
tool_calls = [
{
"id": "toolu_one",
"type": "tool_use",
"name": "litellm_web_search",
"input": {"query": "what is litellm"},
}
]
patch_obj = AgenticLoopRequestPatch(
model="bedrock/claude",
messages=[{"role": "user", "content": "hi"}],
max_tokens=1024,
)
structured = [_make_search_response()]
with patch.object(
logger,
"_build_anthropic_request_patch",
new=AsyncMock(return_value=(patch_obj, structured)),
):
plan = await logger.async_build_agentic_loop_plan(
tools={"tool_calls": tool_calls, "thinking_blocks": []},
model="bedrock/claude",
messages=[],
response=MagicMock(),
anthropic_messages_provider_config=None,
anthropic_messages_optional_request_params={},
logging_obj=MagicMock(model_call_details={}),
stream=False,
kwargs={WEBSEARCH_EMIT_NATIVE_BLOCKS_KEY: True},
)
blocks = plan.metadata.get(WEBSEARCH_NATIVE_BLOCKS_METADATA_KEY)
assert isinstance(blocks, list)
assert len(blocks) == 1
assert blocks[0]["type"] == "web_search_tool_result"
assert blocks[0]["tool_use_id"] == "toolu_one"
assert blocks[0]["content"][0]["url"] == "https://docs.litellm.ai/"
@pytest.mark.asyncio
async def test_metadata_does_not_carry_blocks_when_flag_absent(self):
logger = WebSearchInterceptionLogger(enabled_providers=["bedrock"])
tool_calls = [
{
"id": "toolu_one",
"type": "tool_use",
"name": "litellm_web_search",
"input": {"query": "what is litellm"},
}
]
patch_obj = AgenticLoopRequestPatch(
model="bedrock/claude",
messages=[{"role": "user", "content": "hi"}],
max_tokens=1024,
)
with patch.object(
logger,
"_build_anthropic_request_patch",
new=AsyncMock(return_value=(patch_obj, [_make_search_response()])),
):
plan = await logger.async_build_agentic_loop_plan(
tools={"tool_calls": tool_calls, "thinking_blocks": []},
model="bedrock/claude",
messages=[],
response=MagicMock(),
anthropic_messages_provider_config=None,
anthropic_messages_optional_request_params={},
logging_obj=MagicMock(model_call_details={}),
stream=False,
kwargs={},
)
assert WEBSEARCH_NATIVE_BLOCKS_METADATA_KEY not in plan.metadata
class TestPostHookInjectsBlocks:
"""The post-hook must prepend blocks; absent metadata is a no-op."""
@pytest.mark.asyncio
async def test_injects_when_metadata_present(self):
logger = WebSearchInterceptionLogger(enabled_providers=["bedrock"])
block = WebSearchTransformation.build_web_search_tool_result_block(
tool_use_id="toolu_abc",
search_response=_make_search_response(),
)
plan = AgenticLoopPlan(
run_agentic_loop=True,
metadata={WEBSEARCH_NATIVE_BLOCKS_METADATA_KEY: [block]},
)
response = {
"id": "msg_1",
"type": "message",
"role": "assistant",
"content": [{"type": "text", "text": "Based on the search..."}],
"stop_reason": "end_turn",
}
out = await logger.async_post_agentic_loop_response_hook(
response=response, plan=plan, kwargs={}
)
# Native block must be first so the client can pair it with the
# tool_use before reading the assistant text.
assert out["content"][0]["type"] == "web_search_tool_result"
assert out["content"][0]["tool_use_id"] == "toolu_abc"
assert out["content"][1]["type"] == "text"
@pytest.mark.asyncio
async def test_noop_when_metadata_absent(self):
logger = WebSearchInterceptionLogger(enabled_providers=["bedrock"])
plan = AgenticLoopPlan(run_agentic_loop=True, metadata={})
response = {
"id": "msg_1",
"content": [{"type": "text", "text": "answer"}],
}
out = await logger.async_post_agentic_loop_response_hook(
response=response, plan=plan, kwargs={}
)
assert out == response
@pytest.mark.asyncio
async def test_handles_object_style_response(self):
logger = WebSearchInterceptionLogger(enabled_providers=["bedrock"])
block = WebSearchTransformation.build_web_search_tool_result_block(
tool_use_id="toolu_obj",
search_response=_make_search_response(),
)
plan = AgenticLoopPlan(
run_agentic_loop=True,
metadata={WEBSEARCH_NATIVE_BLOCKS_METADATA_KEY: [block]},
)
class _Resp:
def __init__(self):
self.content = [{"type": "text", "text": "ok"}]
resp = _Resp()
out = await logger.async_post_agentic_loop_response_hook(
response=resp, plan=plan, kwargs={}
)
assert out.content[0]["type"] == "web_search_tool_result"
assert out.content[1]["type"] == "text"
class TestShortCircuitEmitsNativeBlocks:
"""Standalone /v1/messages sub-requests (Cowork's separate search call)
hit ``try_short_circuit_search``, which builds a synthetic response and
never enters the agentic loop. The native-block emission must happen
here too, otherwise the citations panel stays empty."""
@pytest.mark.asyncio
async def test_native_tool_short_circuit_emits_blocks(self):
logger = WebSearchInterceptionLogger(enabled_providers=["github_copilot"])
with patch.object(
logger,
"_execute_search",
new=AsyncMock(return_value=("Title: x\nURL: y", _make_search_response())),
):
result = await logger.try_short_circuit_search(
model="github_copilot/claude-sonnet-4",
messages=[{"role": "user", "content": "search query"}],
tools=[
{
"type": "web_search_20250305",
"name": "web_search",
"max_uses": 3,
}
],
custom_llm_provider="github_copilot",
)
assert result is not None
block_types = [b["type"] for b in result["content"]]
# Order matters: native clients expect tool_use before tool_result.
assert block_types == ["server_tool_use", "web_search_tool_result", "text"]
server_use, tool_result, _ = result["content"]
assert server_use["name"] == "web_search"
assert server_use["input"] == {"query": "search query"}
# tool_use_id must match between the server_tool_use and the
# web_search_tool_result block so the client can pair them.
assert server_use["id"].startswith("srvtoolu_")
assert tool_result["tool_use_id"] == server_use["id"]
# The actual search results carry through (urls + titles).
assert len(tool_result["content"]) == 2
assert tool_result["content"][0]["url"] == "https://docs.litellm.ai/"
@pytest.mark.asyncio
async def test_litellm_standard_tool_short_circuit_stays_text_only(self):
"""Non-native tool → existing text-only short-circuit, no regression."""
logger = WebSearchInterceptionLogger(enabled_providers=["github_copilot"])
with patch.object(
logger,
"_execute_search",
new=AsyncMock(return_value=("Title: x\nURL: y", _make_search_response())),
):
result = await logger.try_short_circuit_search(
model="github_copilot/claude-sonnet-4",
messages=[{"role": "user", "content": "search query"}],
tools=[
{
"name": "litellm_web_search",
"input_schema": {
"type": "object",
"properties": {"query": {"type": "string"}},
"required": ["query"],
},
}
],
custom_llm_provider="github_copilot",
)
assert result is not None
block_types = [b["type"] for b in result["content"]]
assert block_types == ["text"]
@pytest.mark.asyncio
async def test_native_short_circuit_failure_still_emits_blocks(self):
"""Search failure on native path: emit blocks with empty results +
the legacy text-error block, so the client gets a well-formed
response instead of a malformed half-shape."""
logger = WebSearchInterceptionLogger(enabled_providers=["github_copilot"])
with patch.object(logger, "_execute_search", side_effect=RuntimeError("boom")):
result = await logger.try_short_circuit_search(
model="github_copilot/claude-sonnet-4",
messages=[{"role": "user", "content": "search query"}],
tools=[{"type": "web_search_20250305", "name": "web_search"}],
custom_llm_provider="github_copilot",
)
assert result is not None
block_types = [b["type"] for b in result["content"]]
assert block_types == ["server_tool_use", "web_search_tool_result", "text"]
tool_result = result["content"][1]
assert tool_result["content"] == []
text_block = result["content"][2]
assert "Search failed" in text_block["text"]
class TestLegacyPathMatchesNewPath:
"""The legacy ``_execute_agentic_loop`` must inject blocks too."""
@pytest.mark.asyncio
async def test_legacy_path_injects_when_flag_set(self):
logger = WebSearchInterceptionLogger(enabled_providers=["bedrock"])
tool_calls = [
{
"id": "toolu_legacy",
"type": "tool_use",
"name": "litellm_web_search",
"input": {"query": "q"},
}
]
patch_obj = AgenticLoopRequestPatch(
model="bedrock/claude",
messages=[{"role": "user", "content": "hi"}],
max_tokens=1024,
optional_params={},
)
followup_response = {
"id": "msg_followup",
"content": [{"type": "text", "text": "final answer"}],
}
with (
patch.object(
logger,
"_build_anthropic_request_patch",
new=AsyncMock(return_value=(patch_obj, [_make_search_response()])),
),
patch(
"litellm.integrations.websearch_interception.handler.anthropic_messages.acreate",
new=AsyncMock(return_value=followup_response),
),
):
out = await logger._execute_agentic_loop(
model="bedrock/claude",
messages=[],
tool_calls=tool_calls,
thinking_blocks=[],
anthropic_messages_optional_request_params={},
logging_obj=MagicMock(model_call_details={}),
stream=False,
kwargs={WEBSEARCH_EMIT_NATIVE_BLOCKS_KEY: True},
)
assert out["content"][0]["type"] == "web_search_tool_result"
assert out["content"][0]["tool_use_id"] == "toolu_legacy"
assert out["content"][1]["type"] == "text"

34
tests/test_litellm/integrations/websearch_interception/test_websearch_short_circuit.py
View File

@ -30,7 +30,8 @@ class TestTryShortCircuitSearch:
logger, "_execute_search", new_callable=AsyncMock
) as mock_search:
mock_search.return_value = (
"Title: Result\nURL: https://example.com\nSnippet: test"
"Title: Result\nURL: https://example.com\nSnippet: test",
None,
)
result = await logger.try_short_circuit_search(
@ -48,9 +49,15 @@ class TestTryShortCircuitSearch:
assert result["type"] == "message"
assert result["role"] == "assistant"
assert result["stop_reason"] == "end_turn"
assert len(result["content"]) == 1
assert result["content"][0]["type"] == "text"
assert "Result" in result["content"][0]["text"]
# Native web_search_20250305 client → short-circuit emits native
# blocks (server_tool_use + web_search_tool_result) plus the legacy
# text block so Cowork / Claude Desktop citations panels populate.
block_types = [b["type"] for b in result["content"]]
assert "server_tool_use" in block_types
assert "web_search_tool_result" in block_types
assert "text" in block_types
text_block = next(b for b in result["content"] if b["type"] == "text")
assert "Result" in text_block["text"]
mock_search.assert_called_once_with("Search for Claude Code releases")
@pytest.mark.asyncio
@ -173,7 +180,8 @@ class TestTryShortCircuitSearch:
)
assert result is not None
assert "Search failed" in result["content"][0]["text"]
text_block = next(b for b in result["content"] if b["type"] == "text")
assert "Search failed" in text_block["text"]
@pytest.mark.asyncio
async def test_response_has_valid_structure(self):
@ -183,7 +191,7 @@ class TestTryShortCircuitSearch:
with patch.object(
logger, "_execute_search", new_callable=AsyncMock
) as mock_search:
mock_search.return_value = "search results here"
mock_search.return_value = ("search results here", None)
result = await logger.try_short_circuit_search(
model="github_copilot/claude-sonnet-4",
@ -246,7 +254,7 @@ class TestShortCircuitEntryPoint:
with patch.object(
logger, "_execute_search", new_callable=AsyncMock
) as mock_search:
mock_search.return_value = "results"
mock_search.return_value = ("results", None)
with patch("litellm.callbacks", [logger]):
result = await _try_websearch_short_circuit(
model="github_copilot/claude-sonnet-4",
@ -257,7 +265,8 @@ class TestShortCircuitEntryPoint:
)
assert isinstance(result, dict)
assert result["content"][0]["text"] == "results"
text_block = next(b for b in result["content"] if b["type"] == "text")
assert text_block["text"] == "results"
@pytest.mark.asyncio
async def test_returns_stream_iterator_when_streaming(self):
@ -273,7 +282,7 @@ class TestShortCircuitEntryPoint:
with patch.object(
logger, "_execute_search", new_callable=AsyncMock
) as mock_search:
mock_search.return_value = "streaming results"
mock_search.return_value = ("streaming results", None)
with patch("litellm.callbacks", [logger]):
result = await _try_websearch_short_circuit(
model="github_copilot/claude-sonnet-4",
@ -338,7 +347,7 @@ class TestShortCircuitEntryPoint:
with patch.object(
logger, "_execute_search", new_callable=AsyncMock
) as mock_search:
mock_search.return_value = "streaming results"
mock_search.return_value = ("streaming results", None)
with patch("litellm.callbacks", [logger]):
# Simulate what anthropic_messages() does: original_stream=True
# is passed to the short-circuit, even though the hook would have
@ -368,7 +377,7 @@ class TestShortCircuitEntryPoint:
with patch.object(
logger, "_execute_search", new_callable=AsyncMock
) as mock_search:
mock_search.return_value = "results"
mock_search.return_value = ("results", None)
with patch("litellm.callbacks", [logger]):
# Simulate the caller having derived custom_llm_provider from
# the model string before calling _try_websearch_short_circuit
@ -381,4 +390,5 @@ class TestShortCircuitEntryPoint:
)
assert result is not None
assert result["content"][0]["text"] == "results"
text_block = next(b for b in result["content"] if b["type"] == "text")
assert text_block["text"] == "results"

36
tests/test_litellm/integrations/websearch_interception/test_websearch_thinking_constraint.py
View File

@ -68,7 +68,9 @@ class TestThinkingBudgetTokensConstraint:
"litellm.integrations.websearch_interception.handler.anthropic_messages.acreate",
side_effect=_fake_acreate,
),
patch.object(logger, "_execute_search", return_value="search result"),
patch.object(
logger, "_execute_search", return_value=("search result", None)
),
):
await logger._execute_agentic_loop(
@ -102,7 +104,9 @@ class TestThinkingBudgetTokensConstraint:
"litellm.integrations.websearch_interception.handler.anthropic_messages.acreate",
side_effect=_fake_acreate,
),
patch.object(logger, "_execute_search", return_value="search result"),
patch.object(
logger, "_execute_search", return_value=("search result", None)
),
):
await logger._execute_agentic_loop(
@ -136,7 +140,9 @@ class TestThinkingBudgetTokensConstraint:
"litellm.integrations.websearch_interception.handler.anthropic_messages.acreate",
side_effect=_fake_acreate,
),
patch.object(logger, "_execute_search", return_value="search result"),
patch.object(
logger, "_execute_search", return_value=("search result", None)
),
):
await logger._execute_agentic_loop(
@ -170,7 +176,9 @@ class TestThinkingBudgetTokensConstraint:
"litellm.integrations.websearch_interception.handler.anthropic_messages.acreate",
side_effect=_fake_acreate,
),
patch.object(logger, "_execute_search", return_value="search result"),
patch.object(
logger, "_execute_search", return_value=("search result", None)
),
):
await logger._execute_agentic_loop(
@ -201,7 +209,9 @@ class TestThinkingBudgetTokensConstraint:
"litellm.integrations.websearch_interception.handler.anthropic_messages.acreate",
side_effect=_fake_acreate,
),
patch.object(logger, "_execute_search", return_value="search result"),
patch.object(
logger, "_execute_search", return_value=("search result", None)
),
):
await logger._execute_agentic_loop(
@ -286,7 +296,9 @@ class TestLoggingObjExcludedFromFollowUp:
"litellm.integrations.websearch_interception.handler.anthropic_messages.acreate",
side_effect=_fake_acreate,
),
patch.object(logger, "_execute_search", return_value="search result"),
patch.object(
logger, "_execute_search", return_value=("search result", None)
),
):
await logger._execute_agentic_loop(
@ -325,7 +337,9 @@ class TestLoggingObjExcludedFromFollowUp:
"litellm.integrations.websearch_interception.handler.anthropic_messages.acreate",
side_effect=_fake_acreate,
),
patch.object(logger, "_execute_search", return_value="search result"),
patch.object(
logger, "_execute_search", return_value=("search result", None)
),
):
await logger._execute_agentic_loop(
@ -373,7 +387,9 @@ class TestFollowUpErrorScenarios:
"litellm.integrations.websearch_interception.handler.anthropic_messages.acreate",
side_effect=_fail_acreate,
),
patch.object(logger, "_execute_search", return_value="search result"),
patch.object(
logger, "_execute_search", return_value=("search result", None)
),
):
with pytest.raises(Exception, match="max_tokens must be greater"):
@ -450,7 +466,9 @@ class TestFollowUpErrorScenarios:
"litellm.integrations.websearch_interception.handler.anthropic_messages.acreate",
side_effect=_fake_acreate,
),
patch.object(logger, "_execute_search", return_value="search result"),
patch.object(
logger, "_execute_search", return_value=("search result", None)
),
):
await logger._execute_agentic_loop(

9
tests/unified_google_tests/conftest.py
View File

@ -15,6 +15,9 @@ import litellm # noqa: E402,F401
from tests._vcr_conftest_common import ( # noqa: E402
VerboseReporterState,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
vcr_config_dict,
@ -74,6 +77,7 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@ -101,3 +105,8 @@ def pytest_collection_modifyitems(config, items):
# Reorder the items list
items[:] = custom_logger_tests + other_tests
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)