mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-10 22:41:41 +00:00
c85da0a75f
/v1/messages against a non-Anthropic model answers with the Responses id, but the spend row was built from a fresh ModelResponse, so it landed on a chatcmpl- uuid nobody can look up. Carry that id through the same way the Anthropic branch now does, and make the passthrough spend assertions fail on an empty lookup instead of skipping past it.
471 lines
19 KiB
Python
471 lines
19 KiB
Python
"""
|
|
This test ensures that the proxy can passthrough anthropic requests
|
|
"""
|
|
|
|
import pytest
|
|
import anthropic
|
|
import aiohttp
|
|
import asyncio
|
|
import json
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.flaky(retries=3, delay=2)
|
|
async def test_anthropic_basic_completion_with_headers():
|
|
print("making basic completion request to anthropic passthrough with aiohttp")
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer sk-1234",
|
|
"Content-Type": "application/json",
|
|
"Anthropic-Version": "2023-06-01",
|
|
}
|
|
|
|
payload = {
|
|
"model": "claude-sonnet-4-5-20250929",
|
|
"max_tokens": 10,
|
|
"messages": [{"role": "user", "content": "Say 'hello test' and nothing else"}],
|
|
"litellm_metadata": {
|
|
"tags": ["test-tag-1", "test-tag-2"],
|
|
},
|
|
}
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(
|
|
"http://0.0.0.0:4000/anthropic/v1/messages", json=payload, headers=headers
|
|
) as response:
|
|
response_text = await response.text()
|
|
print(f"Response text: {response_text}")
|
|
|
|
response_json = await response.json()
|
|
response_headers = response.headers
|
|
print(
|
|
"non-streaming response",
|
|
json.dumps(response_json, indent=4, default=str),
|
|
)
|
|
reported_usage = response_json.get("usage", None)
|
|
# fix null checks for reported_usage
|
|
anthropic_api_input_tokens = (
|
|
reported_usage.get("input_tokens", None) if reported_usage else None
|
|
)
|
|
anthropic_api_output_tokens = (
|
|
reported_usage.get("output_tokens", None) if reported_usage else None
|
|
)
|
|
anthropic_message_id = response_json.get("id")
|
|
|
|
print(f"Anthropic message ID: {anthropic_message_id}")
|
|
|
|
# Wait for spend to be logged
|
|
await asyncio.sleep(15)
|
|
|
|
# Check spend logs for this specific request with retry logic
|
|
spend_data = None
|
|
max_retries = 2
|
|
for attempt in range(max_retries):
|
|
print(f"Attempt {attempt + 1}/{max_retries} to check spend logs")
|
|
|
|
async with session.get(
|
|
f"http://0.0.0.0:4000/spend/logs?request_id={anthropic_message_id}",
|
|
headers={"Authorization": "Bearer sk-1234"},
|
|
) as spend_response:
|
|
print("text spend response")
|
|
print(f"Spend response: {spend_response}")
|
|
spend_data = await spend_response.json()
|
|
print(f"Spend data: {spend_data}")
|
|
|
|
# Check if spend data exists and has entries
|
|
if spend_data and len(spend_data) > 0:
|
|
print("Spend logs found!")
|
|
break
|
|
else:
|
|
print("Spend logs not found yet...")
|
|
if (
|
|
attempt < max_retries - 1
|
|
): # Don't wait after the last attempt
|
|
print("Waiting 10 seconds before retry...")
|
|
await asyncio.sleep(10)
|
|
|
|
if not isinstance(spend_data, list):
|
|
print(f"Spend endpoint answered with an error response: {spend_data}")
|
|
print("Skipping spend assertions (spend logs unreachable in CI)")
|
|
return
|
|
|
|
assert spend_data, (
|
|
f"GET /spend/logs?request_id={anthropic_message_id} found no row for the id "
|
|
"the caller received"
|
|
)
|
|
|
|
log_entry = spend_data[0]
|
|
|
|
# Basic existence checks
|
|
assert isinstance(log_entry, dict), "Log entry should be a dictionary"
|
|
|
|
# Request metadata assertions
|
|
assert (
|
|
log_entry["request_id"] == anthropic_message_id
|
|
), "Request ID should be the message id the caller received"
|
|
assert (
|
|
log_entry["call_type"] == "pass_through_endpoint"
|
|
), "Call type should be pass_through_endpoint"
|
|
assert (
|
|
log_entry["api_base"] == "https://api.anthropic.com/v1/messages"
|
|
), "API base should be Anthropic's endpoint"
|
|
|
|
# Token and spend assertions
|
|
assert log_entry["spend"] > 0, "Spend value should not be None"
|
|
assert isinstance(
|
|
log_entry["spend"], (int, float)
|
|
), "Spend should be a number"
|
|
assert log_entry["total_tokens"] > 0, "Should have some tokens"
|
|
assert (
|
|
log_entry["prompt_tokens"] == anthropic_api_input_tokens
|
|
), f"Should have prompt tokens matching anthropic api. Expected {anthropic_api_input_tokens} but got {log_entry['prompt_tokens']}"
|
|
assert (
|
|
log_entry["completion_tokens"] == anthropic_api_output_tokens
|
|
), f"Should have completion tokens matching anthropic api. Expected {anthropic_api_output_tokens} but got {log_entry['completion_tokens']}"
|
|
assert (
|
|
log_entry["total_tokens"]
|
|
== log_entry["prompt_tokens"] + log_entry["completion_tokens"]
|
|
), "Total tokens should equal prompt + completion"
|
|
|
|
# Time assertions
|
|
assert all(
|
|
key in log_entry
|
|
for key in ["startTime", "endTime", "completionStartTime"]
|
|
), "Should have all time fields"
|
|
assert (
|
|
log_entry["startTime"] < log_entry["endTime"]
|
|
), "Start time should be before end time"
|
|
|
|
# Metadata assertions
|
|
assert str(log_entry["cache_hit"]).lower() != "true", "Cache should be off"
|
|
assert log_entry["request_tags"] == [
|
|
"test-tag-1",
|
|
"test-tag-2",
|
|
], "Tags should match input"
|
|
assert (
|
|
"user_api_key" in log_entry["metadata"]
|
|
), "Should have user API key in metadata"
|
|
|
|
assert "claude" in log_entry["model"]
|
|
assert log_entry["custom_llm_provider"] == "anthropic"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_anthropic_streaming_with_headers():
|
|
print("making streaming request to anthropic passthrough with aiohttp")
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer sk-1234",
|
|
"Content-Type": "application/json",
|
|
"Anthropic-Version": "2023-06-01",
|
|
}
|
|
|
|
payload = {
|
|
"model": "claude-sonnet-4-5-20250929",
|
|
"max_tokens": 10,
|
|
"messages": [
|
|
{"role": "user", "content": "Say 'hello stream test' and nothing else"}
|
|
],
|
|
"stream": True,
|
|
"litellm_metadata": {
|
|
"tags": ["test-tag-stream-1", "test-tag-stream-2"],
|
|
"user": "test-user-1",
|
|
},
|
|
}
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(
|
|
"http://0.0.0.0:4000/anthropic/v1/messages", json=payload, headers=headers
|
|
) as response:
|
|
print("response status")
|
|
print(response.status)
|
|
assert response.status == 200, "Response should be successful"
|
|
response_headers = response.headers
|
|
print(f"Response headers: {response_headers}")
|
|
|
|
collected_output = []
|
|
async for line in response.content:
|
|
if line:
|
|
text = line.decode("utf-8").strip()
|
|
if text.startswith("data: "):
|
|
collected_output.append(text[6:]) # Remove 'data: ' prefix
|
|
|
|
print("Collected output:", "".join(collected_output))
|
|
anthropic_api_usage_chunks = []
|
|
anthropic_message_id = None
|
|
for chunk in collected_output:
|
|
chunk_json = json.loads(chunk)
|
|
if chunk_json.get("type") == "message_start":
|
|
anthropic_message_id = chunk_json.get("message", {}).get("id")
|
|
if "usage" in chunk_json:
|
|
anthropic_api_usage_chunks.append(chunk_json["usage"])
|
|
elif "message" in chunk_json and "usage" in chunk_json["message"]:
|
|
anthropic_api_usage_chunks.append(chunk_json["message"]["usage"])
|
|
|
|
print(f"Anthropic message ID: {anthropic_message_id}")
|
|
|
|
print(
|
|
"anthropic_api_usage_chunks",
|
|
json.dumps(anthropic_api_usage_chunks, indent=4, default=str),
|
|
)
|
|
|
|
print("anthropic_api_usage_chunks: ", anthropic_api_usage_chunks)
|
|
# Get the most recent value of input tokens (iterate backwards to find last non-zero value)
|
|
anthropic_api_input_tokens = 0
|
|
for usage in reversed(anthropic_api_usage_chunks):
|
|
if usage.get("input_tokens", 0) > 0:
|
|
anthropic_api_input_tokens = usage.get("input_tokens", 0)
|
|
break
|
|
anthropic_api_output_tokens = 0
|
|
for usage in reversed(anthropic_api_usage_chunks):
|
|
if usage.get("output_tokens", 0) > 0:
|
|
anthropic_api_output_tokens = usage.get("output_tokens", 0)
|
|
break
|
|
|
|
print("anthropic_api_input_tokens", anthropic_api_input_tokens)
|
|
print("anthropic_api_output_tokens", anthropic_api_output_tokens)
|
|
|
|
# Wait for spend to be logged
|
|
await asyncio.sleep(20)
|
|
|
|
# Check spend logs for this specific request with retry logic
|
|
spend_data = None
|
|
max_retries = 2
|
|
for attempt in range(max_retries):
|
|
print(f"Attempt {attempt + 1}/{max_retries} to check spend logs")
|
|
|
|
async with session.get(
|
|
f"http://0.0.0.0:4000/spend/logs?request_id={anthropic_message_id}",
|
|
headers={"Authorization": "Bearer sk-1234"},
|
|
) as spend_response:
|
|
spend_data = await spend_response.json()
|
|
print(f"Spend data: {spend_data}")
|
|
|
|
# Check if spend data exists and has entries
|
|
if spend_data and len(spend_data) > 0:
|
|
print("Spend logs found!")
|
|
break
|
|
else:
|
|
print("Spend logs not found yet...")
|
|
if (
|
|
attempt < max_retries - 1
|
|
): # Don't wait after the last attempt
|
|
print("Waiting 10 seconds before retry...")
|
|
await asyncio.sleep(10)
|
|
|
|
if not isinstance(spend_data, list):
|
|
print(f"Spend endpoint answered with an error response: {spend_data}")
|
|
print("Skipping spend assertions (spend logs unreachable in CI)")
|
|
return
|
|
|
|
assert spend_data, (
|
|
f"GET /spend/logs?request_id={anthropic_message_id} found no row for the id "
|
|
"the caller received"
|
|
)
|
|
|
|
log_entry = spend_data[0]
|
|
|
|
# Basic existence checks
|
|
assert isinstance(log_entry, dict), "Log entry should be a dictionary"
|
|
|
|
# Request metadata assertions
|
|
assert (
|
|
log_entry["request_id"] == anthropic_message_id
|
|
), "Request ID should be the message id the caller received"
|
|
assert (
|
|
log_entry["call_type"] == "pass_through_endpoint"
|
|
), "Call type should be pass_through_endpoint"
|
|
# assert (
|
|
# log_entry["api_base"] == "https://api.anthropic.com/v1/messages"
|
|
# ), "API base should be Anthropic's endpoint"
|
|
|
|
# Token and spend assertions
|
|
assert log_entry["spend"] > 0, "Spend value should not be None"
|
|
assert isinstance(
|
|
log_entry["spend"], (int, float)
|
|
), "Spend should be a number"
|
|
assert log_entry["total_tokens"] > 0, "Should have some tokens"
|
|
assert (
|
|
log_entry["prompt_tokens"] == anthropic_api_input_tokens
|
|
), f"Should have prompt tokens matching anthropic api. Expected {anthropic_api_input_tokens} but got {log_entry['prompt_tokens']}"
|
|
assert (
|
|
log_entry["completion_tokens"] == anthropic_api_output_tokens
|
|
), f"Should have completion tokens matching anthropic api. Expected {anthropic_api_output_tokens} but got {log_entry['completion_tokens']}"
|
|
assert (
|
|
log_entry["total_tokens"]
|
|
== log_entry["prompt_tokens"] + log_entry["completion_tokens"]
|
|
), "Total tokens should equal prompt + completion"
|
|
|
|
# Time assertions
|
|
assert all(
|
|
key in log_entry
|
|
for key in ["startTime", "endTime", "completionStartTime"]
|
|
), "Should have all time fields"
|
|
assert (
|
|
log_entry["startTime"] < log_entry["endTime"]
|
|
), "Start time should be before end time"
|
|
|
|
# Metadata assertions
|
|
assert str(log_entry["cache_hit"]).lower() != "true", "Cache should be off"
|
|
assert log_entry["request_tags"] == [
|
|
"test-tag-stream-1",
|
|
"test-tag-stream-2",
|
|
], "Tags should match input"
|
|
assert (
|
|
"user_api_key" in log_entry["metadata"]
|
|
), "Should have user API key in metadata"
|
|
|
|
assert "claude" in log_entry["model"]
|
|
|
|
assert log_entry["end_user"] == "test-user-1"
|
|
assert log_entry["custom_llm_provider"] == "anthropic"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.flaky(retries=3, delay=2)
|
|
async def test_anthropic_messages_streaming_cost_injection():
|
|
"""
|
|
Test that cost is injected into message_delta usage for Anthropic Messages API streaming
|
|
"""
|
|
print("Testing cost injection in Anthropic Messages API streaming response")
|
|
|
|
headers = {
|
|
"Authorization": "Bearer sk-1234",
|
|
"Content-Type": "application/json",
|
|
"anthropic-version": "2023-06-01",
|
|
}
|
|
|
|
payload = {
|
|
"model": "claude-haiku-4-5-20251001",
|
|
"max_tokens": 10,
|
|
"stream": True,
|
|
"messages": [{"role": "user", "content": "Say 'Hi'"}],
|
|
}
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(
|
|
"http://0.0.0.0:4000/v1/messages",
|
|
json=payload,
|
|
headers=headers,
|
|
) as response:
|
|
assert response.status == 200
|
|
|
|
# Collect all SSE events.
|
|
# Split each chunk by newlines to handle both:
|
|
# - Anthropic direct path: chunks arrive as individual lines
|
|
# - OpenAI/Responses API path: chunks are full multi-line SSE events
|
|
events = []
|
|
async for chunk in response.content:
|
|
chunk_str = chunk.decode("utf-8")
|
|
for line in chunk_str.split("\n"):
|
|
line = line.strip()
|
|
if line.startswith("data: "):
|
|
try:
|
|
data = json.loads(line[6:]) # Remove 'data: ' prefix
|
|
events.append(data)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
# Find message_delta event with usage
|
|
message_delta_events = [
|
|
event
|
|
for event in events
|
|
if event.get("type") == "message_delta" and "usage" in event
|
|
]
|
|
|
|
assert (
|
|
len(message_delta_events) > 0
|
|
), "No message_delta events with usage found"
|
|
|
|
# Check that cost is included in usage
|
|
for event in message_delta_events:
|
|
usage = event.get("usage", {})
|
|
assert "cost" in usage, f"Cost not found in usage: {usage}"
|
|
assert isinstance(
|
|
usage["cost"], (int, float)
|
|
), f"Cost should be numeric: {usage['cost']}"
|
|
assert (
|
|
usage["cost"] >= 0
|
|
), f"Cost should be non-negative: {usage['cost']}"
|
|
|
|
print(f"Found message_delta with cost: {usage}")
|
|
|
|
print(
|
|
f"Test passed: Found {len(message_delta_events)} message_delta events with cost"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.flaky(retries=3, delay=2)
|
|
async def test_anthropic_messages_openai_model_streaming_cost_injection():
|
|
"""
|
|
Test that cost is injected into message_delta usage for OpenAI model via Anthropic Messages API
|
|
"""
|
|
print("Testing cost injection in Anthropic Messages API with OpenAI model")
|
|
|
|
headers = {
|
|
"Authorization": "Bearer sk-1234",
|
|
"Content-Type": "application/json",
|
|
"anthropic-version": "2023-06-01",
|
|
}
|
|
|
|
payload = {
|
|
"model": "openai/gpt-4o",
|
|
"max_tokens": 20,
|
|
"stream": True,
|
|
"messages": [{"role": "user", "content": "Say 'Hi'"}],
|
|
}
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(
|
|
"http://0.0.0.0:4000/v1/messages",
|
|
json=payload,
|
|
headers=headers,
|
|
) as response:
|
|
assert response.status == 200
|
|
|
|
# Collect all SSE events.
|
|
# Split each chunk by newlines to handle both:
|
|
# - Direct API paths: chunks arrive as individual lines
|
|
# - OpenAI/Responses API path: AnthropicResponsesStreamWrapper yields
|
|
# full multi-line SSE events as single bytes objects, so a naive
|
|
# startswith('data: ') check on the whole chunk misses them.
|
|
events = []
|
|
async for chunk in response.content:
|
|
chunk_str = chunk.decode("utf-8")
|
|
for line in chunk_str.split("\n"):
|
|
line = line.strip()
|
|
if line.startswith("data: "):
|
|
try:
|
|
data = json.loads(line[6:]) # Remove 'data: ' prefix
|
|
events.append(data)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
# Find message_delta event with usage
|
|
message_delta_events = [
|
|
event
|
|
for event in events
|
|
if event.get("type") == "message_delta" and "usage" in event
|
|
]
|
|
|
|
assert (
|
|
len(message_delta_events) > 0
|
|
), "No message_delta events with usage found"
|
|
|
|
# Check that cost is included in usage
|
|
for event in message_delta_events:
|
|
usage = event.get("usage", {})
|
|
assert "cost" in usage, f"Cost not found in usage: {usage}"
|
|
assert isinstance(
|
|
usage["cost"], (int, float)
|
|
), f"Cost should be numeric: {usage['cost']}"
|
|
assert (
|
|
usage["cost"] >= 0
|
|
), f"Cost should be non-negative: {usage['cost']}"
|
|
|
|
print(f"Found message_delta with cost: {usage}")
|
|
|
|
print(
|
|
f"Test passed: Found {len(message_delta_events)} message_delta events with cost"
|
|
)
|