Merge branch 'BerriAI:main' into xecguard-integration

This commit is contained in:
clyang 2026-04-07 16:00:21 +08:00 committed by GitHub
commit a5cf4f55e8
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
58 changed files with 3417 additions and 5473 deletions

View file

@ -2868,114 +2868,6 @@ jobs:
- store_test_results:
path: test-results
proxy_e2e_azure_batches_tests:
machine:
image: ubuntu-2204:2023.10.1
resource_class: large
working_directory: ~/project
steps:
- checkout
- setup_google_dns
- run:
name: Install Docker CLI
command: |
curl -fsSL https://get.docker.com | sh
sudo usermod -aG docker $USER
docker version
- run:
name: Install Python 3.12
command: |
curl https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh --output miniconda.sh
bash miniconda.sh -b -p $HOME/miniconda
export PATH="$HOME/miniconda/bin:$PATH"
conda init bash
source ~/.bashrc
conda create -n myenv python=3.12 -y
conda activate myenv
python --version
- run:
name: Install Poetry
command: |
export PATH="$HOME/miniconda/bin:$PATH"
source $HOME/miniconda/etc/profile.d/conda.sh
conda activate myenv
pip install poetry
- run:
name: Install dockerize
command: |
wget https://github.com/jwilder/dockerize/releases/download/v0.6.1/dockerize-linux-amd64-v0.6.1.tar.gz
sudo tar -C /usr/local/bin -xzvf dockerize-linux-amd64-v0.6.1.tar.gz
rm dockerize-linux-amd64-v0.6.1.tar.gz
- run:
name: Start PostgreSQL Database
command: |
docker run -d \
--name postgres-db \
-e POSTGRES_USER=llmproxy \
-e POSTGRES_PASSWORD=dbpassword9090 \
-e POSTGRES_DB=litellm \
-p 5432:5432 \
postgres:15
- run:
name: Wait for PostgreSQL to be ready
command: dockerize -wait tcp://localhost:5432 -timeout 1m
- run:
name: Install system dependencies
command: |
sudo apt-get update -y
sudo apt-get install -y libpq-dev
- run:
name: Install Dependencies
command: |
export PATH="$HOME/miniconda/bin:$PATH"
source $HOME/miniconda/etc/profile.d/conda.sh
conda activate myenv
poetry config virtualenvs.in-project true
poetry install --with dev,proxy-dev --extras "proxy"
poetry run pip install psycopg2-binary uvicorn fastapi httpx tenacity
- run:
name: Setup litellm-enterprise
command: |
export PATH="$HOME/miniconda/bin:$PATH"
source $HOME/miniconda/etc/profile.d/conda.sh
conda activate myenv
poetry run pip install --force-reinstall --no-deps -e enterprise/
- run:
name: Generate Prisma client
command: |
export PATH="$HOME/miniconda/bin:$PATH"
source $HOME/miniconda/etc/profile.d/conda.sh
conda activate myenv
poetry run prisma generate --schema litellm/proxy/schema.prisma
- run:
name: Run Prisma migrations
command: |
export PATH="$HOME/miniconda/bin:$PATH"
source $HOME/miniconda/etc/profile.d/conda.sh
conda activate myenv
export DATABASE_URL=postgresql://llmproxy:dbpassword9090@localhost:5432/litellm
cd litellm/proxy
poetry run prisma migrate deploy --schema schema.prisma
cd ../..
- run:
name: Run Azure Batch E2E Tests
command: |
export PATH="$HOME/miniconda/bin:$PATH"
source $HOME/miniconda/etc/profile.d/conda.sh
conda activate myenv
export DATABASE_URL=postgresql://llmproxy:dbpassword9090@localhost:5432/litellm
export USE_LOCAL_LITELLM=true
export USE_MOCK_MODELS=true
export USE_STATE_TRACKER=true
export LITELLM_LOG=DEBUG
poetry run pytest tests/proxy_e2e_azure_batches_tests/test_proxy_e2e_azure_batches.py \
-vv -s -k "test_e2e_managed_batch" \
--tb=short \
--maxfail=3 \
--durations=10 \
--junitxml=test-results/junit.xml
no_output_timeout: 15m
upload-coverage:
docker:
- image: cimg/python:3.9
@ -3605,12 +3497,6 @@ workflows:
only:
- main
- /litellm_.*/
- proxy_e2e_azure_batches_tests:
filters:
branches:
only:
- main
- /litellm_.*/
- llm_translation_testing:
filters:
branches:

View file

@ -27,6 +27,10 @@ on:
required: false
type: number
default: 10
artifact-name:
description: "Unique name for the coverage artifact (must be unique per run)"
required: true
type: string
permissions:
contents: read
@ -93,4 +97,46 @@ jobs:
--reruns "${RERUNS}" \
--reruns-delay 1 \
--dist=loadscope \
--durations=20
--durations=20 \
--cov=litellm \
--cov-report=xml:coverage.xml \
--cov-config=pyproject.toml
- name: Save coverage report
if: always()
uses: actions/upload-artifact@4cec3d8aa04e39d1a68397de0c4cd6fb9dce8ec1 # v4.6.1
with:
name: coverage-${{ inputs.artifact-name }}-${{ github.run_id }}-${{ github.run_attempt }}
path: coverage.xml
retention-days: 1
upload-coverage:
name: Upload coverage to Codecov
needs: run
if: always()
runs-on: ubuntu-latest
permissions:
contents: read
id-token: write
pull-requests: write
steps:
- name: Checkout code
uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
with:
persist-credentials: false
- name: Download coverage report
uses: actions/download-artifact@95815c38cf2ff2164869cbab79da8d1f422bc89e # v4.2.1
with:
pattern: coverage-${{ inputs.artifact-name }}-${{ github.run_id }}-${{ github.run_attempt }}
path: coverage-reports
merge-multiple: true
- name: Upload to Codecov
uses: codecov/codecov-action@75cd11691c0faa626561e295848008c8a7dddffe # v5.5.4
with:
use_oidc: true
directory: coverage-reports
root_dir: ${{ github.workspace }}
fail_ci_if_error: false

View file

@ -37,6 +37,11 @@ on:
required: false
type: boolean
default: false
artifact-name:
description: "Unique name for the coverage artifact (must be unique per run)"
required: false
type: string
default: "run"
secrets:
REDIS_HOST:
required: false
@ -151,7 +156,10 @@ jobs:
--maxfail="${MAX_FAILURES}" \
--reruns "${RERUNS}" \
--reruns-delay 1 \
--durations=20
--durations=20 \
--cov=litellm \
--cov-report=xml:coverage.xml \
--cov-config=pyproject.toml
else
poetry run pytest ${TEST_PATH:?} \
--tb=short -vv \
@ -160,5 +168,47 @@ jobs:
--reruns "${RERUNS}" \
--reruns-delay 1 \
--dist=loadscope \
--durations=20
--durations=20 \
--cov=litellm \
--cov-report=xml:coverage.xml \
--cov-config=pyproject.toml
fi
- name: Save coverage report
if: always()
uses: actions/upload-artifact@4cec3d8aa04e39d1a68397de0c4cd6fb9dce8ec1 # v4.6.1
with:
name: coverage-${{ inputs.artifact-name }}-${{ github.run_id }}-${{ github.run_attempt }}
path: coverage.xml
retention-days: 1
upload-coverage:
name: Upload coverage to Codecov
needs: run
if: always()
runs-on: ubuntu-latest
permissions:
contents: read
id-token: write
pull-requests: write
steps:
- name: Checkout code
uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
with:
persist-credentials: false
- name: Download coverage report
uses: actions/download-artifact@95815c38cf2ff2164869cbab79da8d1f422bc89e # v4.2.1
with:
pattern: coverage-${{ inputs.artifact-name }}-${{ github.run_id }}-${{ github.run_attempt }}
path: coverage-reports
merge-multiple: true
- name: Upload to Codecov
uses: codecov/codecov-action@75cd11691c0faa626561e295848008c8a7dddffe # v5.5.4
with:
use_oidc: true
directory: coverage-reports
root_dir: ${{ github.workspace }}
fail_ci_if_error: false

0
.github/workflows/run_llm_translation_tests.py vendored Executable file → Normal file
View file

View file

@ -1,214 +0,0 @@
name: LiteLLM Unit Tests (Matrix)
on:
pull_request:
branches: [main]
permissions:
contents: read
# Cancel in-progress runs for the same PR
concurrency:
group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
cancel-in-progress: true
jobs:
test:
runs-on: ubuntu-latest
timeout-minutes: 20 # Increased from 15 to 20
strategy:
fail-fast: false
matrix:
test-group:
# tests/test_litellm split by subdirectory (~560 files total)
# Vertex AI tests separated for better isolation (prevent auth/env pollution)
- name: "llms-vertex"
path: "tests/test_litellm/llms/vertex_ai"
workers: 1
reruns: 2
- name: "llms-other"
path: "tests/test_litellm/llms --ignore=tests/test_litellm/llms/vertex_ai"
workers: 2
reruns: 2
# tests/test_litellm/proxy split by subdirectory (~180 files total)
- name: "proxy-guardrails"
path: "tests/test_litellm/proxy/guardrails tests/test_litellm/proxy/management_endpoints tests/test_litellm/proxy/management_helpers"
workers: 2
reruns: 2
- name: "proxy-core"
path: "tests/test_litellm/proxy/auth tests/test_litellm/proxy/client tests/test_litellm/proxy/db tests/test_litellm/proxy/hooks tests/test_litellm/proxy/policy_engine"
workers: 2
reruns: 2
- name: "proxy-misc"
path: "tests/test_litellm/proxy/_experimental tests/test_litellm/proxy/agent_endpoints tests/test_litellm/proxy/anthropic_endpoints tests/test_litellm/proxy/common_utils tests/test_litellm/proxy/discovery_endpoints tests/test_litellm/proxy/experimental tests/test_litellm/proxy/google_endpoints tests/test_litellm/proxy/health_endpoints tests/test_litellm/proxy/image_endpoints tests/test_litellm/proxy/middleware tests/test_litellm/proxy/openai_files_endpoint tests/test_litellm/proxy/pass_through_endpoints tests/test_litellm/proxy/prompts tests/test_litellm/proxy/public_endpoints tests/test_litellm/proxy/response_api_endpoints tests/test_litellm/proxy/spend_tracking tests/test_litellm/proxy/ui_crud_endpoints tests/test_litellm/proxy/vector_store_endpoints tests/test_litellm/proxy/test_*.py"
workers: 2
reruns: 2
- name: "integrations"
path: "tests/test_litellm/integrations"
workers: 2
reruns: 3 # Integration tests tend to be flakier
- name: "core-utils"
path: "tests/test_litellm/litellm_core_utils"
workers: 2
reruns: 1
- name: "other-1"
# responses (5942) + caching (1723) + types (819) ≈ 8.5k lines
path: "tests/test_litellm/responses tests/test_litellm/caching tests/test_litellm/types"
workers: 2
reruns: 2
- name: "other-2"
# enterprise (3062) + google_genai (2511) + router_utils (1982) ≈ 7.6k lines
path: "tests/test_litellm/enterprise tests/test_litellm/google_genai tests/test_litellm/router_utils"
workers: 2
reruns: 2
- name: "other-3"
# remaining dirs ≈ 8.0k lines
path: "tests/test_litellm/router_strategy tests/test_litellm/secret_managers tests/test_litellm/a2a_protocol tests/test_litellm/anthropic_interface tests/test_litellm/completion_extras tests/test_litellm/containers tests/test_litellm/experimental_mcp_client tests/test_litellm/images tests/test_litellm/interactions tests/test_litellm/passthrough tests/test_litellm/vector_stores"
workers: 2
reruns: 2
- name: "root"
path: "tests/test_litellm/test_*.py"
workers: 2
reruns: 2
# tests/proxy_unit_tests split alphabetically (~48 files total)
- name: "proxy-unit-a1"
# test_[a-j]*.py: jwt (1564) + auth_checks (978) + google_gemini (478) + e2e_pod_lock (437) + rest
path: "tests/proxy_unit_tests/test_[a-j]*.py"
workers: 2
reruns: 1
- name: "proxy-unit-a2"
# test_[k-o]*.py: key_generate_prisma (4346) + key_generate_dynamodb + models_fallback
path: "tests/proxy_unit_tests/test_[k-o]*.py"
workers: 2
reruns: 1
- name: "proxy-unit-b1"
# lighter config/utility proxy tests (prisma, project, prompt, proxy_[c-r]*)
path: "tests/proxy_unit_tests/test_prisma*.py tests/proxy_unit_tests/test_project*.py tests/proxy_unit_tests/test_prompt*.py tests/proxy_unit_tests/test_proxy_[c-r]*.py"
workers: 2
reruns: 1
- name: "proxy-unit-b2"
# proxy_server.py alone (2750 lines) - isolated to avoid blocking smaller tests
path: "tests/proxy_unit_tests/test_proxy_server.py"
workers: 2
reruns: 1
- name: "proxy-unit-b3"
# proxy_server_* (618) + proxy_setting_guardrails (71) - smaller server-related tests
path: "tests/proxy_unit_tests/test_proxy_server_*.py tests/proxy_unit_tests/test_proxy_setting_guardrails.py"
workers: 2
reruns: 1
- name: "proxy-unit-b4"
# proxy_utils.py alone (2339 lines) - isolated to avoid blocking token counter
path: "tests/proxy_unit_tests/test_proxy_utils.py"
workers: 2
reruns: 1
- name: "proxy-unit-b5"
# proxy_token_counter (1279) - runs independently from utils
path: "tests/proxy_unit_tests/test_proxy_token_counter.py"
workers: 2
reruns: 1
- name: "proxy-unit-b6"
# test_[r-t]*.py: response_polling (1399) + search_api_logging (202) + server_root (64) + skills_db (261) + realtime_cache (62)
path: "tests/proxy_unit_tests/test_[r-t]*.py"
workers: 2
reruns: 1
- name: "proxy-unit-b7"
# test_[u-z]*.py: user_api_key_auth (1136) + zero_cost (590) + update_spend (305) + unit_test_* (206) + ui_path (157)
path: "tests/proxy_unit_tests/test_[u-z]*.py"
workers: 2
reruns: 1
name: test (${{ matrix.test-group.name }})
steps:
- uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
with:
persist-credentials: false
- name: Set up Python
uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
with:
python-version: "3.12"
- name: Install Poetry
run: pip install 'poetry==2.3.2'
- name: Cache Poetry dependencies
uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4.0.0
with:
path: |
~/.cache/pypoetry
~/.cache/pip
.venv
key: ${{ runner.os }}-poetry-${{ hashFiles('poetry.lock') }}
restore-keys: |
${{ runner.os }}-poetry-
- name: Install dependencies
run: |
poetry config virtualenvs.in-project true
poetry install --with dev,proxy-dev --extras "proxy semantic-router"
# pytest-rerunfailures and pytest-xdist are in pyproject.toml dev dependencies
poetry run pip install google-genai==1.22.0 \
google-cloud-aiplatform==1.115.0 fastapi-offline==1.7.3 python-multipart==0.0.22 openapi-core==0.23.0
- name: Setup litellm-enterprise
run: |
poetry run pip install --force-reinstall --no-deps -e enterprise/
- name: Generate Prisma client
env:
PRISMA_BINARY_CACHE_DIR: ${{ runner.temp }}/prisma-cache
run: |
poetry run pip install nodejs-wheel-binaries==24.13.1
poetry run prisma generate --schema litellm/proxy/schema.prisma
- name: Run tests - ${{ matrix.test-group.name }}
run: |
poetry run pytest ${{ matrix.test-group.path }} \
--tb=short -vv \
--maxfail=10 \
-n ${{ matrix.test-group.workers }} \
--reruns ${{ matrix.test-group.reruns }} \
--reruns-delay 1 \
--dist=loadscope \
--durations=20 \
--cov=litellm \
--cov-report=xml:coverage-${{ matrix.test-group.name }}.xml \
--cov-config=pyproject.toml
- name: Save coverage report
if: always()
uses: actions/upload-artifact@4cec3d8aa04e39d1a68397de0c4cd6fb9dce8ec1 # v4.6.1
with:
name: coverage-${{ matrix.test-group.name }}
path: coverage-${{ matrix.test-group.name }}.xml
retention-days: 1
upload-coverage:
name: Upload coverage to Codecov
needs: test
if: always()
runs-on: ubuntu-latest
permissions:
contents: read
id-token: write # Required for OIDC tokenless upload
pull-requests: write # Required for Codecov PR comments
steps:
- name: Checkout code
uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
- name: Download all coverage reports
uses: actions/download-artifact@95815c38cf2ff2164869cbab79da8d1f422bc89e # v4.2.1
with:
pattern: coverage-*
path: coverage-reports
merge-multiple: true
- name: Upload to Codecov
uses: codecov/codecov-action@aa56896cf108bd10b5eb883cd1d24196da57f695 # v5.5.4
with:
use_oidc: true
directory: coverage-reports
root_dir: ${{ github.workspace }}
fail_ci_if_error: false

View file

