scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-08 03:08:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 115

test: update test - anthropic returns input tokens in final usage chunk now

Browse source
This commit is contained in:
Krrish Dholakia 2025-08-16 12:31:52 -07:00
parent 2de4ddd194
commit 6e853c7de9
2 changed files with 31 additions and 33 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py
View file

@ -96,7 +96,6 @@ class AnthropicPassthroughLoggingHandler:
handles streaming and non-streaming responses
"""
try:
response_cost = litellm.completion_cost(
completion_response=litellm_model_response,
model=model,

63
tests/pass_through_tests/test_anthropic_passthrough.py
View file

@ -57,7 +57,7 @@ async def test_anthropic_basic_completion_with_headers():
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={litellm_call_id}",
headers={"Authorization": "Bearer sk-1234"},
@ -66,32 +66,30 @@ async def test_anthropic_basic_completion_with_headers():
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
if (
attempt < max_retries - 1
): # Don't wait after the last attempt
print("Waiting 10 seconds before retry...")
await asyncio.sleep(10)
assert spend_data is not None, "Should have spend data for the request"
assert len(spend_data) > 0, "Should have at least one spend log entry"
log_entry = spend_data[
0
] # Get the first (and should be only) log entry
log_entry = spend_data[0] # Get the first (and should be only) log entry
# Basic existence checks
assert spend_data is not None, "Should have spend data for the request"
assert isinstance(log_entry, dict), "Log entry should be a dictionary"
# Request metadata assertions
assert (
log_entry["request_id"] == litellm_call_id
), "Request ID should match"
assert log_entry["request_id"] == litellm_call_id, "Request ID should match"
assert (
log_entry["call_type"] == "pass_through_endpoint"
), "Call type should be pass_through_endpoint"
@ -126,9 +124,7 @@ async def test_anthropic_basic_completion_with_headers():
), "Start time should be before end time"
# Metadata assertions
assert (
str(log_entry["cache_hit"]).lower() != "true"
), "Cache should be off"
assert str(log_entry["cache_hit"]).lower() != "true", "Cache should be off"
assert log_entry["request_tags"] == [
"test-tag-1",
"test-tag-2",
@ -197,12 +193,19 @@ async def test_anthropic_streaming_with_headers():
json.dumps(anthropic_api_usage_chunks, indent=4, default=str),
)
anthropic_api_input_tokens = sum(
[usage.get("input_tokens", 0) for usage in anthropic_api_usage_chunks]
)
anthropic_api_output_tokens = max(
[usage.get("output_tokens", 0) for usage in anthropic_api_usage_chunks]
)
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)
@ -214,39 +217,37 @@ async def test_anthropic_streaming_with_headers():
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={litellm_call_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
if (
attempt < max_retries - 1
): # Don't wait after the last attempt
print("Waiting 10 seconds before retry...")
await asyncio.sleep(10)
assert spend_data is not None, "Should have spend data for the request"
assert len(spend_data) > 0, "Should have at least one spend log entry"
log_entry = spend_data[
0
] # Get the first (and should be only) log entry
log_entry = spend_data[0] # Get the first (and should be only) log entry
# Basic existence checks
assert spend_data is not None, "Should have spend data for the request"
assert isinstance(log_entry, dict), "Log entry should be a dictionary"
# Request metadata assertions
assert (
log_entry["request_id"] == litellm_call_id
), "Request ID should match"
assert log_entry["request_id"] == litellm_call_id, "Request ID should match"
assert (
log_entry["call_type"] == "pass_through_endpoint"
), "Call type should be pass_through_endpoint"
@ -281,9 +282,7 @@ async def test_anthropic_streaming_with_headers():
), "Start time should be before end time"
# Metadata assertions
assert (
str(log_entry["cache_hit"]).lower() != "true"
), "Cache should be off"
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",