feat(tests): add proxy e2e azure batches test fixtures

- Mock Azure Batch server (FastAPI) with batch, file, chat, embeddings, S3 callback routes
- Config for mock models and batch polling
- Fix mock_responses: use 'model' instead of undefined 'request_model'

Made-with: Cursor
This commit is contained in:
Sameer Kankute 2026-03-04 17:15:15 +05:30
parent 8541272629
commit e3a93b9597
11 changed files with 1288 additions and 0 deletions

View file

@ -9779,6 +9779,122 @@
}
]
},
"dashscope/qwen3-max-2026-01-23": {
"litellm_provider": "dashscope",
"max_input_tokens": 258048,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"source": "https://www.alibabacloud.com/help/en/model-studio/models",
"supports_function_calling": true,
"supports_reasoning": true,
"supports_tool_choice": true,
"tiered_pricing": [
{
"input_cost_per_token": 1.2e-06,
"output_cost_per_token": 6e-06,
"range": [
0,
32000.0
]
},
{
"input_cost_per_token": 2.4e-06,
"output_cost_per_token": 1.2e-05,
"range": [
32000.0,
128000.0
]
},
{
"input_cost_per_token": 3e-06,
"output_cost_per_token": 1.5e-05,
"range": [
128000.0,
252000.0
]
}
]
},
"dashscope/qwen3-next-80b-a3b-instruct": {
"input_cost_per_token": 1.5e-07,
"litellm_provider": "dashscope",
"max_input_tokens": 262144,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_token": 1.2e-06,
"source": "https://www.alibabacloud.com/help/en/model-studio/model-pricing",
"supports_function_calling": true,
"supports_tool_choice": true
},
"dashscope/qwen3-next-80b-a3b-thinking": {
"input_cost_per_token": 1.5e-07,
"litellm_provider": "dashscope",
"max_input_tokens": 262144,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_token": 1.2e-06,
"source": "https://www.alibabacloud.com/help/en/model-studio/model-pricing",
"supports_function_calling": true,
"supports_reasoning": true,
"supports_tool_choice": true
},
"dashscope/qwen3-vl-235b-a22b-instruct": {
"input_cost_per_token": 4e-07,
"litellm_provider": "dashscope",
"max_input_tokens": 131072,
"max_output_tokens": 32768,
"max_tokens": 32768,
"mode": "chat",
"output_cost_per_token": 1.6e-06,
"source": "https://www.alibabacloud.com/help/en/model-studio/model-pricing",
"supports_function_calling": true,
"supports_tool_choice": true,
"supports_vision": true
},
"dashscope/qwen3-vl-235b-a22b-thinking": {
"input_cost_per_token": 4e-07,
"litellm_provider": "dashscope",
"max_input_tokens": 131072,
"max_output_tokens": 32768,
"max_tokens": 32768,
"mode": "chat",
"output_cost_per_token": 4e-06,
"source": "https://www.alibabacloud.com/help/en/model-studio/model-pricing",
"supports_function_calling": true,
"supports_reasoning": true,
"supports_tool_choice": true,
"supports_vision": true
},
"dashscope/qwen3-vl-32b-instruct": {
"input_cost_per_token": 1.6e-07,
"litellm_provider": "dashscope",
"max_input_tokens": 131072,
"max_output_tokens": 32768,
"max_tokens": 32768,
"mode": "chat",
"output_cost_per_token": 6.4e-07,
"source": "https://www.alibabacloud.com/help/en/model-studio/model-pricing",
"supports_function_calling": true,
"supports_tool_choice": true,
"supports_vision": true
},
"dashscope/qwen3-vl-32b-thinking": {
"input_cost_per_token": 1.6e-07,
"litellm_provider": "dashscope",
"max_input_tokens": 131072,
"max_output_tokens": 32768,
"max_tokens": 32768,
"mode": "chat",
"output_cost_per_token": 2.87e-06,
"source": "https://www.alibabacloud.com/help/en/model-studio/model-pricing",
"supports_function_calling": true,
"supports_reasoning": true,
"supports_tool_choice": true,
"supports_vision": true
},
"dashscope/qwen3-vl-plus": {
"litellm_provider": "dashscope",
"max_input_tokens": 260096,
@ -25806,6 +25922,30 @@
"supports_vision": true,
"tool_use_system_prompt_tokens": 159
},
"openrouter/anthropic/claude-sonnet-4.6": {
"cache_creation_input_token_cost": 3.75e-06,
"cache_creation_input_token_cost_above_200k_tokens": 7.5e-06,
"cache_read_input_token_cost": 3e-07,
"cache_read_input_token_cost_above_200k_tokens": 6e-07,
"input_cost_per_token": 3e-06,
"input_cost_per_token_above_200k_tokens": 6e-06,
"litellm_provider": "openrouter",
"max_input_tokens": 1000000,
"max_output_tokens": 128000,
"max_tokens": 128000,
"mode": "chat",
"output_cost_per_token": 1.5e-05,
"output_cost_per_token_above_200k_tokens": 2.25e-05,
"source": "https://openrouter.ai/anthropic/claude-sonnet-4.6",
"supports_assistant_prefill": true,
"supports_computer_use": true,
"supports_function_calling": true,
"supports_prompt_caching": true,
"supports_reasoning": true,
"supports_tool_choice": true,
"supports_vision": true,
"tool_use_system_prompt_tokens": 159
},
"openrouter/anthropic/claude-opus-4.5": {
"cache_creation_input_token_cost": 6.25e-06,
"cache_read_input_token_cost": 5e-07,
@ -26156,6 +26296,39 @@
"supports_web_search": true,
"tpm": 800000
},
"openrouter/google/gemini-3.1-pro-preview": {
"cache_read_input_token_cost": 2e-07,
"cache_read_input_token_cost_above_200k_tokens": 4e-07,
"cache_creation_input_token_cost_above_200k_tokens": 2.5e-07,
"input_cost_per_token": 2e-06,
"input_cost_per_token_above_200k_tokens": 4e-06,
"litellm_provider": "openrouter",
"max_input_tokens": 1048576,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_token": 1.2e-05,
"output_cost_per_token_above_200k_tokens": 1.8e-05,
"source": "https://openrouter.ai/google/gemini-3.1-pro-preview",
"supported_modalities": [
"text",
"image",
"audio",
"video"
],
"supported_output_modalities": [
"text"
],
"supports_audio_input": true,
"supports_function_calling": true,
"supports_pdf_input": true,
"supports_prompt_caching": true,
"supports_reasoning": true,
"supports_response_schema": true,
"supports_system_messages": true,
"supports_tool_choice": true,
"supports_vision": true
},
"openrouter/gryphe/mythomax-l2-13b": {
"input_cost_per_token": 1.875e-06,
"litellm_provider": "openrouter",
@ -26533,6 +26706,29 @@
"supports_reasoning": true,
"supports_tool_choice": true
},
"openrouter/openai/gpt-5.1-codex-max": {
"cache_read_input_token_cost": 1.25e-07,
"input_cost_per_token": 1.25e-06,
"litellm_provider": "openrouter",
"max_input_tokens": 400000,
"max_output_tokens": 128000,
"max_tokens": 128000,
"mode": "chat",
"output_cost_per_token": 1e-05,
"source": "https://openrouter.ai/openai/gpt-5.1-codex-max",
"supported_modalities": [
"text",
"image"
],
"supported_output_modalities": [
"text"
],
"supports_function_calling": true,
"supports_prompt_caching": true,
"supports_reasoning": true,
"supports_tool_choice": true,
"supports_vision": true
},
"openrouter/openai/gpt-5.2": {
"input_cost_per_image": 0,
"cache_read_input_token_cost": 1.75e-07,
@ -26687,6 +26883,19 @@
"supports_tool_choice": true,
"supports_function_calling": true
},
"openrouter/qwen/qwen3-coder-plus": {
"input_cost_per_token": 1e-06,
"litellm_provider": "openrouter",
"max_input_tokens": 997952,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_token": 5e-06,
"source": "https://openrouter.ai/qwen/qwen3-coder-plus",
"supports_function_calling": true,
"supports_reasoning": true,
"supports_tool_choice": true
},
"openrouter/qwen/qwen3-235b-a22b-2507": {
"input_cost_per_token": 7.1e-08,
"litellm_provider": "openrouter",
@ -26822,6 +27031,19 @@
"supports_vision": true,
"supports_prompt_caching": false
},
"openrouter/z-ai/glm-5": {
"input_cost_per_token": 8e-07,
"litellm_provider": "openrouter",
"max_input_tokens": 202752,
"max_output_tokens": 128000,
"max_tokens": 128000,
"mode": "chat",
"output_cost_per_token": 2.56e-06,
"source": "https://openrouter.ai/z-ai/glm-5",
"supports_function_calling": true,
"supports_reasoning": true,
"supports_tool_choice": true
},
"openrouter/minimax/minimax-m2.1": {
"input_cost_per_token": 2.7e-07,
"output_cost_per_token": 1.2e-06,
@ -34327,6 +34549,36 @@
"supports_tool_choice": true,
"source": "https://aws.amazon.com/bedrock/pricing/"
},
"zai/glm-5": {
"cache_creation_input_token_cost": 0,
"cache_read_input_token_cost": 2e-07,
"input_cost_per_token": 1e-06,
"output_cost_per_token": 3.2e-06,
"litellm_provider": "zai",
"max_input_tokens": 200000,
"max_output_tokens": 128000,
"mode": "chat",
"supports_function_calling": true,
"supports_prompt_caching": true,
"supports_reasoning": true,
"supports_tool_choice": true,
"source": "https://docs.z.ai/guides/overview/pricing"
},
"zai/glm-5-code": {
"cache_creation_input_token_cost": 0,
"cache_read_input_token_cost": 3e-07,
"input_cost_per_token": 1.2e-06,
"output_cost_per_token": 5e-06,
"litellm_provider": "zai",
"max_input_tokens": 200000,
"max_output_tokens": 128000,
"mode": "chat",
"supports_function_calling": true,
"supports_prompt_caching": true,
"supports_reasoning": true,
"supports_tool_choice": true,
"source": "https://docs.z.ai/guides/overview/pricing"
},
"zai/glm-4.7": {
"cache_creation_input_token_cost": 0,
"cache_read_input_token_cost": 1.1e-07,

View file

@ -0,0 +1,56 @@
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

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

View file

@ -0,0 +1,517 @@
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

@ -0,0 +1,124 @@
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

@ -0,0 +1,23 @@
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

@ -0,0 +1,170 @@
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

@ -0,0 +1,98 @@
"""
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

@ -0,0 +1,33 @@
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

@ -0,0 +1,12 @@
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)