@ -1,97 +0,0 @@
name: Proxy E2E Azure Batches Tests
on:
pull_request:
branches: [main]
workflow_dispatch:
concurrency:
group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
cancel-in-progress: true
permissions:
contents: read
jobs:
proxy_e2e_azure_batches_tests:
runs-on: ubuntu-latest
timeout-minutes: 30
services:
postgres:
image: postgres:15
env:
POSTGRES_USER: llmproxy
POSTGRES_PASSWORD: dbpassword9090
POSTGRES_DB: litellm
ports:
- 5432:5432
options: >-
--health-cmd pg_isready
--health-interval 10s
--health-timeout 5s
--health-retries 5
steps:
- uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
with:
persist-credentials: false
- name: Set up Python
uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
with:
python-version: "3.12"
- name: Install Poetry
run: pip install 'poetry==2.3.2'
- name: Cache Poetry dependencies
uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4.0.0
with:
path: |
~/.cache/pypoetry
~/.cache/pip
.venv
key: ${{ runner.os }}-poetry-e2e-batches-${{ hashFiles('poetry.lock') }}
restore-keys: |
${{ runner.os }}-poetry-e2e-batches-
${{ runner.os }}-poetry-
- name: Install dependencies
run: |
poetry config virtualenvs.in-project true
poetry install --with dev,proxy-dev --extras "proxy"
poetry run pip install psycopg2-binary==2.9.11 uvicorn==0.42.0 fastapi==0.135.2 httpx==0.28.1 tenacity==9.1.4
- name: Setup litellm-enterprise
run: |
poetry run pip install --force-reinstall --no-deps -e enterprise/
- name: Generate Prisma client
env:
PRISMA_BINARY_CACHE_DIR: ${{ runner.temp }}/prisma-cache
run: |
poetry run pip install nodejs-wheel-binaries==24.13.1
poetry run prisma generate --schema litellm/proxy/schema.prisma
- name: Run Prisma migrations
env:
DATABASE_URL: postgresql://llmproxy:dbpassword9090@localhost:5432/litellm
run: |
cd litellm/proxy
poetry run prisma migrate deploy --schema schema.prisma
cd ../..
- name: Run Azure Batch E2E Tests
env:
DATABASE_URL: postgresql://llmproxy:dbpassword9090@localhost:5432/litellm
USE_LOCAL_LITELLM: "true"
USE_MOCK_MODELS: "true"
USE_STATE_TRACKER: "true"
LITELLM_LOG: DEBUG
run: |
poetry run pytest tests/proxy_e2e_azure_batches_tests/test_proxy_e2e_azure_batches.py \
-vv -s -k "test_e2e_managed_batch" \
--tb=short \
--maxfail=3 \
--durations=10

View file

@ -8,6 +8,8 @@ on:
permissions:
contents: read
id-token: write
pull-requests: write
concurrency:
group: ${{ github.workflow }}-${{ github.ref }}
@ -29,6 +31,7 @@ jobs:
timeout-minutes: 20
enable-redis: true
enable-postgres: false
artifact-name: caching-redis
secrets:
REDIS_HOST: ${{ secrets.REDIS_HOST }}
REDIS_PORT: ${{ secrets.REDIS_PORT }}

View file

@ -6,6 +6,8 @@ on:
permissions:
contents: read
id-token: write
pull-requests: write
concurrency:
group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
@ -18,3 +20,4 @@ jobs:
test-path: "tests/test_litellm/litellm_core_utils"
workers: 2
reruns: 1
artifact-name: core-utils

View file

@ -6,6 +6,8 @@ on:
permissions:
contents: read
id-token: write
pull-requests: write
concurrency:
group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
@ -22,3 +24,4 @@ jobs:
tests/test_litellm/router_strategy
workers: 2
reruns: 2
artifact-name: enterprise-routing

View file

@ -6,6 +6,8 @@ on:
permissions:
contents: read
id-token: write
pull-requests: write
concurrency:
group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
@ -18,3 +20,4 @@ jobs:
test-path: "tests/test_litellm/integrations"
workers: 2
reruns: 3
artifact-name: integrations

View file

@ -14,16 +14,26 @@ concurrency:
jobs:
vertex-ai:
name: Vertex AI
permissions:
contents: read
id-token: write
pull-requests: write
uses: ./.github/workflows/_test-unit-base.yml
with:
test-path: "tests/test_litellm/llms/vertex_ai"
workers: 1
reruns: 2
artifact-name: llm-vertex-ai
other-providers:
name: All Other Providers
permissions:
contents: read
id-token: write
pull-requests: write
uses: ./.github/workflows/_test-unit-base.yml
with:
test-path: "tests/test_litellm/llms --ignore=tests/test_litellm/llms/vertex_ai"
workers: 2
reruns: 2
artifact-name: llm-other-providers

View file

@ -6,6 +6,8 @@ on:
permissions:
contents: read
id-token: write
pull-requests: write
concurrency:
group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
@ -29,3 +31,4 @@ jobs:
tests/test_litellm/test_*.py
workers: 2
reruns: 2
artifact-name: misc

View file

@ -6,6 +6,8 @@ on:
permissions:
contents: read
id-token: write
pull-requests: write
concurrency:
group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
@ -18,3 +20,4 @@ jobs:
test-path: "tests/test_litellm/proxy/auth tests/test_litellm/proxy/hooks tests/test_litellm/proxy/policy_engine tests/test_litellm/proxy/client"
workers: 2
reruns: 2
artifact-name: proxy-auth

View file

@ -14,6 +14,10 @@ concurrency:
jobs:
proxy-db:
permissions:
contents: read
id-token: write
pull-requests: write
strategy:
fail-fast: false
matrix:
@ -39,6 +43,7 @@ jobs:
timeout-minutes: ${{ matrix.timeout }}
enable-redis: false
enable-postgres: true
artifact-name: proxy-db-${{ matrix.test-group }}
secrets:
DATABASE_URL: ${{ secrets.DATABASE_URL }}
POSTGRES_USER: ${{ secrets.POSTGRES_USER }}

View file

@ -6,6 +6,8 @@ on:
permissions:
contents: read
id-token: write
pull-requests: write
concurrency:
group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
@ -33,3 +35,4 @@ jobs:
tests/test_litellm/proxy/ui_crud_endpoints
workers: 2
reruns: 2
artifact-name: proxy-endpoints

View file

@ -6,6 +6,8 @@ on:
permissions:
contents: read
id-token: write
pull-requests: write
concurrency:
group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
@ -26,3 +28,4 @@ jobs:
tests/test_litellm/proxy/test_*.py
workers: 2
reruns: 2
artifact-name: proxy-infra

View file

@ -6,6 +6,8 @@ on:
permissions:
contents: read
id-token: write
pull-requests: write
concurrency:
group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
@ -18,3 +20,4 @@ jobs:
test-path: "tests/test_litellm/responses tests/test_litellm/caching tests/test_litellm/types"
workers: 2
reruns: 2
artifact-name: responses-caching-types

View file

@ -7,6 +7,8 @@ on:
permissions:
contents: read
id-token: write
pull-requests: write
concurrency:
group: ${{ github.workflow }}-${{ github.ref }}
@ -22,6 +24,7 @@ jobs:
timeout-minutes: 20
enable-redis: false
enable-postgres: true
artifact-name: security
secrets:
DATABASE_URL: ${{ secrets.DATABASE_URL }}
POSTGRES_USER: ${{ secrets.POSTGRES_USER }}

View file

@ -1,12 +0,0 @@
# LiteLLM Trivy Ignore File
# CVEs listed here are temporarily allowlisted pending fixes
# Next.js vulnerabilities in UI dashboard (next@14.2.35)
# Allowlisted: 2026-01-31, 7-day fix timeline
# Fix: Upgrade to Next.js 15.5.10+ or 16.1.5+
# HIGH: DoS via request deserialization
GHSA-h25m-26qc-wcjf
# MEDIUM: Image Optimizer DoS
CVE-2025-59471

View file

@ -254,7 +254,7 @@ See `CLAUDE.md` and the `Makefile` for standard commands. Key notes:
- `openapi-core` must be installed (`poetry run pip install openapi-core`) for the OpenAPI compliance tests in `tests/test_litellm/interactions/`.
- The `--timeout` pytest flag is NOT available; don't pass it.
- Unit tests: `poetry run pytest tests/test_litellm/ -x -vv -n 4`
- Black `--check` may report pre-existing formatting issues; this does not block test runs.
- **Before committing, always run `poetry run black .` to format your code.** Black formatting is enforced in CI.
- If `poetry install` fails with "pyproject.toml changed significantly since poetry.lock was last generated", run `poetry lock` first to regenerate the lock file.
### Lint

View file

@ -20,6 +20,7 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
- `make format` - Apply Black code formatting
- `make lint-ruff` - Run Ruff linting only
- `make lint-mypy` - Run MyPy type checking only
- **Before committing, always run `poetry run black .` to format your code.** Black formatting is enforced in CI.
### Single Test Files
- `poetry run pytest tests/path/to/test_file.py -v` - Run specific test file

View file

@ -149,6 +149,19 @@ Apply formatting (auto-fixes issues):
make format
```
> **Black formatting is enforced in CI.** All PRs must pass the Black formatting check.
>
> - **AI coding agents** (Claude Code, Copilot, Cursor, etc.): `AGENTS.md` and `CLAUDE.md` instruct agents to run `poetry run black .` before committing.
> - **VS Code users**: Install the [Black Formatter extension](https://marketplace.visualstudio.com/items?itemName=ms-python.black-formatter) and enable format-on-save:
> ```json
> {
> "[python]": {
> "editor.defaultFormatter": "ms-python.black-formatter",
> "editor.formatOnSave": true
> }
> }
> ```
### CI Compatibility
To ensure your changes will pass CI, run the exact same checks locally:

View file

@ -1,36 +0,0 @@
ignore:
- vulnerability: CVE-2026-22184
reason: no fixed zlib package is available yet in the Wolfi repositories, so this is ignored temporarily until an upstream release exists
# Wolfi base image: Python 3.13 and Node from apk have no fixed builds in Wolfi yet / not applicable
- vulnerability: CVE-2025-55130
reason: Node in Wolfi apk; only used for Admin UI build/prisma
- vulnerability: CVE-2025-59465
reason: Node in Wolfi apk; only used for Admin UI build/prisma
- vulnerability: CVE-2025-55131
reason: Node in Wolfi apk; only used for Admin UI build/prisma
- vulnerability: CVE-2025-59466
reason: Node in Wolfi apk; only used for Admin UI build/prisma
- vulnerability: CVE-2026-21637
reason: Node in Wolfi apk; only used for Admin UI build/prisma
- vulnerability: CVE-2025-55132
reason: Node in Wolfi apk; only used for Admin UI build/prisma
- vulnerability: GHSA-hx9q-6w63-j58v
reason: orjson dumps recursion; allowlisted
- vulnerability: GHSA-73rr-hh4g-fpgx
reason: diff npm transitive dep; override in package.json, allowlisted
- vulnerability: CVE-2026-0865
reason: Python 3.13 in Wolfi base; no fixed apk build yet
- vulnerability: CVE-2025-15282
reason: Python 3.13 in Wolfi base; no fixed apk build yet
- vulnerability: CVE-2026-0672
reason: Python 3.13 in Wolfi base; no fixed apk build yet
- vulnerability: CVE-2025-15366
reason: Python 3.13 in Wolfi base; no fixed apk build yet
- vulnerability: CVE-2025-15367
reason: Python 3.13 in Wolfi base; no fixed apk build yet
- vulnerability: CVE-2025-11468
reason: Python 3.13 in Wolfi base; no fixed apk build yet
- vulnerability: CVE-2025-12781
reason: Python 3.13 in Wolfi base; no fixed apk build yet
- vulnerability: CVE-2026-1299
reason: Python 3.13 in Wolfi base; no fixed apk build yet

View file

@ -1,261 +0,0 @@
#!/bin/bash
# Security Scans Script for LiteLLM
# This script runs comprehensive security scans including Trivy and Grype
set -e
echo "Starting security scans for LiteLLM..."
# Function to install Trivy and required tools
install_trivy() {
echo "Installing Trivy and required tools..."
TRIVY_VERSION="0.35.0"
sudo apt-get update
sudo apt-get install -y wget jq curl bsdmainutils
wget -qO trivy.deb "https://github.com/aquasecurity/trivy/releases/download/v${TRIVY_VERSION}/trivy_${TRIVY_VERSION}_Linux-64bit.deb"
sudo dpkg -i trivy.deb
rm trivy.deb
echo "Trivy ${TRIVY_VERSION} installed successfully"
}
# Function to install Grype
install_grype() {
echo "Installing Grype..."
curl -sSfL https://raw.githubusercontent.com/anchore/grype/main/install.sh | sudo sh -s -- -b /usr/local/bin
echo "Grype installed successfully"
}
# Function to install ggshield
install_ggshield() {
echo "Installing ggshield..."
pip3 install --upgrade pip
pip3 install ggshield
echo "ggshield installed successfully"
}
# # Function to run secret detection scans
# run_secret_detection() {
# echo "Running secret detection scans..."
# if ! command -v ggshield &> /dev/null; then
# install_ggshield
# fi
# # Check if GITGUARDIAN_API_KEY is set (required for CI/CD)
# if [ -z "$GITGUARDIAN_API_KEY" ]; then
# echo "Warning: GITGUARDIAN_API_KEY environment variable is not set."
# echo "ggshield requires a GitGuardian API key to scan for secrets."
# echo "Please set GITGUARDIAN_API_KEY in your CI/CD environment variables."
# exit 1
# fi
# echo "Scanning codebase for secrets..."
# echo "Note: Large codebases may take several minutes due to API rate limits (50 requests/minute on free plan)"
# echo "ggshield will automatically handle rate limits and retry as needed."
# echo "Binary files, cache files, and build artifacts are excluded via .gitguardian.yaml"
# # Use --recursive for directory scanning and auto-confirm if prompted
# # .gitguardian.yaml will automatically exclude binary files, wheel files, etc.
# # GITGUARDIAN_API_KEY environment variable will be used for authentication
# echo y | ggshield secret scan path . --recursive || {
# echo ""
# echo "=========================================="
# echo "ERROR: Secret Detection Failed"
# echo "=========================================="
# echo "ggshield has detected secrets in the codebase."
# echo "Please review discovered secrets above, revoke any actively used secrets"
# echo "from underlying systems and make changes to inject secrets dynamically at runtime."
# echo ""
# echo "For more information, see: https://docs.gitguardian.com/secrets-detection/"
# echo "=========================================="
# echo ""
# exit 1
# }
# echo "Secret detection scans completed successfully"
# }
# Function to run Trivy scans
run_trivy_scans() {
echo "Running Trivy scans..."
echo "Scanning LiteLLM Docs..."
trivy fs --ignorefile .trivyignore --scanners vuln --dependency-tree --exit-code 1 --severity HIGH,CRITICAL,MEDIUM ./docs/
echo "Scanning LiteLLM UI..."
trivy fs --ignorefile .trivyignore --scanners vuln --dependency-tree --exit-code 1 --severity HIGH,CRITICAL,MEDIUM ./ui/
echo "Trivy scans completed successfully"
}
# Function to build and scan Docker images with Grype
run_grype_scans() {
echo "Running Grype scans..."
# Temporarily add wheel files to .dockerignore for security scans
echo "Temporarily modifying .dockerignore to exclude problematic wheel files..."
cp .dockerignore .dockerignore.backup 2>/dev/null || touch .dockerignore.backup
echo "/*.whl" >> .dockerignore
# Build and scan Dockerfile.database
echo "Building and scanning Dockerfile.database..."
docker build --no-cache -t litellm-database:latest -f ./docker/Dockerfile.database .
grype litellm-database:latest --config ci_cd/.grype.yaml --fail-on critical
# Build and scan main Dockerfile
echo "Building and scanning main Dockerfile..."
docker build --no-cache -t litellm:latest .
grype litellm:latest --config ci_cd/.grype.yaml --fail-on critical
# Restore original .dockerignore
echo "Restoring original .dockerignore..."
mv .dockerignore.backup .dockerignore
# Scan the locally built LiteLLM image for vulnerabilities with CVSS >= 4.0
echo "Scanning locally built LiteLLM image for high-severity vulnerabilities..."
echo "Using locally built image: litellm:latest"
# Allowlist of CVEs to be ignored in failure threshold/reporting
# - CVE-2025-8869: Not applicable on Python >=3.13 (PEP 706 implemented); pip fallback unused; no OS-level fix
# - GHSA-4xh5-x5gv-qwph: GitHub Security Advisory alias for CVE-2025-8869
# - GHSA-5j98-mcp5-4vw2: glob CLI command injection via -c/--cmd; glob CLI is not used in the litellm runtime image,
# and the vulnerable versions are pulled in only via OS-level/node tooling outside of our application code
ALLOWED_CVES=(
"CVE-2025-8869"
"GHSA-4xh5-x5gv-qwph"
"CVE-2025-8291" # no fix available as of Oct 11, 2025
"GHSA-5j98-mcp5-4vw2"
"CVE-2025-13836" # Python 3.13 HTTP response reading OOM/DoS - no fix available in base image
"CVE-2025-12084" # Python 3.13 xml.dom.minidom quadratic algorithm - no fix available in base image
"CVE-2025-60876" # BusyBox wget HTTP request splitting - no fix available in Chainguard Wolfi base image
"CVE-2026-0861" # Wolfi glibc still flagged even on 2.42-r5; upstream patched build unavailable yet
"CVE-2010-4756" # glibc glob DoS - awaiting patched Wolfi glibc build
"CVE-2019-1010022" # glibc stack guard bypass - awaiting patched Wolfi glibc build
"CVE-2019-1010023" # glibc ldd remap issue - awaiting patched Wolfi glibc build
"CVE-2019-1010024" # glibc ASLR mitigation bypass - awaiting patched Wolfi glibc build
"CVE-2019-1010025" # glibc pthread heap address leak - awaiting patched Wolfi glibc build
"CVE-2026-22184" # zlib untgz buffer overflow - untgz unused + no fixed Wolfi build yet
"GHSA-58pv-8j8x-9vj2" # jaraco.context path traversal - setuptools vendored only (v5.3.0), not used in application code (using v6.1.0+)
"GHSA-34x7-hfp2-rc4v" # node-tar hardlink path traversal - not applicable, tar CLI not exposed in application code
"GHSA-r6q2-hw4h-h46w" # node-tar not used by application runtime, Linux-only container, not affect by macOS APFS-specific exploit
"GHSA-8rrh-rw8j-w5fx" # wheel is from chainguard and will be handled by then TODO: Remove this after Chainguard updates the wheel
"CVE-2025-59465" # Node only used for Admin UI build/prisma
"CVE-2025-55131" # Node only used for Admin UI build/prisma
"CVE-2025-59466" # Node only used for Admin UI build/prisma
"CVE-2025-55130" # Node only used for Admin UI build/prisma
"CVE-2025-59467" # Node only used for Admin UI build/prisma
"CVE-2026-21637" # Node only used for Admin UI build/prisma
"CVE-2025-55132" # Node only used for Admin UI build/prisma
"GHSA-hx9q-6w63-j58v" # orjson dumps recursion; allowlisted
"CVE-2025-15281" # No fix available yet
"CVE-2026-0865" # No fix available yet
"CVE-2025-15282" # No fix available yet
"CVE-2026-0672" # No fix available yet
"CVE-2025-15366" # No fix available yet
"CVE-2025-15367" # No fix available yet
"CVE-2025-12781" # No fix available yet
"CVE-2025-11468" # No fix available yet
"CVE-2026-1299" # Python 3.13 email module header injection - not applicable, LiteLLM doesn't use BytesGenerator for email serialization
"CVE-2026-0775" # npm cli incorrect permission assignment - no fix available yet, npm is only used at build/prisma-generate time
"GHSA-3ppc-4f35-3m26" # minimatch ReDoS via repeated wildcards - from nodejs_wheel bundled npm, not used in application runtime code
"GHSA-83g3-92jg-28cx" # tar arbitrary file read/write via hardlink - from nodejs_wheel bundled npm, not used in application runtime code
"CVE-2026-2297" # Python 3.13 SourcelessFileLoader audit hook bypass - no fix available in base image
"GHSA-qffp-2rhf-9h96" # tar hardlink path traversal - from nodejs_wheel bundled npm, not used in application runtime code
"CVE-2026-2673" # OpenSSL 3.6.1 TLS 1.3 key exchange group negotiation issue - no fix available yet
"CVE-2026-3644" # Python 3.13 vulnerability - no fix available in base image
"CVE-2026-4224" # Python 3.13 Expat parser stack overflow in ElementDeclHandler - no fix available in base image
)
# Build JSON array of allowlisted CVE IDs for jq
ALLOWED_IDS_JSON=$(printf '%s\n' "${ALLOWED_CVES[@]}" | jq -R . | jq -s .)
echo "Checking for vulnerabilities with CVSS score >= 4.0..."
echo "Allowlisted CVEs (ignored in threshold): ${ALLOWED_CVES[*]}"
echo ""
# Show all high-severity vulnerabilities for transparency
TOTAL_HIGH_SEVERITY=$(grype litellm:latest -o json | jq -r '
.matches[]
| select(.vulnerability.cvss[]?.metrics.baseScore >= 4.0)
| .vulnerability.id' | wc -l)
if [ "$TOTAL_HIGH_SEVERITY" -gt 0 ]; then
echo "Total vulnerabilities found with CVSS >= 4.0: $TOTAL_HIGH_SEVERITY"
echo ""
echo "All high-severity vulnerabilities (including allowlisted):"
grype litellm:latest -o json | jq --argjson allow "$ALLOWED_IDS_JSON" -r '
["Package", "Version", "Vulnerability ID", "CVSS Score", "Allowlisted"],
(.matches[]
| select(.vulnerability.cvss[]?.metrics.baseScore >= 4.0)
| [.artifact.name, .artifact.version, .vulnerability.id, .vulnerability.cvss[0].metrics.baseScore, (if (.vulnerability.id as $id | $allow | index($id)) then "YES" else "NO" end)])
| @tsv' | column -t -s $'\t'
echo ""
fi
HIGH_SEVERITY_COUNT=$(grype litellm:latest -o json | jq --argjson allow "$ALLOWED_IDS_JSON" -r '
.matches[]
| select(.vulnerability.cvss[]?.metrics.baseScore >= 4.0)
| select((.vulnerability.id as $id | $allow | index($id) | not))
| .vulnerability.id' | wc -l)
if [ "$HIGH_SEVERITY_COUNT" -gt 0 ]; then
echo ""
echo "=========================================="
echo "ERROR: Security Scan Failed"
echo "=========================================="
echo "Found $HIGH_SEVERITY_COUNT non-allowlisted vulnerabilities with CVSS score >= 4.0 in litellm:latest"
echo ""
echo "These vulnerabilities are NOT in the allowlist and must be addressed."
echo "Current allowlisted CVEs: ${ALLOWED_CVES[*]}"
echo ""
echo "Detailed vulnerability report:"
echo ""
grype litellm:latest -o json | jq --argjson allow "$ALLOWED_IDS_JSON" -r '
["Package", "Version", "Vulnerability ID", "CVSS Score", "Severity", "Fix Version", "Description"],
(.matches[]
| select(.vulnerability.cvss[]?.metrics.baseScore >= 4.0)
| select((.vulnerability.id as $id | $allow | index($id) | not))
| [.artifact.name, .artifact.version, .vulnerability.id, .vulnerability.cvss[0].metrics.baseScore, .vulnerability.severity, (.vulnerability.fix.versions[0] // "No fix available"), .vulnerability.description])
| @tsv' | column -t -s $'\t'
echo ""
echo "=========================================="
echo "Action Required:"
echo "=========================================="
echo "1. If a fix is available, update the package to the fixed version"
echo "2. If the vulnerability is not applicable or has no fix:"
echo " - Add the CVE/GHSA ID to ALLOWED_CVES array in ci_cd/security_scans.sh"
echo " - Add a comment explaining why it's safe to ignore"
echo ""
echo "Note: Some vulnerabilities may have multiple IDs (CVE-XXXX and GHSA-XXXX)."
echo "Add all relevant IDs to the allowlist if they refer to the same issue."
echo "=========================================="
echo ""
exit 1
else
echo "No high-severity vulnerabilities (CVSS >= 4.0) found in litellm:latest"
fi
echo "Grype scans completed successfully"
}
# Main execution
main() {
echo "Installing security scanning tools..."
install_trivy
install_grype
# echo "Running secret detection scans..."
# run_secret_detection
echo "Running filesystem vulnerability scans..."
run_trivy_scans
echo "Running Docker image vulnerability scans..."
run_grype_scans
echo "All security scans completed successfully!"
}
# Execute main function
main "$@"

View file

@ -1,7 +0,0 @@
# js-yaml CVE-2025-64718
# This vulnerability is not applicable because we've forced js-yaml to version 4.1.1
# via npm overrides in package.json. Trivy incorrectly reports this based on
# dependency requirements in the lockfile, but the actual installed version is 4.1.1.
# Verified with: npm list js-yaml
CVE-2025-64718

View file

@ -27,6 +27,27 @@ Building on the roadmap from our [security incident](https://docs.litellm.ai/blo
- Validation and release are separated into different repositories, making it harder for an attacker to reach release credentials.
- Trusted Publishing for PyPI releases - this means no long-lived credentials are used to publish releases.
- Immutable Docker release tags - this means no tampering of Docker release tags after they are published [Learn more](https://docs.docker.com/docker-hub/repos/manage/hub-images/immutable-tags/). Note: work for GHCR docker releases is planned as well.
- Docker image signing with [Cosign](https://github.com/sigstore/cosign) - all release images are signed so users can independently verify they came from us.
## Verify Docker image signatures
Starting from `v1.83.0-nightly`, all LiteLLM Docker images published to GHCR are signed with [cosign](https://docs.sigstore.dev/cosign/overview/). To verify the integrity of an image before deploying:
```bash
cosign verify \
--key https://raw.githubusercontent.com/BerriAI/litellm/<release-tag>/cosign.pub \
ghcr.io/berriai/litellm:<release-tag>
```
Replace `<release-tag>` with the version you are deploying (e.g. `v1.83.0-stable`).
Expected output:
```
The following checks were performed on each of these signatures:
- The cosign claims were validated
- The signatures were verified against the specified public key
```
## What's next

View file

@ -143,8 +143,27 @@ This will ensure, your releases are safe, even when:
- Tampered registry artifacts are published
- Tag mutations are made after the release is published
We believe that [Cosign](https://github.com/sigstore/cosign) is a good fit for this, and have already begun working on it [PR](https://github.com/BerriAI/litellm/pull/24683).
We believe that [Cosign](https://github.com/sigstore/cosign) is a good fit for this, and have shipped it in [PR #24683](https://github.com/BerriAI/litellm/pull/24683).
#### How to verify a Docker image with Cosign
Starting from `v1.83.0-nightly`, all LiteLLM Docker images published to GHCR are signed with [cosign](https://docs.sigstore.dev/cosign/overview/). To verify the integrity of an image before deploying:
```bash
cosign verify \
--key https://raw.githubusercontent.com/BerriAI/litellm/<release-tag>/cosign.pub \
ghcr.io/berriai/litellm:<release-tag>
```
Replace `<release-tag>` with the version you are deploying (e.g. `v1.83.0-stable`).
Expected output:
```
The following checks were performed on each of these signatures:
- The cosign claims were validated
- The signatures were verified against the specified public key
```
### Avoid Compromised Packages

View file

@ -708,6 +708,26 @@ The LiteLLM AI Gateway team has already taken the following steps:
- Engaged Google's Mandiant security team to assist with forensic analysis of the build and publishing chain
## Verify Docker image signatures
Starting from `v1.83.0-nightly`, all LiteLLM Docker images published to GHCR are signed with [cosign](https://docs.sigstore.dev/cosign/overview/). To verify the integrity of an image before deploying:
```bash
cosign verify \
--key https://raw.githubusercontent.com/BerriAI/litellm/<release-tag>/cosign.pub \
ghcr.io/berriai/litellm:<release-tag>
```
Replace `<release-tag>` with the version you are deploying (e.g. `v1.83.0-stable`).
Expected output:
```
The following checks were performed on each of these signatures:
- The cosign claims were validated
- The signatures were verified against the specified public key
```
## Verified safe versions
We have audited every LiteLLM release published between v1.78.0 and v1.82.6 across both PyPI and Docker. Each artifact was verified by:

View file

@ -1032,6 +1032,7 @@ router_settings:
| SENDGRID_SENDER_EMAIL | Email address used as the sender in SendGrid email transactions
| SPEND_LOGS_URL | URL for retrieving spend logs
| SPEND_LOG_CLEANUP_BATCH_SIZE | Number of logs deleted per batch during cleanup. Default is 1000
| STALE_OBJECT_CLEANUP_BATCH_SIZE | Max number of stale managed objects updated per cleanup cycle. Default is 1000
| SSL_CERTIFICATE | Path to the SSL certificate file
| SSL_ECDH_CURVE | ECDH curve for SSL/TLS key exchange (e.g., 'X25519' to disable PQC).
| SSL_SECURITY_LEVEL | [BETA] Security level for SSL/TLS connections. E.g. `DEFAULT@SECLEVEL=1`

View file

@ -65,7 +65,29 @@ docker compose up
</TabItem>
</Tabs>
### Docker Run
### Verify Docker image signatures
All LiteLLM Docker images are signed with [cosign](https://docs.sigstore.dev/cosign/overview/). You can verify the integrity of an image before deploying:
```bash
cosign verify \
--key https://raw.githubusercontent.com/BerriAI/litellm/<release-tag>/cosign.pub \
ghcr.io/berriai/litellm:<release-tag>
```
Replace `<release-tag>` with the version you are deploying (e.g. `v1.83.0-stable`).
Expected output:
```
The following checks were performed on each of these signatures:
- The cosign claims were validated
- The signatures were verified against the specified public key
```
Learn more about LiteLLM's release signing in the [CI/CD v2 announcement](https://docs.litellm.ai/blog/ci-cd-v2-improvements#verify-docker-image-signatures).
### Docker Run
#### Step 1. CREATE config.yaml

View file

@ -11,6 +11,7 @@ from litellm._logging import verbose_proxy_logger
from litellm.constants import (
MANAGED_OBJECT_STALENESS_CUTOFF_DAYS,
MAX_OBJECTS_PER_POLL_CYCLE,
STALE_OBJECT_CLEANUP_BATCH_SIZE,
)
if TYPE_CHECKING:
@ -32,21 +33,49 @@ class CheckResponsesCost:
self.prisma_client: PrismaClient = prisma_client
self.llm_router: Router = llm_router
async def _expire_stale_rows(
self, cutoff: datetime, batch_size: int
) -> int:
"""Execute the bounded UPDATE that marks stale rows as 'stale_expired'.
Isolated so it can be swapped / mocked in tests without touching the
orchestration logic in ``_cleanup_stale_managed_objects``.
Uses PostgreSQL syntax (``$1::timestamptz``, ``LIMIT``, double-quoted
identifiers) which is the only dialect the proxy supports every
``schema.prisma`` in the repo sets ``provider = "postgresql"``.
Same pattern as ``spend_log_cleanup.py``.
"""
return await self.prisma_client.db.execute_raw(
"""
UPDATE "LiteLLM_ManagedObjectTable"
SET "status" = 'stale_expired'
WHERE "id" IN (
SELECT "id" FROM "LiteLLM_ManagedObjectTable"
WHERE "file_purpose" = 'response'
AND "status" NOT IN ('completed', 'complete', 'failed', 'expired', 'cancelled', 'stale_expired')
AND "created_at" < $1::timestamptz
ORDER BY "created_at" ASC
LIMIT $2
)
""",
cutoff,
batch_size,
)
async def _cleanup_stale_managed_objects(self) -> None:
"""
Mark managed objects older than MANAGED_OBJECT_STALENESS_CUTOFF_DAYS days
in non-terminal states as 'stale_expired'. These will never complete and
should not be polled.
Runs as a single DB query with a subquery LIMIT so no rows are loaded
into Python memory. Processes at most STALE_OBJECT_CLEANUP_BATCH_SIZE
rows per invocation to avoid overwhelming the DB when there is a large
backlog.
"""
cutoff = datetime.now(timezone.utc) - timedelta(days=MANAGED_OBJECT_STALENESS_CUTOFF_DAYS)
result = await self.prisma_client.db.litellm_managedobjecttable.update_many(
where={
"file_purpose": "response",
"status": {"not_in": ["completed", "complete", "failed", "expired", "cancelled", "stale_expired"]},
"created_at": {"lt": cutoff},
},
data={"status": "stale_expired"},
)
result = await self._expire_stale_rows(cutoff, STALE_OBJECT_CLEANUP_BATCH_SIZE)
if result > 0:
verbose_proxy_logger.warning(
f"CheckResponsesCost: marked {result} stale managed objects "

View file

@ -1367,6 +1367,9 @@ MAX_OBJECTS_PER_POLL_CYCLE = max(1, int(os.getenv("MAX_OBJECTS_PER_POLL_CYCLE",
MANAGED_OBJECT_STALENESS_CUTOFF_DAYS = max(
1, int(os.getenv("MANAGED_OBJECT_STALENESS_CUTOFF_DAYS", 7))
)
STALE_OBJECT_CLEANUP_BATCH_SIZE = max(
1, int(os.getenv("STALE_OBJECT_CLEANUP_BATCH_SIZE", 1000))
)
# Set PROXY_BATCH_POLLING_ENABLED=false to disable the CheckBatchCost and
# CheckResponsesCost background polling jobs entirely (e.g. to avoid DB load on
# installations with large numbers of stale managed objects).

View file

@ -100,6 +100,8 @@ from litellm.types.proxy.management_endpoints.common_daily_activity import (
from litellm.types.proxy.management_endpoints.team_endpoints import (
BulkTeamMemberAddRequest,
BulkTeamMemberAddResponse,
BulkUpdateTeamMemberPermissionsRequest,
BulkUpdateTeamMemberPermissionsResponse,
GetTeamMemberPermissionsResponse,
TeamListItem,
TeamListResponse,
@ -4274,6 +4276,151 @@ async def update_team_member_permissions(
return updated_team
@router.post(
"/team/permissions_bulk_update",
tags=["team management"],
dependencies=[Depends(user_api_key_auth)],
response_model=BulkUpdateTeamMemberPermissionsResponse,
)
@management_endpoint_wrapper
async def bulk_update_team_member_permissions(
data: BulkUpdateTeamMemberPermissionsRequest,
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
"""
Append permissions to existing teams.
Either pass team_ids to target specific teams, or set
apply_to_all_teams=True to update every team. For each team,
the provided permissions are merged with the team's existing
permissions (duplicates are skipped).
"""
from litellm.proxy.proxy_server import prisma_client
if prisma_client is None:
raise HTTPException(status_code=500, detail={"error": "No db connected"})
if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value:
raise HTTPException(
status_code=403,
detail={"error": "Only proxy admins can bulk-update team permissions"},
)
if not data.permissions:
return {
"message": "No permissions provided",
"teams_updated": 0,
}
if not data.apply_to_all_teams and not data.team_ids:
raise HTTPException(
status_code=400,
detail={
"error": "Must provide team_ids or set apply_to_all_teams=true"
},
)
if data.apply_to_all_teams and data.team_ids:
raise HTTPException(
status_code=400,
detail={
"error": "Cannot set both apply_to_all_teams=true and team_ids"
},
)
permissions_to_add = set(data.permissions)
if data.team_ids:
teams_updated = await _append_permissions_to_specific_teams(
prisma_client, data.team_ids, permissions_to_add
)
else:
teams_updated = await _append_permissions_to_all_teams(
prisma_client, permissions_to_add
)
return {
"message": "Team permissions updated successfully",
"teams_updated": teams_updated,
"permissions_appended": data.permissions,
}
async def _compute_and_batch_updates(prisma_client, teams, permissions_to_add: set) -> int:
"""Compute merged permissions and batch-write updates. Returns count of teams updated."""
updates = []
for team in teams:
existing = set(team.team_member_permissions or [])
if permissions_to_add <= existing:
continue
merged = sorted(existing | permissions_to_add) # normalise to alphabetical order
updates.append((team.team_id, merged))
if updates:
batcher = prisma_client.db.batch_()
for team_id, merged_perms in updates:
batcher.litellm_teamtable.update(
where={"team_id": team_id},
data={"team_member_permissions": merged_perms},
)
await batcher.commit()
return len(updates)
async def _append_permissions_to_specific_teams(
prisma_client, team_ids: List[str], permissions_to_add: set
) -> int:
"""Fetch specific teams by ID and append permissions."""
teams = await prisma_client.db.litellm_teamtable.find_many(
where={"team_id": {"in": team_ids}},
)
found_ids = {team.team_id for team in teams}
missing_ids = set(team_ids) - found_ids
if missing_ids:
raise HTTPException(
status_code=404,
detail={"error": f"Team(s) not found: {sorted(missing_ids)}"},
)
return await _compute_and_batch_updates(prisma_client, teams, permissions_to_add)
async def _append_permissions_to_all_teams(
prisma_client, permissions_to_add: set
) -> int:
"""Paginated read + batched write across all teams."""
teams_updated = 0
cursor = None
BATCH_SIZE = 500
while True:
find_args: dict = {
"take": BATCH_SIZE,
"order": {"team_id": "asc"},
}
if cursor is not None:
find_args["cursor"] = {"team_id": cursor}
find_args["skip"] = 1
teams = await prisma_client.db.litellm_teamtable.find_many(**find_args)
if not teams:
break
teams_updated += await _compute_and_batch_updates(
prisma_client, teams, permissions_to_add
)
cursor = teams[-1].team_id
if len(teams) < BATCH_SIZE:
break
return teams_updated
@router.get(
"/team/daily/activity",
response_model=SpendAnalyticsPaginatedResponse,

View file

@ -3,6 +3,7 @@ from typing import Any, Dict, List, Optional, Union
from pydantic import BaseModel
from litellm.proxy._types import (
KeyManagementRoutes,
LiteLLM_DeletedTeamTable,
LiteLLM_TeamMembership,
LiteLLM_TeamTable,
@ -43,6 +44,27 @@ class UpdateTeamMemberPermissionsRequest(BaseModel):
team_member_permissions: List[str]
class BulkUpdateTeamMemberPermissionsRequest(BaseModel):
"""Request to bulk-update team member permissions across teams."""
permissions: List[KeyManagementRoutes]
"""Permissions to append to the target teams (duplicates are skipped)."""
team_ids: Optional[List[str]] = None
"""Specific team IDs to update. Required unless apply_to_all_teams is True."""
apply_to_all_teams: bool = False
"""When True, update all teams. Mutually exclusive with team_ids."""
class BulkUpdateTeamMemberPermissionsResponse(BaseModel):
"""Response for bulk team member permissions update."""
message: str
teams_updated: int
permissions_appended: Optional[List[str]] = None
class TeamListItem(LiteLLM_TeamTable):
"""A team item in the paginated list response, enriched with computed fields."""

3900
poetry.lock generated

File diff suppressed because it is too large Load diff

View file

@ -1,6 +1,6 @@
[tool.poetry]
name = "litellm"
version = "1.83.3"
version = "1.83.4"
description = "Library to easily interface with LLM API providers"
authors = ["BerriAI"]
license = "MIT"
@ -65,7 +65,7 @@ mcp = {version = "1.26.0", optional = true, python = ">=3.10"}
a2a-sdk = {version = "0.3.25", optional = true, python = ">=3.10"}
litellm-proxy-extras = {version = "0.4.65", optional = true}
rich = {version = "13.9.4", optional = true}
litellm-enterprise = {version = "0.1.36", optional = true}
litellm-enterprise = {version = "0.1.37", optional = true}
diskcache = {version = "5.6.3", optional = true}
polars = {version = "1.39.3", optional = true, python = ">=3.10"}
semantic-router = {version = "0.1.12", optional = true, python = ">=3.9,<3.14"}
@ -181,7 +181,7 @@ requires = ["poetry-core", "wheel"]
build-backend = "poetry.core.masonry.api"
[tool.commitizen]
version = "1.83.3"
version = "1.83.4"
version_files = [
"pyproject.toml:^version"
]

View file

@ -85,4 +85,4 @@ requests-toolbelt==1.0.0 # transitive dep (langfuse)
########################
# LITELLM ENTERPRISE DEPENDENCIES
########################
litellm-enterprise==0.1.36
litellm-enterprise==0.1.37

View file

@ -373,35 +373,6 @@ def test_openai_azure_embedding_optional_arg():
# test_openai_embedding()
@pytest.mark.parametrize(
"model, api_base",
[
("embed-english-v2.0", None),
],
)
@pytest.mark.parametrize("sync_mode", [True, False])
@pytest.mark.asyncio
async def test_cohere_embedding(sync_mode, model, api_base):
try:
# litellm.set_verbose=True
data = {
"model": model,
"input": ["good morning from litellm", "this is another item"],
"input_type": "search_query",
"api_base": api_base,
}
if sync_mode:
response = embedding(**data)
else:
response = await litellm.aembedding(**data)
print(f"response:", response)
assert isinstance(response.usage, litellm.Usage)
except Exception as e:
pytest.fail(f"Error occurred: {e}")
# test_cohere_embedding()

View file

@ -1,494 +0,0 @@
"""Base class for LiteLLM integration tests.
Supports both local (mock) and remote testing modes via environment variables:
- USE_LOCAL_LITELLM: When "true", uses local LiteLLM at localhost:4000 (default: false)
- USE_MOCK_MODELS: When "true", uses mock model names (default: false)
- LITELLM_API_KEY: API key for remote LiteLLM (required when USE_LOCAL_LITELLM=false)
- LITELLM_BASE_URL: Base URL for remote LiteLLM (required when USE_LOCAL_LITELLM=false)
"""
import enum
import os
import time
import uuid
from abc import ABC
from collections import defaultdict
from typing import Any, Callable, Dict, List, Tuple, Union
import httpx
import openai
import pytest
import requests
from urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)
LOCAL_LITELLM_BASE_URL = "http://localhost:4000"
LOCAL_MOCK_SERVER_URL = "http://localhost:8090"
if "USE_LOCAL_LITELLM" not in os.environ:
os.environ["USE_LOCAL_LITELLM"] = "true"
if "USE_MOCK_MODELS" not in os.environ:
os.environ["USE_MOCK_MODELS"] = "true"
if "USE_STATE_TRACKER" not in os.environ:
os.environ["USE_STATE_TRACKER"] = "true"
if "DATABASE_URL" not in os.environ:
os.environ["DATABASE_URL"] = "postgresql://llmproxy:dbpassword9090@localhost:5432/litellm"
def use_local_litellm() -> bool:
return os.environ.get("USE_LOCAL_LITELLM", "false").lower() == "true"
def use_remote_litellm() -> bool:
return not use_local_litellm()
def use_mock_models() -> bool:
return os.environ.get("USE_MOCK_MODELS", "false").lower() == "true"
def get_local_litellm_base_url() -> str:
return LOCAL_LITELLM_BASE_URL
def get_remote_litellm_base_url() -> str:
return os.environ.get("LITELLM_BASE_URL", "").rstrip("/")
def get_litellm_base_url() -> str:
if use_local_litellm():
return get_local_litellm_base_url()
return get_remote_litellm_base_url()
def get_litellm_api_key() -> str:
if use_local_litellm():
return "sk-1234"
return os.environ.get("LITELLM_API_KEY", "")
def get_mock_server_base_url() -> str:
return LOCAL_MOCK_SERVER_URL
def get_responses_model_name() -> str:
if use_mock_models():
return "openai-fake-gpt-4o"
return "gpt-4o-mini-2024-07-18"
def model_id(param) -> str:
"""Generate a test ID from a model name or tuple containing model name.
Handles both:
- String: "gpt-4o-mini" -> "gpt_4o_mini"
- Tuple: ("gpt-4o", "openai/gpt-4o") -> "gpt_4o"
"""
if isinstance(param, tuple):
name = param[0]
else:
name = param
return name.replace("-", "_").replace(".", "_")
def generate_test_id(
params: Tuple[str, ...],
test_name: str = "test",
) -> str:
"""Generate test ID from model parameters tuple.
Handles two tuple formats:
- 6 elements: (provider, deployment, model_name, api_version, action, reason)
- 7 elements: (provider, deployment, model_name, api_version, model_id, action, reason)
Uses model_id (position 4) if 7 elements, otherwise model_name (position 2).
"""
provider = params[0]
deployment = params[1]
api_version = params[3]
if len(params) == 7:
identifier = params[4] # model_id
else:
identifier = params[2] # model_name
test_id = "/".join([provider, deployment, api_version, identifier, test_name])
return test_id.replace("-", "_").replace(".", "_")
class ModelTestAction(enum.Enum):
NOT_APPLICABLE = 1
SKIP = 2
RUN = 3
WARN_ON_FAIL = 4
def applicable(self) -> bool:
return self.value != ModelTestAction.NOT_APPLICABLE.value
class BaseLiteLLMIntegrationTest(ABC):
"""Base class for all LiteLLM integration tests.
Supports both local/mock and remote testing based on environment variables.
"""
@staticmethod
def get_api_key() -> str:
return get_litellm_api_key()
@staticmethod
def get_base_url() -> str:
return get_litellm_base_url()
@staticmethod
def get_ca_bundle_path() -> str:
current_dir = os.path.dirname(os.path.abspath(__file__))
# change if needed
@classmethod
def _get_ssl_verify_setting(cls) -> Union[bool, str]:
"""Get the appropriate SSL verification setting based on mode.
Returns path string (not SSLContext) for compatibility with both
requests and httpx libraries.
"""
if use_local_litellm():
return False
ca_bundle_path = cls.get_ca_bundle_path()
if os.path.exists(ca_bundle_path):
return ca_bundle_path
return True
@classmethod
def setup_class(cls):
cls.api_key = cls.get_api_key()
cls.base_url = cls.get_base_url()
if not cls.api_key:
pytest.fail(
"API key is not available. Set LITELLM_API_KEY or USE_LOCAL_LITELLM=true",
)
if not cls.base_url:
pytest.fail(
"Base URL is not available. Set LITELLM_BASE_URL or USE_LOCAL_LITELLM=true",
)
verify_setting = cls._get_ssl_verify_setting()
if use_remote_litellm() and isinstance(verify_setting, str):
os.environ["REQUESTS_CA_BUNDLE"] = verify_setting
os.environ["CURL_CA_BUNDLE"] = verify_setting
print(f"Using CA bundle: {verify_setting}")
cls.openai_client = openai.OpenAI(
base_url=cls.base_url,
api_key=cls.api_key,
http_client=httpx.Client(verify=verify_setting),
)
@classmethod
def make_request(
cls,
method: str,
endpoint: str,
timeout_secs: int,
**kwargs,
) -> requests.Response:
headers = kwargs.get("headers", {})
headers["Authorization"] = f"Bearer {cls.api_key}"
kwargs["headers"] = headers
kwargs.setdefault("timeout", timeout_secs)
kwargs.setdefault("verify", cls._get_ssl_verify_setting())
url = f"{cls.base_url}{endpoint}"
return requests.request(method, url, **kwargs)
@staticmethod
def generate_request_id() -> str:
return f"req-{uuid.uuid4().hex[:8]}"
@staticmethod
def get_timeout_secs(model_name: str) -> int:
model_lower = model_name.lower()
slow_models = ["gpt-5", "gpt_5", "o1", "claude-opus", "claude_opus", "o3", "o4"]
if any(slow_model in model_lower for slow_model in slow_models):
return 300
return 60
@staticmethod
def generate_unique_filename(extension: str = "txt") -> str:
return f"test_{time.time()}.{extension}"
@staticmethod
def extract_model_params(model_data: Dict[str, Any]) -> Tuple[str, str, str, str]:
"""Extract standardized parameters from model data."""
model_name = model_data.get("model_name", "")
model_info = model_data.get("model_info", {})
provider = model_info.get("litellm_provider", "unknown")
litellm_params = model_data.get("litellm_params", {})
if provider == "azure":
api_base = litellm_params.get("api_base", "unknown")
if api_base != "unknown" and "//" in api_base:
domain_name = api_base.split("//")[1]
deployment = domain_name.split(".")[0]
else:
deployment = "unknown"
api_version = litellm_params.get("api_version", "unknown")
elif provider in ["bedrock", "bedrock_converse"]:
deployment = litellm_params.get("aws_region_name", "unknown")
api_version = "unknown"
else:
deployment = "unknown"
api_version = "unknown"
return provider, deployment, model_name, api_version
@classmethod
def _fetch_all_models_from_litellm(cls) -> List[Dict[str, Any]]:
base_url = cls.get_base_url()
api_key = cls.get_api_key()
if not api_key or not base_url:
return []
verify_setting = cls._get_ssl_verify_setting()
response = requests.get(
f"{base_url}/model/info",
headers={"Authorization": f"Bearer {api_key}"},
verify=verify_setting,
timeout=30,
)
if response.status_code != 200:
raise RuntimeError(
f"Failed to fetch all models from {base_url}. Response code: {response.status_code}",
)
data = response.json()
return data.get("data", [])
@classmethod
def _fetch_all_approved_models(cls) -> List[Dict[str, Any]]:
return cls._fetch_all_models_from_litellm()
@classmethod
def build_model_test_params(
cls,
should_skip_model: Callable[
[str, str, str, str, Dict[str, Any]],
Tuple["ModelTestAction", str],
],
include_model_id: bool = False,
include_load_balanced: bool = False,
) -> List[Tuple[str, ...]]:
"""Build test parameters from all approved models.
Args:
should_skip_model: Callback that determines if a model should be skipped.
Signature: (provider, deployment, model_name, api_version, model_info) -> (action, reason)
include_model_id: If True, includes model_id in tuple (7 elements), else 6 elements.
include_load_balanced: If True, adds extra tests for load-balanced model groups.
Returns:
List of tuples with model test parameters.
- 6-element: (provider, deployment, model_name, api_version, action, reason)
- 7-element: (provider, deployment, model_name, api_version, model_id, action, reason)
"""
models = cls._fetch_all_approved_models()
test_params: List[Tuple[str, ...]] = []
models_by_model_name: Dict[str, List[Tuple[str, ...]]] = defaultdict(list)
for model_data in models:
model_info = model_data.get("model_info", {}) or {}
provider, deployment, model_name, api_version = cls.extract_model_params(
model_data,
)
model_test_action, model_test_action_reason = should_skip_model(
provider,
deployment,
model_name,
api_version,
model_info,
)
if model_test_action.applicable():
if include_model_id:
model_id = str(model_info.get("id"))
params_tuple: Tuple[str, ...] = (
provider,
deployment,
model_name,
api_version,
model_id,
model_test_action,
model_test_action_reason,
)
else:
params_tuple = (
provider,
deployment,
model_name,
api_version,
model_test_action,
model_test_action_reason,
)
test_params.append(params_tuple)
if include_load_balanced:
models_by_model_name[model_name].append(params_tuple)
if include_load_balanced and include_model_id:
for load_balanced_model_name, deployments in models_by_model_name.items():
if len(deployments) <= 1:
continue
first_deployment = deployments[0]
test_params.append(
(
first_deployment[0], # provider
"load_balanced",
load_balanced_model_name,
"load_balanced",
load_balanced_model_name, # model_id = model_name for LB
first_deployment[5], # model_test_action
first_deployment[6], # model_test_action_reason
),
)
return test_params
class UserKeyTestMixin:
"""Mixin for tests that need to create users and API keys."""
allowed_routes: list[str] = []
_base_url: str = None
_master_api_key: str = None
admin_client: httpx.Client = None
@classmethod
def setup_admin_client(cls):
cls._base_url = get_litellm_base_url()
cls._master_api_key = get_litellm_api_key()
verify_setting = (
False
if use_local_litellm()
else BaseLiteLLMIntegrationTest._get_ssl_verify_setting()
)
cls.admin_client = httpx.Client(base_url=cls._base_url, verify=verify_setting)
@classmethod
def teardown_admin_client(cls):
if cls.admin_client:
cls.admin_client.close()
@staticmethod
def unique_suffix() -> str:
return f"{time.strftime('%Y%m%d%H%M%S')}{int(time.time() * 1000) % 1000:03d}"
@classmethod
def create_user_and_key(cls, user_suffix: str) -> tuple[str, str, str]:
user_email = f"test-user-{user_suffix}-{cls.unique_suffix()}@test.com"
user_response = cls.admin_client.post(
"/user/new",
json={
"user_email": user_email,
"user_alias": user_email,
"user_role": "internal_user",
"auto_create_key": "false",
},
headers={
"Authorization": f"Bearer {cls._master_api_key}",
"Content-Type": "application/json",
},
timeout=30,
)
assert user_response.status_code == 200, (
f"Failed to create user: {user_response.status_code} - {user_response.text}"
)
user_id = user_response.json().get("user_id")
key_alias = user_email.replace("@", "-at-").replace(".", "-")
key_response = cls.admin_client.post(
"/key/generate",
json={
"user_id": user_id,
"key_alias": key_alias,
"allowed_routes": cls.allowed_routes,
},
headers={
"Authorization": f"Bearer {cls._master_api_key}",
"Content-Type": "application/json",
},
timeout=30,
)
assert key_response.status_code == 200, (
f"Failed to create key: {key_response.status_code} - {key_response.text}"
)
api_key = key_response.json().get("key")
print(f"Created user {user_email}")
return user_id, api_key, user_email
@classmethod
def create_user_key_and_client(
cls,
user_suffix: str,
) -> tuple[str, str, str, openai.OpenAI]:
user_id, api_key, user_email = cls.create_user_and_key(user_suffix)
verify_setting = (
False
if use_local_litellm()
else BaseLiteLLMIntegrationTest._get_ssl_verify_setting()
)
client = openai.OpenAI(
base_url=cls._base_url,
api_key=api_key,
http_client=httpx.Client(verify=verify_setting),
)
return user_id, api_key, user_email, client
@classmethod
def create_key_and_client(
cls,
user_id: str,
key_suffix: str,
) -> tuple[str, openai.OpenAI]:
key_alias = f"additional-key-{key_suffix}-{cls.unique_suffix()}"
key_response = cls.admin_client.post(
"/key/generate",
json={
"user_id": user_id,
"key_alias": key_alias,
"allowed_routes": cls.allowed_routes,
},
headers={
"Authorization": f"Bearer {cls._master_api_key}",
"Content-Type": "application/json",
},
timeout=30,
)
assert key_response.status_code == 200, (
f"Failed to create additional key: {key_response.status_code} - {key_response.text}"
)
api_key = key_response.json().get("key")
verify_setting = (
False
if use_local_litellm()
else BaseLiteLLMIntegrationTest._get_ssl_verify_setting()
)
client = openai.OpenAI(
base_url=cls._base_url,
api_key=api_key,
http_client=httpx.Client(verify=verify_setting),
)
print(f"Created additional key for user {user_id}")
return api_key, client

View file

@ -1,311 +0,0 @@
"""
Pytest configuration for Azure Batch E2E Tests.
This conftest manages:
1. Mock Azure Batch server (FastAPI on port 8090)
2. LiteLLM proxy server (port 4000)
3. PostgreSQL database setup
"""
import asyncio
import os
import subprocess
import sys
import time
from pathlib import Path
from typing import Generator
import httpx
import pytest
_test_dir = Path(__file__).parent
sys.path.insert(0, str(_test_dir.parent.parent)) # litellm root
sys.path.insert(0, str(_test_dir)) # test directory for local imports
LOG_DIR = _test_dir
def pytest_configure(config):
"""Ensure test directory is in Python path before collection."""
test_dir = Path(__file__).parent
if str(test_dir) not in sys.path:
sys.path.insert(0, str(test_dir))
MOCK_SERVER_PORT = 8090
MOCK_SERVER_URL = f"http://localhost:{MOCK_SERVER_PORT}"
LITELLM_PROXY_PORT = 4000
LITELLM_PROXY_URL = f"http://localhost:{LITELLM_PROXY_PORT}"
DATABASE_URL = "postgresql://llmproxy:dbpassword9090@localhost:5432/litellm"
def kill_process_on_port(port: int) -> None:
"""Kill any process using the specified port."""
try:
result = subprocess.run(
["lsof", "-ti", f":{port}"],
capture_output=True,
text=True,
timeout=5,
)
if result.stdout.strip():
pids = result.stdout.strip().split("\n")
for pid in pids:
try:
subprocess.run(["kill", "-9", pid.strip()], timeout=5)
except Exception:
pass
time.sleep(1)
except Exception:
pass
def wait_for_server(url: str, max_attempts: int = 30, delay: float = 1.0) -> bool:
"""Wait for a server to become available at url/health.
Any HTTP response (including 401) means the server is up.
Only connection errors count as "not ready yet".
"""
for attempt in range(max_attempts):
try:
response = httpx.get(f"{url}/health", timeout=2.0)
return True
except (httpx.ConnectError, httpx.TimeoutException, httpx.NetworkError):
pass
except Exception:
pass
if attempt < max_attempts - 1:
time.sleep(delay)
return False
def _read_log_tail(log_path: Path, max_lines: int = 80) -> str:
"""Read the last N lines of a log file, returning empty string if not found."""
if not log_path.exists():
return "(log file not found)"
try:
text = log_path.read_text()
lines = text.strip().splitlines()
if len(lines) > max_lines:
return f"... ({len(lines) - max_lines} lines truncated) ...\n" + "\n".join(
lines[-max_lines:]
)
return text
except Exception as e:
return f"(error reading log: {e})"
def _check_process_alive(process: subprocess.Popen, label: str, log_path: Path):
"""Check if a subprocess crashed immediately after starting.
Raises pytest.fail with log output if the process has already exited.
"""
time.sleep(1)
exit_code = process.poll()
if exit_code is not None:
log_output = _read_log_tail(log_path)
pytest.fail(
f"{label} exited immediately with code {exit_code}.\n"
f"--- {label} log ({log_path}) ---\n{log_output}\n"
f"--- end log ---"
)
def setup_database() -> bool:
"""Ensure PostgreSQL database exists and is accessible."""
try:
import psycopg2
conn = psycopg2.connect(
host="localhost",
port=5432,
database="litellm",
user="llmproxy",
password="dbpassword9090",
connect_timeout=5,
)
conn.close()
return True
except ImportError:
print("WARNING: psycopg2 not installed — cannot verify database")
return False
except Exception:
return False
@pytest.fixture(scope="session")
def mock_azure_server() -> Generator[str, None, None]:
"""Start mock Azure batch server as a subprocess."""
print(f"\n{'=' * 60}")
print("Setting up Mock Azure Batch Server")
print(f"{'=' * 60}")
kill_process_on_port(MOCK_SERVER_PORT)
runner_script = Path(__file__).parent / "fixtures" / "run_mock_server.py"
runner_script.write_text(
"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
from fixtures.mock_azure_batch_server import create_mock_azure_batch_server
import uvicorn
if __name__ == "__main__":
app = create_mock_azure_batch_server()
uvicorn.run(app, host="0.0.0.0", port=8090, log_level="info", access_log=False)
"""
)
mock_log = LOG_DIR / "mock_server.log"
log_file = open(mock_log, "w")
print(f"Starting mock server on port {MOCK_SERVER_PORT}...")
print(f"Log file: {mock_log}")
process = subprocess.Popen(
[sys.executable, str(runner_script)],
stdout=log_file,
stderr=subprocess.STDOUT,
cwd=Path(__file__).parent,
)
_check_process_alive(process, "Mock server", mock_log)
if not wait_for_server(MOCK_SERVER_URL, max_attempts=30, delay=1.0):
log_output = _read_log_tail(mock_log)
exit_code = process.poll()
process.terminate()
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.kill()
process.wait()
log_file.close()
pytest.fail(
f"Mock server failed to start on port {MOCK_SERVER_PORT} "
f"(process exit_code={exit_code}).\n"
f"--- mock server log ---\n{log_output}\n--- end log ---\n"
f"Hint: ensure 'uvicorn' and 'fastapi' are installed."
)
print(f"Mock Azure server ready at {MOCK_SERVER_URL}")
yield MOCK_SERVER_URL
print("\nShutting down mock server...")
try:
process.terminate()
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.kill()
process.wait()
log_file.close()
print("Mock server stopped")
@pytest.fixture(scope="session")
def litellm_proxy_server(mock_azure_server: str) -> Generator[str, None, None]:
"""Start LiteLLM proxy server for the test session."""
print(f"\n{'=' * 60}")
print("Setting up LiteLLM Proxy Server")
print(f"{'=' * 60}")
if not setup_database():
pytest.skip(
"PostgreSQL database not available at localhost:5432. "
"Start PostgreSQL and create a 'litellm' database:\n"
" docker run -d --name litellm-db -p 5432:5432 "
'-e POSTGRES_USER=llmproxy -e POSTGRES_PASSWORD=dbpassword9090 '
"-e POSTGRES_DB=litellm postgres:15\n"
"Then run: prisma db push --schema=litellm/proxy/schema.prisma"
)
print("Database connection verified")
config_path = Path(__file__).parent / "fixtures" / "config.yml"
if not config_path.exists():
pytest.fail(f"Config file not found: {config_path}")
print("Config file found")
kill_process_on_port(LITELLM_PROXY_PORT)
os.environ["MOCK_SERVER_URL_V1"] = f"{mock_azure_server}/v1"
os.environ["MOCK_SERVER_URL_OPENAI_V1"] = f"{mock_azure_server}/openai/v1"
os.environ["DATABASE_URL"] = DATABASE_URL
os.environ["USE_LOCAL_LITELLM"] = "true"
os.environ["USE_MOCK_MODELS"] = "true"
os.environ["USE_STATE_TRACKER"] = "true"
os.environ["PROXY_BATCH_POLLING_INTERVAL"] = "10"
print("Environment configured")
print(f"Starting LiteLLM proxy on port {LITELLM_PROXY_PORT}...")
litellm_root = Path(__file__).parent.parent.parent
cmd = [
sys.executable,
"-m",
"litellm.proxy.proxy_cli",
"--config",
str(config_path),
"--port",
str(LITELLM_PROXY_PORT),
"--detailed_debug",
]
proxy_log = LOG_DIR / "proxy_server.log"
log_file = open(proxy_log, "w")
print(f"Log file: {proxy_log}")
process = subprocess.Popen(
cmd,
stdout=log_file,
stderr=subprocess.STDOUT,
env=os.environ.copy(),
cwd=litellm_root,
)
_check_process_alive(process, "LiteLLM proxy", proxy_log)
if not wait_for_server(LITELLM_PROXY_URL, max_attempts=60, delay=1.0):
log_output = _read_log_tail(proxy_log)
exit_code = process.poll()
process.terminate()
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.kill()
process.wait()
log_file.close()
pytest.fail(
f"LiteLLM proxy failed to start on port {LITELLM_PROXY_PORT} "
f"(process exit_code={exit_code}).\n"
f"--- proxy log (last 80 lines) ---\n{log_output}\n--- end log ---\n"
f"Hints:\n"
f" 1. Ensure Prisma client is generated: "
f"cd {litellm_root} && prisma generate --schema=litellm/proxy/schema.prisma\n"
f" 2. Ensure DB migrations are applied: "
f"prisma db push --schema=litellm/proxy/schema.prisma\n"
f" 3. Check the full log at: {proxy_log}"
)
print(f"LiteLLM proxy ready at {LITELLM_PROXY_URL}")
yield LITELLM_PROXY_URL
print("\nShutting down LiteLLM proxy...")
try:
process.terminate()
process.wait(timeout=10)
except subprocess.TimeoutExpired:
process.kill()
process.wait()
log_file.close()
print("LiteLLM proxy stopped")
@pytest.fixture(scope="session")
def event_loop():
"""Provide an event loop for async tests."""
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
yield loop
loop.close()

View file

@ -1,56 +0,0 @@
model_list:
- model_name: openai-fake-gpt-3.5-turbo
litellm_params:
model: openai/openai-fake-gpt-3.5-turbo
api_base: os.environ/MOCK_SERVER_URL_V1
api_key: fake-key
- model_name: openai-fake-gpt-4
litellm_params:
model: openai/openai-fake-gpt-4
api_base: os.environ/MOCK_SERVER_URL_V1
api_key: fake-key
- model_name: openai-fake-gpt-4o
litellm_params:
model: openai/openai-fake-gpt-4o
api_base: os.environ/MOCK_SERVER_URL_V1
api_key: fake-key
- model_name: fake-text-embedding-3-small
litellm_params:
model: openai/fake-text-embedding-3-small
api_base: os.environ/MOCK_SERVER_URL_V1
api_key: fake-key
- model_name: o3-mini-batch-2025-01-31
litellm_params:
model: openai/o3-mini-batch-2025-01-31
api_base: os.environ/MOCK_SERVER_URL_OPENAI_V1
api_key: fake-key
model_info:
mode: batch
- model_name: azure-fake-gpt-5-batch-2025-08-07
litellm_params:
api_base: http://0.0.0.0:8090
api_key: fake-key
api_version: 2025-03-01-preview
base_model: azure/gpt-5
model: azure/gpt-5-mini
custom_llm_provider: azure
general_settings:
master_key: sk-1234
database_url: os.environ/DATABASE_URL
proxy_batch_polling_interval: 10
litellm_settings:
drop_params: true
set_verbose: true
json_logs: true
# S3 callback for batch completion logging (points to mock server)
callbacks: ["s3_v2"]
s3_callback_params:
s3_bucket_name: litellm-test-bucket
s3_region_name: us-east-1
s3_endpoint_url: http://0.0.0.0:8090
s3_aws_access_key_id: fake-key
s3_aws_secret_access_key: fake-secret
s3_use_ssl: false
s3_verify: false

View file

@ -1,3 +0,0 @@
from .server import create_mock_azure_batch_server
__all__ = ["create_mock_azure_batch_server"]

View file

@ -1,517 +0,0 @@
import asyncio
import io
import json
import logging
import time
import uuid
from typing import Dict, List, Optional
from fastapi import FastAPI, HTTPException, Query, Request, UploadFile
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class FileObject(BaseModel):
id: str
object: str = "file"
bytes: int
created_at: int
filename: str
purpose: str
status: str = "processed"
status_details: Optional[str] = None
expires_at: Optional[int] = None
class BatchObject(BaseModel):
id: str
object: str = "batch"
endpoint: str
errors: Optional[Dict] = None
input_file_id: str
completion_window: str
status: str
output_file_id: Optional[str] = None
error_file_id: Optional[str] = None
created_at: int
in_progress_at: Optional[int] = None
expires_at: Optional[int] = None
finalizing_at: Optional[int] = None
completed_at: Optional[int] = None
failed_at: Optional[int] = None
expired_at: Optional[int] = None
cancelling_at: Optional[int] = None
cancelled_at: Optional[int] = None
request_counts: Optional[Dict[str, int]] = None
metadata: Optional[Dict] = None
class BatchListResponse(BaseModel):
object: str = "list"
data: List[Dict]
first_id: Optional[str] = None
last_id: Optional[str] = None
has_more: bool = False
file_storage: Dict[str, Dict] = {}
batch_storage: Dict[str, BatchObject] = {}
batch_results: Dict[str, List[Dict]] = {}
PROCESSING_DELAY_SECONDS = float(1)
VALIDATING_DELAY_SECONDS = float(3)
async def process_batch(batch_id: str):
logger.info(f"Starting batch processing for {batch_id}")
try:
batch = batch_storage[batch_id]
await asyncio.sleep(VALIDATING_DELAY_SECONDS)
batch.status = "in_progress"
batch.in_progress_at = int(time.time())
logger.info(f"Batch {batch_id} status: in_progress")
await process_batch_requests(batch_id)
await asyncio.sleep(PROCESSING_DELAY_SECONDS)
batch.status = "finalizing"
batch.finalizing_at = int(time.time())
logger.info(f"Batch {batch_id} status: finalizing")
await asyncio.sleep(PROCESSING_DELAY_SECONDS)
await create_output_file(batch_id)
batch.status = "completed"
batch.completed_at = int(time.time())
logger.info(f"Batch {batch_id} status: completed")
except Exception as e:
logger.error(f"Batch {batch_id} failed: {e}")
batch = batch_storage[batch_id]
batch.status = "failed"
batch.failed_at = int(time.time())
batch.errors = {
"object": "list",
"data": [{"code": "processing_error", "message": str(e)}],
}
async def process_batch_requests(batch_id: str):
batch = batch_storage[batch_id]
input_file = file_storage[batch.input_file_id]
requests = []
for line in input_file["content"].split("\n"):
if line.strip():
try:
requests.append(json.loads(line))
except json.JSONDecodeError as e:
logger.warning(f"Invalid JSON line in batch {batch_id}: {e}")
logger.info(f"Batch {batch_id} has {len(requests)} requests")
results = []
failed_count = 0
for req in requests:
result = await process_single_request(req)
if result.get("error"):
failed_count += 1
results.append(result)
batch_results[batch_id] = results
batch.request_counts = {
"total": len(requests),
"completed": len(results) - failed_count,
"failed": failed_count,
}
async def process_single_request(request_data: Dict) -> Dict:
custom_id = request_data.get("custom_id")
url = request_data.get("url", "/v1/chat/completions")
body = request_data.get("body", {})
if "/chat/completions" in url:
response_body = {
"id": f"chatcmpl-{uuid.uuid4().hex}",
"object": "chat.completion",
"created": int(time.time()),
"model": body.get("model", "gpt-4o"),
"choices": [
{
"index": 0,
"message": {"role": "assistant", "content": "Mock batch response."},
"finish_reason": "stop",
},
],
"usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
}
status_code = 200
else:
response_body = {"error": {"message": f"Unsupported endpoint: {url}"}}
status_code = 400
return {
"id": f"batch_req_{uuid.uuid4().hex[:12]}",
"custom_id": custom_id,
"response": {
"status_code": status_code,
"request_id": f"req_{uuid.uuid4().hex[:12]}",
"body": response_body,
},
"error": None,
}
async def create_output_file(batch_id: str):
results = batch_results.get(batch_id, [])
output_lines = [json.dumps(result) for result in results]
output_content = "\n".join(output_lines)
output_file_id = f"file-batch-output-{uuid.uuid4().hex[:12]}"
file_storage[output_file_id] = {
"content": output_content,
"filename": f"batch_output_{batch_id}.jsonl",
"purpose": "batch_output",
"bytes": len(output_content.encode()),
"created_at": int(time.time()),
}
batch = batch_storage[batch_id]
batch.output_file_id = output_file_id
logger.info(f"Created output file {output_file_id} for batch {batch_id}")
def validate_batch_input(content: str) -> tuple[bool, str, List[Dict]]:
requests = []
custom_ids = set()
lines = content.strip().split("\n")
if not lines or all(not line.strip() for line in lines):
return False, "empty_batch", []
for line_num, line in enumerate(lines, 1):
if not line.strip():
continue
try:
req = json.loads(line)
except json.JSONDecodeError:
return False, "invalid_json_line", []
for field in ["custom_id", "method", "url", "body"]:
if field not in req:
return False, "invalid_request", []
if req["custom_id"] in custom_ids:
return False, "duplicate_custom_id", []
custom_ids.add(req["custom_id"])
requests.append(req)
if len(requests) > 100000:
return False, "too_many_tasks", []
return True, "", requests
def setup_batch_routes(app: FastAPI):
# Files endpoints (OpenAI and Azure paths)
@app.post("/openai/v1/files")
@app.post("/openai/files")
@app.post("/v1/files")
@app.post("/files")
async def create_file(request: Request):
form = await request.form()
logger.info(f"File upload form fields: {list(form.keys())}")
file: UploadFile = form.get("file")
purpose: str = form.get("purpose", "batch")
if not file:
raise HTTPException(status_code=400, detail="No file provided")
logger.info(f"Uploading file: {file.filename}, purpose: {purpose}")
content = await file.read()
content_str = content.decode("utf-8")
file_id = f"file-{uuid.uuid4().hex[:24]}"
created_at = int(time.time())
expires_at = None
expires_after_seconds = form.get("expires_after[seconds]")
if expires_after_seconds:
try:
seconds = int(expires_after_seconds)
logger.info(f"expires_after[seconds] = {seconds}")
if seconds < 259200 or seconds > 2592000:
raise HTTPException(
status_code=400,
detail={
"error": {
"code": "invalidPayload",
"message": "Value for Seconds must be between 259200 and 2592000.",
},
},
)
expires_at = created_at + seconds
logger.info(f"Calculated expires_at: {expires_at}")
except ValueError as e:
logger.warning(f"Failed to parse expires_after[seconds]: {e}")
file_storage[file_id] = {
"content": content_str,
"filename": file.filename or "batch_input.jsonl",
"purpose": purpose,
"bytes": len(content),
"created_at": created_at,
"expires_at": expires_at,
}
logger.info(f"Created file {file_id}, expires_at={expires_at}")
return FileObject(
id=file_id,
bytes=len(content),
created_at=created_at,
filename=file.filename or "batch_input.jsonl",
purpose=purpose,
expires_at=expires_at,
).model_dump()
@app.get("/openai/v1/files/{file_id}")
@app.get("/openai/files/{file_id}")
@app.get("/v1/files/{file_id}")
@app.get("/files/{file_id}")
async def get_file(file_id: str):
logger.info(f"Getting file: {file_id}")
if file_id not in file_storage:
raise HTTPException(status_code=404, detail="File not found")
file_data = file_storage[file_id]
return FileObject(
id=file_id,
bytes=file_data["bytes"],
created_at=file_data["created_at"],
filename=file_data["filename"],
purpose=file_data["purpose"],
expires_at=file_data.get("expires_at"),
).model_dump()
@app.get("/openai/v1/files/{file_id}/content")
@app.get("/openai/files/{file_id}/content")
@app.get("/v1/files/{file_id}/content")
@app.get("/files/{file_id}/content")
async def get_file_content(file_id: str):
logger.info(f"Getting file content: {file_id}")
if file_id not in file_storage:
raise HTTPException(status_code=404, detail="File not found")
file_data = file_storage[file_id]
content = file_data["content"]
return StreamingResponse(
io.StringIO(content),
media_type="application/octet-stream",
headers={
"Content-Disposition": f"attachment; filename={file_data['filename']}",
},
)
@app.delete("/openai/v1/files/{file_id}")
@app.delete("/openai/files/{file_id}")
@app.delete("/v1/files/{file_id}")
@app.delete("/files/{file_id}")
async def delete_file(file_id: str):
logger.info(f"Deleting file: {file_id}")
if file_id not in file_storage:
raise HTTPException(status_code=404, detail="File not found")
del file_storage[file_id]
return {"id": file_id, "object": "file", "deleted": True}
@app.get("/openai/v1/files")
@app.get("/openai/files")
@app.get("/v1/files")
@app.get("/files")
async def list_files(
purpose: Optional[str] = None,
limit: int = Query(10000, le=10000),
):
logger.info(f"Listing files, purpose: {purpose}, limit: {limit}")
files = []
for file_id, file_data in file_storage.items():
if purpose is None or file_data.get("purpose") == purpose:
files.append(
FileObject(
id=file_id,
bytes=file_data["bytes"],
created_at=file_data["created_at"],
filename=file_data["filename"],
purpose=file_data["purpose"],
expires_at=file_data.get("expires_at"),
).model_dump(),
)
return {"object": "list", "data": files[:limit]}
# Batches endpoints (OpenAI and Azure paths)
@app.post("/openai/v1/batches")
@app.post("/openai/batches")
@app.post("/v1/batches")
@app.post("/batches")
async def create_batch(request_data: dict):
input_file_id = request_data.get("input_file_id")
endpoint = request_data.get("endpoint", "/v1/chat/completions")
completion_window = request_data.get("completion_window", "24h")
metadata = request_data.get("metadata", {})
output_expires_after = request_data.get("output_expires_after")
logger.info(
f"Creating batch with input_file: {input_file_id}, endpoint: {endpoint}, output_expires_after: {output_expires_after}",
)
if not input_file_id or input_file_id not in file_storage:
raise HTTPException(status_code=400, detail="Input file not found")
input_file = file_storage[input_file_id]
is_valid, error_code, _ = validate_batch_input(input_file["content"])
if not is_valid:
raise HTTPException(
status_code=400,
detail={
"error": {
"code": error_code,
"message": f"Validation failed: {error_code}",
},
},
)
batch_id = f"batch_{uuid.uuid4()}"
created_at = int(time.time())
if output_expires_after:
seconds = (
output_expires_after.get("seconds", 0)
if isinstance(output_expires_after, dict)
else 0
)
expires_at = created_at + seconds
logger.info(
f"Using output_expires_after: {seconds}s, expires_at: {expires_at}",
)
elif completion_window == "24h":
expires_at = created_at + (24 * 60 * 60)
else:
expires_at = created_at + (24 * 60 * 60)
batch = BatchObject(
id=batch_id,
endpoint=endpoint,
input_file_id=input_file_id,
completion_window=completion_window,
status="validating",
created_at=created_at,
expires_at=expires_at,
request_counts={"total": 0, "completed": 0, "failed": 0},
metadata=metadata,
)
batch_storage[batch_id] = batch
logger.info(f"Created batch {batch_id}")
asyncio.create_task(process_batch(batch_id))
return batch.model_dump()
@app.get("/openai/v1/batches/{batch_id}")
@app.get("/openai/batches/{batch_id}")
@app.get("/v1/batches/{batch_id}")
@app.get("/batches/{batch_id}")
async def get_batch(batch_id: str):
logger.info(f"Getting batch: {batch_id}")
if batch_id not in batch_storage:
raise HTTPException(status_code=404, detail="Batch not found")
return batch_storage[batch_id].model_dump()
@app.get("/openai/v1/batches")
@app.get("/openai/batches")
@app.get("/v1/batches")
@app.get("/batches")
async def list_batches(
after: Optional[str] = Query(None),
limit: int = Query(20, le=100),
):
logger.info(f"Listing batches, after: {after}, limit: {limit}")
batches = list(batch_storage.values())
batches.sort(key=lambda x: x.created_at, reverse=True)
if after:
after_index = next((i for i, b in enumerate(batches) if b.id == after), -1)
if after_index >= 0:
batches = batches[after_index + 1 :]
batches = batches[:limit]
return BatchListResponse(
data=[batch.model_dump() for batch in batches],
first_id=batches[0].id if batches else None,
last_id=batches[-1].id if batches else None,
has_more=len(batches) == limit,
).model_dump()
@app.post("/openai/v1/batches/{batch_id}/cancel")
@app.post("/openai/batches/{batch_id}/cancel")
@app.post("/v1/batches/{batch_id}/cancel")
@app.post("/batches/{batch_id}/cancel")
async def cancel_batch(batch_id: str):
logger.info(f"Cancelling batch: {batch_id}")
if batch_id not in batch_storage:
raise HTTPException(status_code=404, detail="Batch not found")
batch = batch_storage[batch_id]
if batch.status in ["completed", "failed", "cancelled", "expired"]:
raise HTTPException(
status_code=400,
detail=f"Cannot cancel batch in {batch.status} status",
)
batch.status = "cancelled"
batch.cancelled_at = int(time.time())
logger.info(f"Batch {batch_id} cancelled")
return batch.model_dump()
# Debug endpoints
@app.get("/debug/batches")
async def debug_list_batches():
return {
"batches": {
batch_id: batch.model_dump()
for batch_id, batch in batch_storage.items()
},
"files": {
file_id: {k: v for k, v in data.items() if k != "content"}
for file_id, data in file_storage.items()
},
}
@app.post("/reset")
@app.post("/debug/clear")
async def reset_all():
file_storage.clear()
batch_storage.clear()
batch_results.clear()
logger.info("All data cleared")
return {"message": "All data cleared"}
@app.get("/debug/status")
async def debug_status():
return {
"files_count": len(file_storage),
"batches_count": len(batch_storage),
"batch_statuses": {bid: b.status for bid, b in batch_storage.items()},
}

View file

@ -1,124 +0,0 @@
import json
import time
import uuid
from datetime import datetime
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
def get_request_details(request: Request, body: dict = None) -> str:
details = {
"method": request.method,
"url": str(request.url),
"path": request.url.path,
"headers": dict(request.headers),
"query_params": dict(request.query_params),
}
return json.dumps(details, indent=2)
def data_generator(response_details: str, model: str):
response_id = uuid.uuid4().hex
content = response_details
chunk_size = 50
for i in range(0, len(content), chunk_size):
text_chunk = content[i : i + chunk_size]
chunk = {
"id": f"chatcmpl-{response_id}",
"object": "chat.completion.chunk",
"created": int(time.time()),
"model": model,
"choices": [{"index": 0, "delta": {"content": text_chunk}}],
}
yield f"data: {json.dumps(chunk)}\n\n"
final_chunk = {
"id": f"chatcmpl-{response_id}",
"object": "chat.completion.chunk",
"created": int(time.time()),
"model": model,
"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
}
yield f"data: {json.dumps(final_chunk)}\n\n"
yield "data: [DONE]\n\n"
def setup_chat_routes(app: FastAPI):
@app.post("/chat/completions")
@app.post("/v1/chat/completions")
@app.post("/openai/deployments/{model:path}/chat/completions")
async def completion(request: Request):
data = await request.json()
model = data.get("model", "unknown")
request_details = get_request_details(request, data)
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
response_details = f"Request:{request_details}, Canned Response:{timestamp}"
if data.get("stream"):
return StreamingResponse(
content=data_generator(response_details, model),
media_type="text/event-stream",
)
else:
response_id = uuid.uuid4().hex
response = {
"id": f"chatcmpl-{response_id}",
"object": "chat.completion",
"created": int(time.time()),
"model": model,
"system_fingerprint": "fp_mock_server",
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": response_details,
},
"logprobs": None,
"finish_reason": "stop",
},
],
"usage": {
"prompt_tokens": 9,
"completion_tokens": 12,
"total_tokens": 21,
},
}
return response
@app.post("/completions")
@app.post("/v1/completions")
async def text_completion(request: Request):
data = await request.json()
model = data.get("model", "unknown")
request_details = get_request_details(request, data)
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
response_details = f"Request:{request_details}, Canned Response:{timestamp}"
if data.get("stream"):
return StreamingResponse(
content=data_generator(response_details, model),
media_type="text/event-stream",
)
else:
response = {
"id": f"cmpl-{uuid.uuid4().hex}",
"choices": [
{
"finish_reason": "stop",
"index": 0,
"logprobs": None,
"text": response_details,
},
],
"created": int(time.time()),
"model": model,
"object": "text_completion",
"system_fingerprint": None,
"usage": {
"completion_tokens": 16,
"prompt_tokens": 10,
"total_tokens": 26,
},
}
return response

View file

@ -1,23 +0,0 @@
from fastapi import FastAPI, Request
def setup_embeddings_routes(app: FastAPI):
@app.post("/embeddings")
@app.post("/v1/embeddings")
@app.post("/openai/deployments/{model:path}/embeddings")
async def embeddings(request: Request):
data = await request.json()
model = data.get("model", "unknown")
_small_embedding = [
-0.006929283495992422,
-0.005336422007530928,
-4.547132266452536e-05,
-0.024047505110502243,
]
big_embedding = _small_embedding * 100
return {
"object": "list",
"data": [{"object": "embedding", "index": 0, "embedding": big_embedding}],
"model": model,
"usage": {"prompt_tokens": 5, "total_tokens": 5},
}

View file

@ -1,170 +0,0 @@
import json
import re
import time
import uuid
from datetime import datetime
from typing import Any
from fastapi import FastAPI, Request, HTTPException
# Header to identify which model/deployment this request targets (simulates Azure model-specific encryption).
# When set, the mock validates that encrypted_content in input was produced by this model.
MOCK_AZURE_MODEL_HEADER = "X-Mock-Azure-Model"
# Prefix we use in mock encrypted_content: gAAA_model_<model_id>_<32hex uuid>
# Model id can contain underscores (e.g. gpt-5.1-codex-openai-2).
ENCRYPTED_CONTENT_MODEL_PREFIX = re.compile(r"^gAAA_model_(.+)_[0-9a-f]{32}$")
def _extract_model_from_encrypted_content(encrypted: str) -> str | None:
"""Extract model id from our mock encrypted_content format, or None if not our format."""
if not isinstance(encrypted, str) or not encrypted.startswith("gAAA"):
return None
m = ENCRYPTED_CONTENT_MODEL_PREFIX.match(encrypted)
return m.group(1) if m else None
def _collect_encrypted_contents(obj, out: list[str]) -> None:
"""Recursively collect all encrypted_content string values from input structure."""
if isinstance(obj, dict):
if "encrypted_content" in obj and obj["encrypted_content"]:
out.append(obj["encrypted_content"])
for v in obj.values():
_collect_encrypted_contents(v, out)
elif isinstance(obj, list):
for item in obj:
_collect_encrypted_contents(item, out)
def _validate_encrypted_content_model(request_model: str | None, input_data: Any) -> str | None:
"""
If request_model is set, check that all encrypted_content in input was produced by this model.
Returns error message if validation fails, else None.
Content with our format (gAAA_model_<id>_) must match request_model.
"""
if not request_model:
return None
encrypted_values: list[str] = []
_collect_encrypted_contents(input_data, encrypted_values)
for enc in encrypted_values:
content_model = _extract_model_from_encrypted_content(enc)
if content_model is not None and content_model != request_model:
err = enc[:50] + "..." if len(enc) > 50 else enc
return f"The encrypted content {err} could not be verified."
return None
def get_request_details(request: Request, body: dict = None) -> str:
details = {
"method": request.method,
"url": str(request.url),
"path": request.url.path,
"headers": dict(request.headers),
"query_params": dict(request.query_params),
}
return json.dumps(details, indent=2)
def setup_responses_routes(app: FastAPI):
@app.post("/responses")
@app.post("/v1/responses")
@app.post("/openai/responses")
async def responses_api(request: Request):
data = await request.json()
model = data.get("model", "unknown")
# Simulate Azure: encrypted content from one model cannot be verified by another.
input_data = data.get("input")
err_msg = _validate_encrypted_content_model(model, input_data)
if err_msg is not None:
raise HTTPException(
status_code=400,
detail={
"error": {
"message": err_msg,
"type": "invalid_request_error",
"param": None,
"code": "invalid_encrypted_content",
}
},
)
request_details = get_request_details(request, data)
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
response_details = f"Request:{request_details}, Canned Response:{timestamp}"
response_id = uuid.uuid4().hex
message_id = f"msg_{uuid.uuid4().hex[:34]}"
reasoning_id = f"rs_{uuid.uuid4().hex[:34]}"
output_items: list[dict[str, Any]] = [
{
"id": message_id,
"content": [
{
"annotations": [],
"text": response_details,
"type": "output_text",
"logprobs": [],
},
],
"role": "assistant",
"status": "completed",
"type": "message",
},
]
if model:
output_items.append(
{
"id": reasoning_id,
"type": "reasoning",
"status": "completed",
"encrypted_content": f"gAAA_model_{model}_{uuid.uuid4().hex}",
}
)
return {
"id": f"resp_{response_id}",
"created_at": int(time.time()),
"error": None,
"incomplete_details": None,
"instructions": None,
"metadata": {},
"model": model,
"object": "response",
"output": output_items,
"parallel_tool_calls": True,
"temperature": data.get("temperature", 1.0),
"tool_choice": data.get("tool_choice", "auto"),
"tools": data.get("tools", []),
"top_p": data.get("top_p", 1.0),
"max_output_tokens": data.get("max_output_tokens"),
"previous_response_id": None,
"reasoning": {"effort": None, "summary": None},
"status": "completed",
"text": {"format": {"type": "text"}, "verbosity": "medium"},
"truncation": "disabled",
"usage": {
"input_tokens": 11,
"input_tokens_details": {
"audio_tokens": None,
"cached_tokens": 0,
"text_tokens": None,
},
"output_tokens": 19,
"output_tokens_details": {"reasoning_tokens": 0, "text_tokens": None},
"total_tokens": 30,
"cost": None,
},
"user": None,
"store": True,
"background": False,
"content_filters": None,
"max_tool_calls": None,
"prompt_cache_key": None,
"safety_identifier": None,
"service_tier": "default",
"top_logprobs": 0,
}

View file

@ -1,98 +0,0 @@
"""
Mock S3 callback receiver for testing LiteLLM S3 callbacks.
This module provides S3-compatible endpoints that capture callback data
sent by LiteLLM's s3_v2 callback handler after batch completion.
"""
import json
import logging
import time
from typing import Any, Dict, List, Optional
from fastapi import FastAPI, Request
from pydantic import BaseModel
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class S3CallbackRecord(BaseModel):
key: str
bucket: str
content: Dict[str, Any]
timestamp: int
content_type: Optional[str] = None
callback_storage: List[S3CallbackRecord] = []
def setup_s3_callback_routes(app: FastAPI):
@app.put("/{bucket}/{key:path}")
async def s3_put_object(bucket: str, key: str, request: Request):
content_type = request.headers.get("content-type", "application/json")
body = await request.body()
try:
content = json.loads(body.decode("utf-8"))
except (json.JSONDecodeError, UnicodeDecodeError):
content = {"raw": body.decode("utf-8", errors="replace")}
record = S3CallbackRecord(
key=key,
bucket=bucket,
content=content,
timestamp=int(time.time()),
content_type=content_type,
)
callback_storage.append(record)
logger.info(f"S3 callback received: bucket={bucket}, key={key}")
logger.debug(f"Callback content: {json.dumps(content, indent=2)[:500]}")
return {
"ETag": f'"{hash(body)}"',
"VersionId": None,
}
@app.get("/mock-s3/callbacks")
async def list_callbacks(
bucket: Optional[str] = None,
key_prefix: Optional[str] = None,
limit: int = 100,
):
results = callback_storage
if bucket:
results = [r for r in results if r.bucket == bucket]
if key_prefix:
results = [r for r in results if r.key.startswith(key_prefix)]
return {
"count": len(results),
"callbacks": [r.model_dump() for r in results[-limit:]],
}
@app.get("/mock-s3/callbacks/count")
async def count_callbacks(bucket: Optional[str] = None):
if bucket:
count = sum(1 for r in callback_storage if r.bucket == bucket)
else:
count = len(callback_storage)
return {"count": count}
@app.get("/mock-s3/callbacks/latest")
async def get_latest_callback():
if not callback_storage:
return {"callback": None}
return {"callback": callback_storage[-1].model_dump()}
@app.delete("/mock-s3/callbacks")
async def clear_callbacks():
count = len(callback_storage)
callback_storage.clear()
logger.info(f"Cleared {count} S3 callbacks")
return {"cleared": count}

View file

@ -1,33 +0,0 @@
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from .mock_azure_batch import setup_batch_routes
from .mock_chat import setup_chat_routes
from .mock_embeddings import setup_embeddings_routes
from .mock_responses import setup_responses_routes
from .mock_s3_callback import setup_s3_callback_routes
def create_mock_azure_batch_server() -> FastAPI:
"""Create a FastAPI app that mocks Azure Batch API and S3 callbacks."""
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/health")
async def health():
return {"status": "ok"}
setup_chat_routes(app)
setup_responses_routes(app)
setup_embeddings_routes(app)
setup_batch_routes(app)
setup_s3_callback_routes(app)
return app

View file

@ -1,12 +0,0 @@
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
from fixtures.mock_azure_batch_server import create_mock_azure_batch_server
import uvicorn
if __name__ == "__main__":
app = create_mock_azure_batch_server()
uvicorn.run(app, host="0.0.0.0", port=8090, log_level="info", access_log=False)

View file

@ -1,41 +0,0 @@
"""
Smoke test to verify fixtures start and stop correctly.
Run this first to ensure the infrastructure works before running full E2E tests.
"""
import httpx
import pytest
pytestmark = pytest.mark.usefixtures("mock_azure_server", "litellm_proxy_server")
def test_mock_server_health(mock_azure_server):
"""Verify mock Azure server is running and healthy."""
response = httpx.get(f"{mock_azure_server}/health", timeout=5.0)
assert response.status_code == 200
assert response.json() == {"status": "ok"}
print(f"✓ Mock Azure server is healthy at {mock_azure_server}")
def test_litellm_proxy_health(litellm_proxy_server):
"""Verify LiteLLM proxy is running and healthy."""
response = httpx.get(f"{litellm_proxy_server}/health", timeout=5.0)
assert response.status_code == 200
print(f"✓ LiteLLM proxy is healthy at {litellm_proxy_server}")
def test_litellm_proxy_model_list(litellm_proxy_server):
"""Verify LiteLLM proxy can list models."""
response = httpx.get(
f"{litellm_proxy_server}/v1/models",
headers={"Authorization": "Bearer sk-1234"},
timeout=5.0,
)
assert response.status_code == 200
data = response.json()
assert "data" in data
models = [m["id"] for m in data["data"]]
print(f"✓ LiteLLM proxy has {len(models)} models configured")
assert "azure-fake-gpt-5-batch-2025-08-07" in models
print(f"✓ Azure batch model is configured")

File diff suppressed because it is too large Load diff

View file

@ -1,324 +0,0 @@
import base64
import os
import sys
import time
import warnings
import httpx
import openai
import pytest
from tenacity import RetryError
sys.path.insert(0, os.path.abspath("../.."))
from base_integration_test import (
get_mock_server_base_url,
model_id,
use_mock_models,
UserKeyTestMixin,
)
from test_managed_files_base import (
ManagedFilesBase,
MIN_EXPIRY_SECONDS,
get_batch_model_names,
)
MANAGED_FILE_ID_PREFIX = "litellm_proxy"
pytestmark = [
pytest.mark.usefixtures("mock_azure_server", "litellm_proxy_server"),
pytest.mark.skipif(
os.environ.get("SKIP_E2E_TESTS", "false").lower() == "true",
reason="E2E tests disabled via SKIP_E2E_TESTS env var"
),
]
def is_managed_id(file_id: str) -> bool:
"""Check if a file ID is a base64-encoded LiteLLM managed/unified ID."""
try:
padded = file_id + "=" * (-len(file_id) % 4)
decoded = base64.urlsafe_b64decode(padded).decode()
return decoded.startswith(MANAGED_FILE_ID_PREFIX)
except Exception:
return False
def assert_managed_id(file_id: str, label: str):
assert is_managed_id(file_id), f"{label} should be a managed ID, got raw: {file_id}"
def wip_features_enabled() -> bool:
return os.environ.get("WIP_FEATURES", "").lower() == "true"
class TestManagedFilesAPI(ManagedFilesBase, UserKeyTestMixin):
@classmethod
def setup_class(cls):
super().setup_class()
cls.setup_admin_client()
@classmethod
def teardown_class(cls):
cls.teardown_admin_client()
@pytest.fixture(autouse=True)
def setup_test(self):
print(
f"\nBase URL: {self.base_url}, Using mock models: {use_mock_models()}",
)
self.clear_s3_callbacks()
user_id, api_key, user_email, client = self.create_user_key_and_client(
"e2e-batch",
)
self.test_user_id = user_id
self.openai_client = client
print(f"Using user {user_email} (id={user_id})")
def _create_and_verify_batch_input_file(self, tmp_path, model_name):
request_file = self.create_batch_request_file_on_disk(tmp_path, model_name)
print("Creating batch input file...")
batch_input_file = self.create_batch_input_file(
self.openai_client,
request_file,
MIN_EXPIRY_SECONDS,
target_model_names=model_name,
)
print(f"Created batch input file: {self.shorten_id(batch_input_file.id)}")
assert_managed_id(batch_input_file.id, "batch_input_file.id")
print("Retrieving batch input file metadata...")
metadata = self.openai_client.files.retrieve(batch_input_file.id)
assert_managed_id(metadata.id, "files.retrieve(input).id")
assert metadata.id == batch_input_file.id, (
f"Input file ID mismatch: retrieve returned '{metadata.id}' but expected '{batch_input_file.id}'"
)
assert metadata.object == "file"
assert metadata.bytes > 0, "bytes not set"
assert metadata.filename == "modified_file.jsonl"
assert metadata.purpose == "batch"
assert metadata.status in ["uploaded", "processed", "error"]
assert metadata.created_at > 0
if wip_features_enabled():
assert metadata.expires_at > 0, "expires_at not set"
self.print_file_metadata(metadata, "Input file")
return batch_input_file
def _create_and_verify_batch(self, input_file_id):
print("\nCreating batch...")
batch = self.create_batch(
self.openai_client,
input_file_id,
MIN_EXPIRY_SECONDS,
)
print(f"Created batch: {self.shorten_id(batch.id)}")
assert batch.id, "No batch ID returned"
assert_managed_id(batch.id, "batch.id")
assert_managed_id(batch.input_file_id, "batch.input_file_id")
assert batch.input_file_id == input_file_id, "batch.input_file_id mismatch"
assert batch.status in ["validating", "in_progress", "finalizing", "completed"]
if not batch.expires_at:
warnings.warn("batch expires_at not set")
else:
assert batch.expires_at > 0
if not batch.endpoint:
warnings.warn("batch.endpoint empty - Azure API quirk, not a bug")
else:
assert batch.endpoint == "/v1/chat/completions"
assert batch.completion_window == "24h"
assert batch.created_at > 0
self.print_batch_metadata(batch)
return batch
def _list_batches(self, batch_id, model_name):
if not wip_features_enabled():
return
print("\nListing batches...")
try:
batches_list = self.wait_for_batch_list(
model_name,
max_seconds=30,
wait_seconds=5,
)
batch_ids = [b.id for b in (batches_list.data if batches_list else [])]
if batch_id not in batch_ids:
warnings.warn(
f"Batch {batch_id} not found in list. "
f"batches.list returns raw IDs, not encoded IDs. raw IDs: {batch_ids}",
)
except openai.APIError as e:
pytest.fail(f"batches.list() failed: {e}")
def _wait_for_batch_completion(self, batch_id, tracker):
print(f"\nWaiting for batch {self.shorten_id(batch_id)} to complete...")
try:
batch_response = self.wait_for_batch_state(
self.openai_client,
batch_id,
"completed",
max_seconds=25 * 60,
wait_seconds=15,
state_tracker=tracker,
)
except RetryError:
tracker.print_state("Timeout waiting for batch completion")
raise TimeoutError("Timed out waiting for batch to be in state: completed")
assert_managed_id(batch_response.id, "batch_response.id")
assert batch_response.id == batch_id, (
f"batch_response.id mismatch: got '{batch_response.id}' but expected '{batch_id}'"
)
assert_managed_id(batch_response.input_file_id, "batch_response.input_file_id")
assert_managed_id(
batch_response.output_file_id,
"batch_response.output_file_id",
)
return batch_response
def _get_and_verify_batch_output(self, output_file_id):
print("\nRetrieving batch output file metadata...")
metadata = self.openai_client.files.retrieve(output_file_id)
assert_managed_id(metadata.id, "files.retrieve(output_file_id).id")
assert metadata.id == output_file_id, (
f"Output file ID mismatch: retrieve returned '{metadata.id}' but expected '{output_file_id}'"
)
assert metadata.object == "file"
assert metadata.bytes > 0, "bytes not set"
assert metadata.filename, "filename not set"
assert metadata.purpose in ["batch_output", "batch"]
assert metadata.created_at > 0
self.print_file_metadata(metadata, "Output file")
print("\nFetching batch output file content...")
content = self.openai_client.files.content(output_file_id)
assert content.text, "No batch file content returned"
assert len(content.text) > 0, "Batch file content is empty"
print(f"Output file content ({len(content.text)} bytes):")
for line in content.text.strip().split("\n")[:3]:
print(f"\t{line}")
return metadata
def _delete_file(self, file_id, label, max_retries=10, retry_delay=5):
print(f"\nDeleting {label}: {self.shorten_id(file_id)}")
for attempt in range(max_retries):
try:
self.openai_client.files.delete(file_id)
return
except openai.BadRequestError as e:
if "batch_processed" in str(e) and attempt < max_retries - 1:
print(
f" File still referenced by unprocessed batch, "
f"retrying in {retry_delay}s ({attempt + 1}/{max_retries})"
)
time.sleep(retry_delay)
else:
pytest.fail(f"files.delete({label}) failed: {e}")
except openai.APIError as e:
pytest.fail(f"files.delete({label}) failed: {e}")
def _verify_file_deleted(self, file_id, label):
print(f"Verifying {label} is deleted...")
try:
self.openai_client.files.content(file_id)
assert False, f"{label} {file_id} still accessible after deletion"
except openai.NotFoundError:
print(f"{label} correctly not accessible after deletion")
# ------------------------------------------------------------------
# Tests
# ------------------------------------------------------------------
@pytest.mark.flaky(reruns=2)
@pytest.mark.parametrize(
"model_name",
get_batch_model_names(),
ids=model_id,
)
def test_e2e_managed_batch(self, tmp_path, model_name):
print(
f"\n\nStarting test with base_url={self.base_url} and model_name={model_name}\n",
)
self.reset_mock_server()
tracker = self.create_state_tracker()
batch_input_file = self._create_and_verify_batch_input_file(
tmp_path,
model_name,
)
tracker.set_file_id(batch_input_file.id)
tracker.print_state("After creating batch input file")
batch = self._create_and_verify_batch(batch_input_file.id)
tracker.set_batch_id(batch.id)
tracker.print_state("After creating batch")
self._list_batches(batch.id, model_name)
batch_response = self._wait_for_batch_completion(batch.id, tracker)
tracker.print_state("After batch completed")
self._get_and_verify_batch_output(batch_response.output_file_id)
tracker.print_state("After retrieving output file")
tracker.print_state("Final state after cleanup")
tracker.wait_and_print_s3_callbacks()
tracker.assert_batch_cost_callback()
self._delete_file(batch_input_file.id, "input file")
self._delete_file(batch_response.output_file_id, "output file")
self._verify_file_deleted(batch_input_file.id, "input file")
self._verify_file_deleted(batch_response.output_file_id, "output file")
def cleanup_batches_in_database(self):
import psycopg2
print("Cleaning up stale batch records from database...")
try:
conn = psycopg2.connect(
host="localhost",
port=5432,
database="litellm",
user="llmproxy",
password="dbpassword9090",
)
with conn.cursor() as cur:
cur.execute("""
DELETE FROM "LiteLLM_ManagedObjectTable"
WHERE file_purpose = 'batch' AND status = 'validating'
""")
deleted = cur.rowcount
conn.commit()
if deleted > 0:
print(f"Deleted {deleted} stale batch records")
conn.close()
except Exception as e:
print(f"Warning: Could not clean up database: {e}")
def clear_s3_callbacks(self):
clear_response = httpx.delete(f"{get_mock_server_base_url()}/mock-s3/callbacks")
assert clear_response.status_code == 200, (
f"Failed to clear callbacks: {clear_response.text}"
)
return clear_response.json()
@pytest.mark.skipif(
True,
reason="Skipping managed files test till managed files feature is available",
)
@pytest.mark.parametrize(
"model_name",
get_batch_model_names(),
ids=model_id,
)
def test_error_files(self, tmp_path, model_name):
raise NotImplementedError(
"To implement. Fail a batch and retrieve the error file.",
)

View file

@ -1,119 +0,0 @@
#!/usr/bin/env python
"""
Validation script for Azure Batch E2E test setup.
Run this before running the actual tests to verify all components are accessible.
"""
import os
import sys
from pathlib import Path
sys.path.insert(0, os.path.abspath("../.."))
def check_imports():
"""Verify all required imports work."""
print("Checking imports...")
try:
from base_integration_test import (
get_mock_server_base_url,
get_litellm_base_url,
get_litellm_api_key,
)
print(" ✓ base_integration_test imports OK")
from test_managed_files_base import ManagedFilesBase, get_batch_model_names
print(" ✓ test_managed_files_base imports OK")
from fixtures.mock_azure_batch_server import create_mock_azure_batch_server
print(" ✓ mock_azure_batch_server imports OK")
import httpx
import openai
import psycopg2
import uvicorn
print(" ✓ All external dependencies OK")
return True
except ImportError as e:
print(f" ✗ Import error: {e}")
return False
def check_config_file():
"""Verify config file exists."""
print("\nChecking config file...")
config_path = Path(__file__).parent / "fixtures" / "config.yml"
if config_path.exists():
print(f" ✓ Config file found: {config_path}")
return True
else:
print(f" ✗ Config file not found: {config_path}")
return False
def check_database():
"""Verify database connection."""
print("\nChecking database connection...")
try:
import psycopg2
conn = psycopg2.connect(
host="localhost",
port=5432,
database="litellm",
user="llmproxy",
password="dbpassword9090",
)
conn.close()
print(" ✓ Database connection OK")
return True
except Exception as e:
print(f" ✗ Database connection failed: {e}")
print(" Start PostgreSQL with:")
print(" docker run --name litellm-postgres -e POSTGRES_USER=llmproxy \\")
print(" -e POSTGRES_PASSWORD=dbpassword9090 -e POSTGRES_DB=litellm \\")
print(" -p 5432:5432 -d postgres:15")
return False
def check_ports():
"""Check if required ports are available."""
print("\nChecking ports...")
import socket
for port, name in [(4000, "LiteLLM Proxy"), (8090, "Mock Server")]:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
try:
s.bind(("localhost", port))
print(f" ✓ Port {port} ({name}) is available")
except OSError:
print(f" ⚠ Port {port} ({name}) is in use (will reuse if healthy)")
return True
def main():
print("=" * 70)
print("Azure Batch E2E Test Setup Validation")
print("=" * 70)
checks = [
check_imports(),
check_config_file(),
check_database(),
check_ports(),
]
print("\n" + "=" * 70)
if all(checks):
print("✓ All checks passed! Ready to run E2E tests.")
print("\nRun tests with:")
print(" cd litellm")
print(" export DATABASE_URL='postgresql://llmproxy:dbpassword9090@localhost:5432/litellm'")
print(" poetry run pytest tests/proxy_e2e_azure_batches_tests/test_proxy_e2e_azure_batches.py -vv")
return 0
else:
print("✗ Some checks failed. Please fix the issues above.")
return 1
if __name__ == "__main__":
sys.exit(main())

View file

@ -8,6 +8,7 @@ import sys
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from fastapi import HTTPException
sys.path.insert(
0, os.path.abspath("../../../")
@ -485,3 +486,327 @@ class TestSafeDbOverrides:
from litellm.constants import LITELLM_SETTINGS_SAFE_DB_OVERRIDES
assert "default_internal_user_params" in LITELLM_SETTINGS_SAFE_DB_OVERRIDES
# ---------------------------------------------------------------------------
# POST /team/permissions/bulk_update
# ---------------------------------------------------------------------------
class TestBulkUpdateTeamMemberPermissions:
"""Tests for the bulk_update_team_member_permissions endpoint."""
def _make_team(self, team_id: str, permissions: list):
"""Create a mock team object."""
team = MagicMock()
team.team_id = team_id
team.team_member_permissions = permissions
return team
def _admin_key_dict(self):
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
return UserAPIKeyAuth(
user_role=LitellmUserRoles.PROXY_ADMIN.value,
api_key="sk-1234",
)
def _non_admin_key_dict(self):
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
return UserAPIKeyAuth(
user_role=LitellmUserRoles.INTERNAL_USER.value,
api_key="sk-user",
)
# --- apply_to_all_teams tests ---
@pytest.mark.asyncio
async def test_all_teams_appends_preserving_existing(self, monkeypatch):
"""apply_to_all_teams: permissions are merged, not overwritten."""
from litellm.proxy.management_endpoints.team_endpoints import (
bulk_update_team_member_permissions,
)
from litellm.types.proxy.management_endpoints.team_endpoints import (
BulkUpdateTeamMemberPermissionsRequest,
)
team_a = self._make_team("team-a", ["/key/generate"])
team_b = self._make_team("team-b", ["/key/delete", "/key/update"])
mock_batcher = MagicMock()
mock_batcher.commit = AsyncMock(return_value=None)
mock_prisma = MagicMock()
mock_prisma.db.litellm_teamtable.find_many = AsyncMock(return_value=[team_a, team_b])
mock_prisma.db.batch_ = MagicMock(return_value=mock_batcher)
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma)
data = BulkUpdateTeamMemberPermissionsRequest(
permissions=["/team/daily/activity"], apply_to_all_teams=True
)
result = await bulk_update_team_member_permissions(data=data, user_api_key_dict=self._admin_key_dict())
assert result["teams_updated"] == 2
calls = mock_batcher.litellm_teamtable.update.call_args_list
assert len(calls) == 2
team_a_call = [c for c in calls if c.kwargs["where"]["team_id"] == "team-a"][0]
assert "/key/generate" in team_a_call.kwargs["data"]["team_member_permissions"]
assert "/team/daily/activity" in team_a_call.kwargs["data"]["team_member_permissions"]
team_b_call = [c for c in calls if c.kwargs["where"]["team_id"] == "team-b"][0]
assert "/key/delete" in team_b_call.kwargs["data"]["team_member_permissions"]
assert "/key/update" in team_b_call.kwargs["data"]["team_member_permissions"]
@pytest.mark.asyncio
async def test_all_teams_skips_teams_that_already_have_permission(self, monkeypatch):
"""apply_to_all_teams: teams that already have the permission are skipped."""
from litellm.proxy.management_endpoints.team_endpoints import (
bulk_update_team_member_permissions,
)
from litellm.types.proxy.management_endpoints.team_endpoints import (
BulkUpdateTeamMemberPermissionsRequest,
)
team_has = self._make_team("team-has", ["/team/daily/activity", "/key/update"])
team_missing = self._make_team("team-missing", ["/key/generate"])
mock_batcher = MagicMock()
mock_batcher.commit = AsyncMock(return_value=None)
mock_prisma = MagicMock()
mock_prisma.db.litellm_teamtable.find_many = AsyncMock(return_value=[team_has, team_missing])
mock_prisma.db.batch_ = MagicMock(return_value=mock_batcher)
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma)
data = BulkUpdateTeamMemberPermissionsRequest(
permissions=["/team/daily/activity"], apply_to_all_teams=True
)
result = await bulk_update_team_member_permissions(data=data, user_api_key_dict=self._admin_key_dict())
assert result["teams_updated"] == 1
calls = mock_batcher.litellm_teamtable.update.call_args_list
assert len(calls) == 1
assert calls[0].kwargs["where"]["team_id"] == "team-missing"
@pytest.mark.asyncio
async def test_all_teams_pagination(self, monkeypatch):
"""apply_to_all_teams: cursor-based pagination processes multiple pages."""
from litellm.proxy.management_endpoints.team_endpoints import (
bulk_update_team_member_permissions,
)
from litellm.types.proxy.management_endpoints.team_endpoints import (
BulkUpdateTeamMemberPermissionsRequest,
)
page1 = [self._make_team(f"team-{i}", []) for i in range(500)]
page2 = [self._make_team(f"team-{i}", []) for i in range(500, 502)]
mock_batcher = MagicMock()
mock_batcher.commit = AsyncMock(return_value=None)
mock_prisma = MagicMock()
mock_prisma.db.litellm_teamtable.find_many = AsyncMock(side_effect=[page1, page2])
mock_prisma.db.batch_ = MagicMock(return_value=mock_batcher)
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma)
data = BulkUpdateTeamMemberPermissionsRequest(
permissions=["/team/daily/activity"], apply_to_all_teams=True
)
result = await bulk_update_team_member_permissions(data=data, user_api_key_dict=self._admin_key_dict())
assert result["teams_updated"] == 502
find_calls = mock_prisma.db.litellm_teamtable.find_many.call_args_list
assert len(find_calls) == 2
assert find_calls[1].kwargs["cursor"] == {"team_id": "team-499"}
assert mock_batcher.commit.call_count == 2
# --- team_ids tests ---
@pytest.mark.asyncio
async def test_team_ids_updates_only_specified_teams(self, monkeypatch):
"""team_ids: only the specified teams are fetched and updated."""
from litellm.proxy.management_endpoints.team_endpoints import (
bulk_update_team_member_permissions,
)
from litellm.types.proxy.management_endpoints.team_endpoints import (
BulkUpdateTeamMemberPermissionsRequest,
)
team_a = self._make_team("team-a", ["/key/generate"])
team_b = self._make_team("team-b", ["/key/delete"])
mock_batcher = MagicMock()
mock_batcher.commit = AsyncMock(return_value=None)
mock_prisma = MagicMock()
mock_prisma.db.litellm_teamtable.find_many = AsyncMock(return_value=[team_a, team_b])
mock_prisma.db.batch_ = MagicMock(return_value=mock_batcher)
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma)
data = BulkUpdateTeamMemberPermissionsRequest(
permissions=["/team/daily/activity"], team_ids=["team-a", "team-b"]
)
result = await bulk_update_team_member_permissions(data=data, user_api_key_dict=self._admin_key_dict())
assert result["teams_updated"] == 2
# Verify find_many was called with the team_ids filter
find_call = mock_prisma.db.litellm_teamtable.find_many.call_args
assert find_call.kwargs["where"] == {"team_id": {"in": ["team-a", "team-b"]}}
@pytest.mark.asyncio
async def test_team_ids_skips_teams_that_already_have_permission(self, monkeypatch):
"""team_ids: teams that already have the permission are skipped."""
from litellm.proxy.management_endpoints.team_endpoints import (
bulk_update_team_member_permissions,
)
from litellm.types.proxy.management_endpoints.team_endpoints import (
BulkUpdateTeamMemberPermissionsRequest,
)
team_has = self._make_team("team-has", ["/team/daily/activity"])
team_missing = self._make_team("team-missing", [])
mock_batcher = MagicMock()
mock_batcher.commit = AsyncMock(return_value=None)
mock_prisma = MagicMock()
mock_prisma.db.litellm_teamtable.find_many = AsyncMock(return_value=[team_has, team_missing])
mock_prisma.db.batch_ = MagicMock(return_value=mock_batcher)
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma)
data = BulkUpdateTeamMemberPermissionsRequest(
permissions=["/team/daily/activity"], team_ids=["team-has", "team-missing"]
)
result = await bulk_update_team_member_permissions(data=data, user_api_key_dict=self._admin_key_dict())
assert result["teams_updated"] == 1
calls = mock_batcher.litellm_teamtable.update.call_args_list
assert calls[0].kwargs["where"]["team_id"] == "team-missing"
@pytest.mark.asyncio
async def test_team_ids_returns_404_for_missing_teams(self, monkeypatch):
"""If any provided team_ids don't exist, return 404."""
from litellm.proxy.management_endpoints.team_endpoints import (
bulk_update_team_member_permissions,
)
from litellm.types.proxy.management_endpoints.team_endpoints import (
BulkUpdateTeamMemberPermissionsRequest,
)
team_a = self._make_team("team-a", ["/key/generate"])
mock_prisma = MagicMock()
# Only team-a exists, team-b does not
mock_prisma.db.litellm_teamtable.find_many = AsyncMock(return_value=[team_a])
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma)
data = BulkUpdateTeamMemberPermissionsRequest(
permissions=["/team/daily/activity"], team_ids=["team-a", "team-b"]
)
with pytest.raises(HTTPException) as exc_info:
await bulk_update_team_member_permissions(data=data, user_api_key_dict=self._admin_key_dict())
assert exc_info.value.status_code == 404
assert "team-b" in str(exc_info.value.detail)
# --- validation tests ---
@pytest.mark.asyncio
async def test_rejects_when_no_team_ids_and_no_apply_all(self, monkeypatch):
"""Must provide team_ids or set apply_to_all_teams=True."""
from litellm.proxy.management_endpoints.team_endpoints import (
bulk_update_team_member_permissions,
)
from litellm.types.proxy.management_endpoints.team_endpoints import (
BulkUpdateTeamMemberPermissionsRequest,
)
mock_prisma = MagicMock()
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma)
data = BulkUpdateTeamMemberPermissionsRequest(permissions=["/team/daily/activity"])
with pytest.raises(HTTPException) as exc_info:
await bulk_update_team_member_permissions(data=data, user_api_key_dict=self._admin_key_dict())
assert exc_info.value.status_code == 400
@pytest.mark.asyncio
async def test_rejects_when_both_team_ids_and_apply_all(self, monkeypatch):
"""Cannot set both team_ids and apply_to_all_teams."""
from litellm.proxy.management_endpoints.team_endpoints import (
bulk_update_team_member_permissions,
)
from litellm.types.proxy.management_endpoints.team_endpoints import (
BulkUpdateTeamMemberPermissionsRequest,
)
mock_prisma = MagicMock()
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma)
data = BulkUpdateTeamMemberPermissionsRequest(
permissions=["/team/daily/activity"],
team_ids=["team-a"],
apply_to_all_teams=True,
)
with pytest.raises(HTTPException) as exc_info:
await bulk_update_team_member_permissions(data=data, user_api_key_dict=self._admin_key_dict())
assert exc_info.value.status_code == 400
@pytest.mark.asyncio
async def test_empty_permissions_list_is_noop(self, monkeypatch):
"""Passing an empty permissions list returns immediately with 0 updated."""
from litellm.proxy.management_endpoints.team_endpoints import (
bulk_update_team_member_permissions,
)
from litellm.types.proxy.management_endpoints.team_endpoints import (
BulkUpdateTeamMemberPermissionsRequest,
)
mock_prisma = MagicMock()
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma)
data = BulkUpdateTeamMemberPermissionsRequest(permissions=[])
result = await bulk_update_team_member_permissions(data=data, user_api_key_dict=self._admin_key_dict())
assert result["teams_updated"] == 0
mock_prisma.db.litellm_teamtable.find_many.assert_not_called()
@pytest.mark.asyncio
async def test_non_admin_gets_403(self, monkeypatch):
"""Non-admin users are rejected with 403."""
from litellm.proxy.management_endpoints.team_endpoints import (
bulk_update_team_member_permissions,
)
from litellm.types.proxy.management_endpoints.team_endpoints import (
BulkUpdateTeamMemberPermissionsRequest,
)
mock_prisma = MagicMock()
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma)
data = BulkUpdateTeamMemberPermissionsRequest(
permissions=["/team/daily/activity"], apply_to_all_teams=True
)
with pytest.raises(HTTPException) as exc_info:
await bulk_update_team_member_permissions(data=data, user_api_key_dict=self._non_admin_key_dict())
assert exc_info.value.status_code == 403
def test_invalid_permission_rejected_by_pydantic(self):
"""Invalid permission strings are rejected at the type level by Pydantic."""
from pydantic import ValidationError
from litellm.types.proxy.management_endpoints.team_endpoints import (
BulkUpdateTeamMemberPermissionsRequest,
)
with pytest.raises(ValidationError):
BulkUpdateTeamMemberPermissionsRequest(permissions=["/not/a/real/permission"])

View file

@ -148,7 +148,10 @@ async def test_get_prompt_info_by_base_id():
)
# Mock In-Memory Registry
# Patch prisma_client to None to avoid leaking state from other tests
with patch(
"litellm.proxy.proxy_server.prisma_client", None
), patch(
"litellm.proxy.prompts.prompt_registry.IN_MEMORY_PROMPT_REGISTRY"
) as mock_registry:
# Setup mocks behavior

View file

@ -1,7 +0,0 @@
# js-yaml CVE-2025-64718
# This vulnerability is not applicable because we've forced js-yaml to version 4.1.1
# via npm overrides in package.json. Trivy incorrectly reports this based on
# dependency requirements in the lockfile, but the actual installed version is 4.1.1.
# Verified with: npm list js-yaml
CVE-2025-64718