mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-29 06:33:28 +00:00
fix(stream): the last usage frame carries litellm's bill in usage.cost (#48955)
* fix(proxy): copy the stream's hidden response_cost into the client's streamed usage.cost The stream is priced once into hidden response_cost (stream builder for chat and text completions, the Responses iterator for the terminal event) and the proxy egress copies that bill into the streamed usage.cost on a copy of the event. Inside the pipeline usage.cost stays what the upstream reported. Spend rows do not change; the billing fix for upstream-reported usage.cost follows in the next PR cut from #48287. * fix(proxy): type the bridge bill carry and the builder's reported-cost guard for the lint gates * style: format the chat-over-Responses bridge helper signature * refactor(bridge): build the billed Responses event with model_copy instead of a TypeAdapter chain * fix(stream): keep writing the computed cost into usage.cost as on main, the hidden copy is only for the proxy egress * fix(stream): write litellm's bill into the usage frame's usage.cost instead of a hidden field the proxy copies * test(stream): keep main's formatting in test_streaming_handler.py, only the new test changes * refactor(stream): drop the docstring on _stamp_usage_cost, the body says it --------- Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
parent
019a9de603
commit
4fc2bba1f2
5 changed files with 151 additions and 22 deletions
|
|
@ -92,6 +92,18 @@ def _next_sync_or_exhausted(it: Iterator[object]) -> object:
|
|||
return _SYNC_ITER_EXHAUSTED
|
||||
|
||||
|
||||
def _stamp_usage_cost(
|
||||
usage_chunk: ModelResponse | TextCompletionResponse | ModelResponseStream,
|
||||
complete_response: ModelResponse | TextCompletionResponse | ModelResponseStream,
|
||||
) -> None:
|
||||
bill: Final = (get_hidden_params(complete_response) or {}).get("response_cost")
|
||||
usage: Final[object] = getattr(usage_chunk, "usage", None)
|
||||
if usage is None or isinstance(bill, bool) or not isinstance(bill, (int, float)):
|
||||
return
|
||||
if bill > 0 or isinstance(getattr(usage, "cost", None), (int, float)):
|
||||
setattr(usage, "cost", float(bill)) # noqa: B010 # cost is an optional extra on the usage model
|
||||
|
||||
|
||||
def _stamp_served_service_tier(response: ModelResponseStream, complete_streaming_response: ModelResponse) -> None:
|
||||
served_tier: Final = complete_streaming_response.model_dump().get("service_tier")
|
||||
if isinstance(served_tier, str) and served_tier:
|
||||
|
|
@ -2141,6 +2153,7 @@ class CustomStreamWrapper:
|
|||
getattr(complete_streaming_response, "usage"),
|
||||
)
|
||||
_stamp_served_service_tier(response, complete_streaming_response)
|
||||
_stamp_usage_cost(response, complete_streaming_response)
|
||||
try:
|
||||
_cache_copy = complete_streaming_response.model_copy(deep=True)
|
||||
_log_copy = complete_streaming_response.model_copy(deep=True)
|
||||
|
|
@ -2361,6 +2374,7 @@ class CustomStreamWrapper:
|
|||
getattr(complete_streaming_response, "usage"),
|
||||
)
|
||||
_stamp_served_service_tier(response, complete_streaming_response)
|
||||
_stamp_usage_cost(response, complete_streaming_response)
|
||||
try:
|
||||
_copy = complete_streaming_response.model_copy(deep=True)
|
||||
except RuntimeError:
|
||||
|
|
|
|||
|
|
@ -39,7 +39,11 @@ from litellm.constants import (
|
|||
)
|
||||
from litellm.exceptions import MidStreamFallbackError, RateLimitError
|
||||
from litellm.litellm_core_utils.asyncify import run_async_function
|
||||
from litellm.litellm_core_utils.core_helpers import process_response_headers
|
||||
from litellm.litellm_core_utils.core_helpers import (
|
||||
process_response_headers,
|
||||
provider_reported_usage_cost,
|
||||
set_response_cost_in_hidden_params,
|
||||
)
|
||||
from litellm.litellm_core_utils.exception_mapping_utils import with_deployment_api_base
|
||||
from litellm.litellm_core_utils.internal_call_metadata import (
|
||||
is_background_response,
|
||||
|
|
@ -1825,20 +1829,32 @@ def _estimate_usage_safely(
|
|||
def stamp_responses_usage_cost(
|
||||
response_obj: ResponsesAPIResponse | None, logging_obj: LiteLLMLoggingObj | None
|
||||
) -> None:
|
||||
if response_obj is None or logging_obj is None:
|
||||
"""The terminal response's ``usage.cost`` is the bill litellm writes to the spend row. It is priced on a copy
|
||||
shaped the way the success handler shapes the spend row's result (the reported ``usage.cost`` in the
|
||||
response-cost header, the usage in chat shape), so the client and the row read the same number."""
|
||||
if response_obj is None:
|
||||
return
|
||||
usage_obj: Final[ResponseAPIUsage | None] = _usage_as_model(getattr(response_obj, "usage", None))
|
||||
if usage_obj is None:
|
||||
return
|
||||
response_obj.usage = usage_obj # rebind-ok: the stamped cost has to ride on the response the client receives
|
||||
if isinstance(getattr(usage_obj, "cost", None), (int, float)):
|
||||
response_obj.usage = usage_obj # rebind-ok: the typed usage has to ride on the response the client receives
|
||||
if logging_obj is None:
|
||||
return
|
||||
priced: Final = response_obj.model_copy(
|
||||
update={"usage": ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(usage_obj)}
|
||||
)
|
||||
priced.hidden_params = {}
|
||||
set_response_cost_in_hidden_params(
|
||||
priced, provider_reported_usage_cost(usage_obj, logging_obj.model_call_details.get("custom_llm_provider"))
|
||||
)
|
||||
try:
|
||||
cost: Final[float | None] = logging_obj.response_cost_calculator(result=response_obj)
|
||||
except Exception:
|
||||
bill: Final = logging_obj.response_cost_calculator(result=priced)
|
||||
except Exception: # noqa: BLE001 # a pricing failure leaves the client usage as the upstream sent it
|
||||
return
|
||||
if isinstance(cost, (int, float)) and cost > 0:
|
||||
setattr(usage_obj, "cost", cost)
|
||||
if isinstance(bill, bool) or not isinstance(bill, (int, float)):
|
||||
return
|
||||
if bill > 0 or isinstance(getattr(usage_obj, "cost", None), (int, float)):
|
||||
setattr(usage_obj, "cost", float(bill))
|
||||
|
||||
|
||||
def opening_item_payload(payload: Mapping[str, object]) -> Mapping[str, object]:
|
||||
|
|
|
|||
|
|
@ -6316,6 +6316,22 @@ def test_chunk_creator_preserves_hidden_provider_specific_fields_from_parsed_chu
|
|||
assert assembled._hidden_params["provider_specific_fields"] == {"traffic_type": "ON_DEMAND_FLEX"}
|
||||
|
||||
|
||||
def test_the_usage_chunk_carries_the_bill_the_builder_priced_for_the_spend_row() -> None:
|
||||
from litellm.litellm_core_utils.streaming_handler import (
|
||||
_stamp_usage_cost, # pyright: ignore[reportPrivateUsage] # the helper under test
|
||||
)
|
||||
|
||||
complete: Final = ModelResponse(model="omni-model", usage=Usage(prompt_tokens=100, completion_tokens=20))
|
||||
complete.hidden_params = {"response_cost": 0.00032}
|
||||
chunk_usage: Final = Usage(prompt_tokens=100, completion_tokens=20, total_tokens=120)
|
||||
setattr(chunk_usage, "cost", 0.037)
|
||||
usage_chunk: Final = ModelResponseStream(id="chunk-1", model="omni-model", choices=[], usage=chunk_usage)
|
||||
|
||||
_stamp_usage_cost(usage_chunk, complete)
|
||||
|
||||
assert usage_chunk.usage is not None and usage_chunk.usage.cost == 0.00032, "the client sees what the row bills"
|
||||
|
||||
|
||||
def test_chunk_creator_keeps_provider_model_private_across_stream():
|
||||
from litellm.router_utils.add_retry_fallback_headers import (
|
||||
get_hidden_params_dict,
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ before response.completed, and that every event of a bridged stream carries the
|
|||
spend tracking stores, so a follow-up previous_response_id still finds the conversation.
|
||||
"""
|
||||
|
||||
import itertools
|
||||
import json
|
||||
from collections.abc import Callable
|
||||
from typing import Final
|
||||
|
|
@ -23,6 +24,7 @@ from litellm.responses.litellm_completion_transformation.streaming_iterator impo
|
|||
from litellm.responses.utils import ResponsesAPIRequestUtils
|
||||
from litellm.types.llms.openai import (
|
||||
BaseLiteLLMOpenAIResponseObject,
|
||||
OutputTextDeltaEvent,
|
||||
ResponsesAPIStreamEvents,
|
||||
)
|
||||
from litellm.types.responses.main import build_web_search_call
|
||||
|
|
@ -1658,3 +1660,34 @@ async def test_a_compaction_block_without_its_summary_streams_no_compaction_item
|
|||
|
||||
assert [event["item"]["type"] for event in events if "item" in event] == ["message", "message"]
|
||||
assert [item["type"] for item in events[-1]["response"]["output"]] == ["message"]
|
||||
|
||||
|
||||
def test_the_partial_response_billed_on_disconnect_carries_the_logging_objects_bill_as_hidden_response_cost() -> None:
|
||||
from litellm.litellm_core_utils.hidden_params import get_hidden_params
|
||||
|
||||
wrapper: Final = _FakeStreamWrapper([_chunk("Hello"), _chunk(" world", finish_reason="stop")])
|
||||
wrapper.logging_obj.response_cost_calculator.return_value = 0.00032
|
||||
iterator: Final = LiteLLMCompletionStreamingIterator(
|
||||
model="claude-haiku-4-5",
|
||||
litellm_custom_stream_wrapper=wrapper, # pyright: ignore[reportArgumentType] # the fake stands in for CustomStreamWrapper, as in _build_iterator
|
||||
request_input="What is the weather in San Francisco?",
|
||||
responses_api_request={},
|
||||
custom_llm_provider="anthropic",
|
||||
litellm_metadata={},
|
||||
)
|
||||
relayed: Final = tuple(
|
||||
itertools.takewhile(lambda event: not isinstance(event, OutputTextDeltaEvent), iterator)
|
||||
)
|
||||
assert relayed, "the stream must relay at least one chunk before the client goes away"
|
||||
|
||||
partial: Final = iterator.partial_response_on_disconnect()
|
||||
|
||||
assert partial is not None
|
||||
hidden: Final = get_hidden_params(partial.response)
|
||||
assert hidden is not None and hidden["response_cost"] == 0.00032
|
||||
priced: Final = wrapper.logging_obj.response_cost_calculator.call_args.kwargs["result"]
|
||||
assert priced.id == partial.response.id
|
||||
assert (priced.usage.prompt_tokens, priced.usage.completion_tokens) == (
|
||||
partial.response.usage.input_tokens,
|
||||
partial.response.usage.output_tokens,
|
||||
), "the bill is priced on the partial's own usage"
|
||||
|
|
|
|||
|
|
@ -24,7 +24,7 @@ from litellm.exceptions import MidStreamFallbackError
|
|||
from litellm.integrations.custom_logger import CustomLogger
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
|
||||
from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig
|
||||
from litellm.responses.streaming_iterator import(
|
||||
from litellm.responses.streaming_iterator import (
|
||||
CachedResponsesAPIStreamingIterator,
|
||||
MockResponsesAPIStreamingIterator,
|
||||
ResponsesAPIStreamingIterator,
|
||||
|
|
@ -845,25 +845,33 @@ def test_stamp_responses_usage_cost_stamps_computed_cost():
|
|||
|
||||
response = _responses_api_response_with_usage()
|
||||
logging_obj = Mock(spec=LiteLLMLoggingObj)
|
||||
logging_obj.model_call_details = {"custom_llm_provider": "openai"}
|
||||
logging_obj.response_cost_calculator.return_value = 0.000704
|
||||
|
||||
stamp_responses_usage_cost(response, logging_obj)
|
||||
|
||||
assert getattr(response.usage, "cost", None) == pytest.approx(0.000704)
|
||||
logging_obj.response_cost_calculator.assert_called_once_with(result=response)
|
||||
priced = logging_obj.response_cost_calculator.call_args.kwargs["result"]
|
||||
assert priced.usage.prompt_tokens == response.usage.input_tokens, (
|
||||
"priced in the chat usage shape the spend row uses"
|
||||
)
|
||||
|
||||
|
||||
def test_stamp_responses_usage_cost_keeps_provider_reported_cost():
|
||||
def test_stamp_responses_usage_cost_bills_the_spend_row_price_over_the_provider_reported_cost():
|
||||
from litellm.litellm_core_utils.core_helpers import response_cost_in_hidden_params
|
||||
from litellm.responses.streaming_iterator import stamp_responses_usage_cost
|
||||
|
||||
response = _responses_api_response_with_usage()
|
||||
setattr(response.usage, "cost", 0.5)
|
||||
logging_obj = Mock(spec=LiteLLMLoggingObj)
|
||||
logging_obj.model_call_details = {"custom_llm_provider": "openai"}
|
||||
logging_obj.response_cost_calculator.return_value = 0.00032
|
||||
|
||||
stamp_responses_usage_cost(response, logging_obj)
|
||||
|
||||
assert getattr(response.usage, "cost", None) == pytest.approx(0.5)
|
||||
logging_obj.response_cost_calculator.assert_not_called()
|
||||
assert getattr(response.usage, "cost", None) == pytest.approx(0.00032), "the client sees what the row bills"
|
||||
priced = logging_obj.response_cost_calculator.call_args.kwargs["result"]
|
||||
assert response_cost_in_hidden_params(priced) == pytest.approx(0.5), "the reported cost reaches the calculator"
|
||||
|
||||
|
||||
def _unvalidated_response_with_dict_usage(usage: dict) -> ResponsesAPIResponse:
|
||||
|
|
@ -881,6 +889,7 @@ def _unvalidated_response_with_dict_usage(usage: dict) -> ResponsesAPIResponse:
|
|||
|
||||
def test_stamp_responses_usage_cost_keeps_provider_cost_from_dict_usage():
|
||||
from litellm.responses.streaming_iterator import stamp_responses_usage_cost
|
||||
|
||||
response = _unvalidated_response_with_dict_usage(
|
||||
{
|
||||
"input_tokens": 29,
|
||||
|
|
@ -891,26 +900,29 @@ def test_stamp_responses_usage_cost_keeps_provider_cost_from_dict_usage():
|
|||
}
|
||||
)
|
||||
logging_obj = Mock(spec=LiteLLMLoggingObj)
|
||||
logging_obj.model_call_details = {"custom_llm_provider": "openai"}
|
||||
logging_obj.response_cost_calculator.return_value = 3e-05
|
||||
|
||||
stamp_responses_usage_cost(response, logging_obj)
|
||||
|
||||
assert isinstance(response.usage, ResponseAPIUsage)
|
||||
assert response.usage.cost == pytest.approx(3e-05)
|
||||
assert response.usage.output_tokens_details.reasoning_tokens == 117
|
||||
logging_obj.response_cost_calculator.assert_not_called()
|
||||
|
||||
|
||||
def test_stamp_responses_usage_cost_computes_cost_for_dict_usage_without_cost():
|
||||
from litellm.responses.streaming_iterator import stamp_responses_usage_cost
|
||||
|
||||
response = _unvalidated_response_with_dict_usage({"input_tokens": 29, "output_tokens": 120, "total_tokens": 149})
|
||||
logging_obj = Mock(spec=LiteLLMLoggingObj)
|
||||
logging_obj.model_call_details = {"custom_llm_provider": "openai"}
|
||||
logging_obj.response_cost_calculator.return_value = 0.000704
|
||||
|
||||
stamp_responses_usage_cost(response, logging_obj)
|
||||
|
||||
assert isinstance(response.usage, ResponseAPIUsage)
|
||||
assert response.usage.cost == pytest.approx(0.000704)
|
||||
logging_obj.response_cost_calculator.assert_called_once_with(result=response)
|
||||
logging_obj.response_cost_calculator.assert_called_once()
|
||||
|
||||
|
||||
def test_stamp_responses_usage_cost_survives_calculator_failure():
|
||||
|
|
@ -918,6 +930,7 @@ def test_stamp_responses_usage_cost_survives_calculator_failure():
|
|||
|
||||
response = _responses_api_response_with_usage()
|
||||
logging_obj = Mock(spec=LiteLLMLoggingObj)
|
||||
logging_obj.model_call_details = {"custom_llm_provider": "openai"}
|
||||
logging_obj.response_cost_calculator.side_effect = RuntimeError("cost map unavailable")
|
||||
|
||||
stamp_responses_usage_cost(response, logging_obj)
|
||||
|
|
@ -1207,9 +1220,7 @@ async def test_completed_event_survives_a_failing_usage_estimate():
|
|||
falls back to usage None."""
|
||||
malformed_input: Final = [{"type": "message", "role": "user", "content": 42}]
|
||||
with pytest.raises(ValueError, match="Invalid content type"):
|
||||
_estimate_usage_from_text(
|
||||
"gpt-4o-mini", malformed_input, {"input": malformed_input}, "hello world"
|
||||
)
|
||||
_estimate_usage_from_text("gpt-4o-mini", malformed_input, {"input": malformed_input}, "hello world")
|
||||
|
||||
response = _responses_api_response_without_usage()
|
||||
iterator = _make_iterator(
|
||||
|
|
@ -1305,7 +1316,14 @@ async def test_completed_event_with_a_dict_response_is_typed_and_billed():
|
|||
assert usage.input_tokens > 0
|
||||
assert usage.output_tokens > 0
|
||||
assert usage.cost == pytest.approx(0.000704)
|
||||
logging_obj.response_cost_calculator.assert_any_call(result=completed_response)
|
||||
priced: Final = [
|
||||
call.kwargs["result"]
|
||||
for call in logging_obj.response_cost_calculator.call_args_list
|
||||
if isinstance(call.kwargs.get("result"), ResponsesAPIResponse)
|
||||
]
|
||||
assert [(r.id, r.usage.prompt_tokens) for r in priced] == [(completed_response.id, usage.input_tokens)], (
|
||||
"priced once, in the chat usage shape the spend row uses"
|
||||
)
|
||||
|
||||
|
||||
def test_billed_terminal_response_keeps_a_response_that_already_has_usage():
|
||||
|
|
@ -1375,6 +1393,7 @@ def _vcr_outcome_gate(request, vcr):
|
|||
yield
|
||||
record_vcr_outcome(request, vcr)
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def setup_and_teardown():
|
||||
"""
|
||||
|
|
@ -1392,6 +1411,7 @@ def setup_and_teardown():
|
|||
loop.close()
|
||||
asyncio.set_event_loop(None)
|
||||
|
||||
|
||||
class _FakeLoggingObj:
|
||||
def __init__(self):
|
||||
self.success_calls = 0
|
||||
|
|
@ -1432,10 +1452,12 @@ class _FakeLoggingObj:
|
|||
self.completion_start_time = completion_start_time
|
||||
self.model_call_details["completion_start_time"] = completion_start_time
|
||||
|
||||
|
||||
async def _await_tasks_spawned_since(existing: set["asyncio.Task[object]"]) -> None:
|
||||
spawned: Final = asyncio.all_tasks() - existing
|
||||
await asyncio.wait_for(asyncio.gather(*spawned, return_exceptions=True), timeout=5)
|
||||
|
||||
|
||||
def _make_completed_response(response_id: str = "resp_test") -> ResponseCompletedEvent:
|
||||
return ResponseCompletedEvent(
|
||||
type=ResponsesAPIStreamEvents.RESPONSE_COMPLETED,
|
||||
|
|
@ -1463,6 +1485,7 @@ def _make_completed_response(response_id: str = "resp_test") -> ResponseComplete
|
|||
),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
@pytest.mark.asyncio
|
||||
async def test_log_background_task_failure_logs_task_exceptions(monkeypatch):
|
||||
|
|
@ -1485,6 +1508,7 @@ async def test_log_background_task_failure_logs_task_exceptions(monkeypatch):
|
|||
task.exception(),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
@pytest.mark.asyncio
|
||||
async def test_log_background_task_failure_ignores_cancelled_tasks(monkeypatch):
|
||||
|
|
@ -1500,6 +1524,7 @@ async def test_log_background_task_failure_ignores_cancelled_tasks(monkeypatch):
|
|||
|
||||
error_logger.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
def test_content_part_done_event_supports_refusal_and_reasoning_text():
|
||||
refusal_event = streaming_module._build_content_part_done_event(
|
||||
|
|
@ -1527,6 +1552,7 @@ def test_content_part_done_event_supports_refusal_and_reasoning_text():
|
|||
assert reasoning_event.part.reasoning == "because"
|
||||
assert unsupported_event is None
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
def test_dump_response_object_handles_model_and_unknown_values():
|
||||
response = ResponsesAPIResponse(
|
||||
|
|
@ -1542,6 +1568,7 @@ def test_dump_response_object_handles_model_and_unknown_values():
|
|||
assert streaming_module._dump_response_object({"type": "message"}) == {"type": "message"}
|
||||
assert streaming_module._dump_response_object(object()) == {}
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
@pytest.mark.asyncio
|
||||
async def test_responses_streaming_triggers_hooks(monkeypatch):
|
||||
|
|
@ -1598,6 +1625,7 @@ async def test_responses_streaming_triggers_hooks(monkeypatch):
|
|||
assert seen["request_data"].get("litellm_params") is not None
|
||||
assert seen["call_type"] == CallTypes.responses
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
@pytest.mark.asyncio
|
||||
async def test_responses_streaming_calls_post_streaming_deployment_hook(monkeypatch):
|
||||
|
|
@ -1638,6 +1666,7 @@ async def test_responses_streaming_calls_post_streaming_deployment_hook(monkeypa
|
|||
# reset callbacks
|
||||
litellm.callbacks = original_callbacks
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
@pytest.mark.asyncio
|
||||
async def test_responses_streaming_failure_triggers_failure_handlers():
|
||||
|
|
@ -1667,6 +1696,7 @@ async def test_responses_streaming_failure_triggers_failure_handlers():
|
|||
assert logging_obj.failure_calls >= 1
|
||||
assert logging_obj.async_failure_calls >= 1
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
def test_process_chunk_requires_provider_config():
|
||||
iterator = ResponsesAPIStreamingIterator(
|
||||
|
|
@ -1681,6 +1711,7 @@ def test_process_chunk_requires_provider_config():
|
|||
with pytest.raises(ValueError, match="responses_api_provider_config is required"):
|
||||
iterator._process_chunk(json.dumps({"type": "response.completed"}))
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
def test_process_chunk_wraps_encrypted_content_with_model_id():
|
||||
openai_types = streaming_module._get_openai_response_types()
|
||||
|
|
@ -1715,6 +1746,7 @@ def test_process_chunk_wraps_encrypted_content_with_model_id():
|
|||
assert event.item.encrypted_content.startswith("litellm_enc:")
|
||||
assert event.item.encrypted_content.endswith(";ciphertext")
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
def test_process_chunk_completed_response_updates_id_and_usage_cost(monkeypatch):
|
||||
original_include_cost = litellm.include_cost_in_streaming_usage
|
||||
|
|
@ -1768,6 +1800,7 @@ def test_process_chunk_completed_response_updates_id_and_usage_cost(monkeypatch)
|
|||
assert event.response.usage.cost == 1.23
|
||||
completion_handler.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
def test_process_chunk_failed_response_triggers_failure_logging(monkeypatch):
|
||||
openai_types = streaming_module._get_openai_response_types()
|
||||
|
|
@ -1803,6 +1836,7 @@ def test_process_chunk_failed_response_triggers_failure_logging(monkeypatch):
|
|||
assert iterator.completed_response is event
|
||||
failure_handler.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_logging_failed_response_uses_response_error_message():
|
||||
|
|
@ -1835,6 +1869,7 @@ async def test_handle_logging_failed_response_uses_response_error_message():
|
|||
assert logging_obj.failure_calls == 1
|
||||
assert logging_obj.async_failure_calls == 1
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
def test_process_chunk_returns_none_for_invalid_json_and_non_dict_payload():
|
||||
class _NoopConfig:
|
||||
|
|
@ -1853,6 +1888,7 @@ def test_process_chunk_returns_none_for_invalid_json_and_non_dict_payload():
|
|||
assert iterator._process_chunk("not-json") is None
|
||||
assert iterator._process_chunk(json.dumps(["not", "a", "dict"])) is None
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
def test_process_chunk_cost_annotation_failure_is_nonfatal(monkeypatch):
|
||||
original_include_cost = litellm.include_cost_in_streaming_usage
|
||||
|
|
@ -1900,6 +1936,7 @@ def test_process_chunk_cost_annotation_failure_is_nonfatal(monkeypatch):
|
|||
assert event.response.usage.cost is None
|
||||
completion_handler.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
def test_get_completed_response_object_accepts_direct_response():
|
||||
logging_obj = _FakeLoggingObj()
|
||||
|
|
@ -1916,6 +1953,7 @@ def test_get_completed_response_object_accepts_direct_response():
|
|||
|
||||
assert iterator._get_completed_response_object() is direct_response
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
@pytest.mark.asyncio
|
||||
async def test_responses_streaming_completed_event_persists_async_cache():
|
||||
|
|
@ -1964,6 +2002,7 @@ async def test_responses_streaming_completed_event_persists_async_cache():
|
|||
assert json.loads(litellm.cache.async_add_cache.call_args.args[0])["id"] == iterator.completed_response.response.id
|
||||
litellm.cache = original_cache
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
def test_responses_streaming_completed_event_persists_sync_cache():
|
||||
logging_obj = _FakeLoggingObj()
|
||||
|
|
@ -2009,6 +2048,7 @@ def test_responses_streaming_completed_event_persists_sync_cache():
|
|||
assert json.loads(litellm.cache.add_cache.call_args.args[0])["id"] == iterator.completed_response.response.id
|
||||
litellm.cache = original_cache
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
def test_log_completed_response_sync_direct_path(monkeypatch):
|
||||
hook_calls = {"post_call": 0, "metadata": 0}
|
||||
|
|
@ -2050,6 +2090,7 @@ def test_log_completed_response_sync_direct_path(monkeypatch):
|
|||
assert hook_calls["post_call"] == 1
|
||||
assert hook_calls["metadata"] == 1
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
def test_log_completed_response_falls_back_when_model_validate_fails(monkeypatch):
|
||||
class _BadSerializableResponse:
|
||||
|
|
@ -2079,6 +2120,7 @@ def test_log_completed_response_falls_back_when_model_validate_fails(monkeypatch
|
|||
assert logging_obj.success_calls == 1
|
||||
assert logging_obj.async_success_calls == 1
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
@pytest.mark.parametrize(
|
||||
"scenario",
|
||||
|
|
@ -2143,6 +2185,7 @@ def test_persist_completed_response_to_cache_guard_branches(monkeypatch, scenari
|
|||
expected_cached_flag = scenario == "already_cached"
|
||||
assert iterator._completed_response_cached is expected_cached_flag
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
def test_build_synthetic_response_events_covers_annotations_function_calls_and_refusals():
|
||||
original_include_cost = litellm.include_cost_in_streaming_usage
|
||||
|
|
@ -2201,6 +2244,7 @@ def test_build_synthetic_response_events_covers_annotations_function_calls_and_r
|
|||
assert "response.function_call_arguments.done" in event_types
|
||||
assert event_types[-1] == "response.completed"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
@pytest.mark.asyncio
|
||||
async def test_mock_responses_streaming_iterator_async_iteration_logs_completion(
|
||||
|
|
@ -2251,6 +2295,7 @@ async def test_mock_responses_streaming_iterator_async_iteration_logs_completion
|
|||
assert hook_calls["post_call"] == 1
|
||||
assert hook_calls["metadata"] == 1
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
def test_mock_responses_streaming_iterator_sync_iteration_logs_completion(monkeypatch):
|
||||
hook_calls = {"post_call": 0, "metadata": 0}
|
||||
|
|
@ -2296,6 +2341,7 @@ def test_mock_responses_streaming_iterator_sync_iteration_logs_completion(monkey
|
|||
assert hook_calls["post_call"] == 1
|
||||
assert hook_calls["metadata"] == 1
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
@pytest.mark.asyncio
|
||||
async def test_cached_responses_stream_async_hit_triggers_success_callbacks(
|
||||
|
|
@ -2355,6 +2401,7 @@ async def test_cached_responses_stream_async_hit_triggers_success_callbacks(
|
|||
litellm.cache.add_cache.assert_not_called()
|
||||
litellm.cache = original_cache
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
|
||||
def test_cached_responses_stream_sync_hit_triggers_success_callbacks(monkeypatch):
|
||||
hook_calls = {"post_call": 0, "metadata": 0}
|
||||
|
|
@ -2453,7 +2500,9 @@ def test_faked_stream_bridged_to_chat_sends_tool_arguments_once():
|
|||
tool_call_deltas: Final = tuple(
|
||||
tool_call
|
||||
for chunk in chunks
|
||||
for tool_call in (chunk.choices[0].delta.tool_calls or ()) # comprehension-ok: flattening each chunk's tool calls
|
||||
for tool_call in (
|
||||
chunk.choices[0].delta.tool_calls or ()
|
||||
) # comprehension-ok: flattening each chunk's tool calls
|
||||
)
|
||||
joined_arguments: Final = "".join(tool_call.function.arguments or "" for tool_call in tool_call_deltas)
|
||||
joined_text: Final = "".join(chunk.choices[0].delta.content or "" for chunk in chunks)
|
||||
|
|
@ -2509,7 +2558,6 @@ def test_faked_stream_opens_items_in_progress_and_empty_like_a_real_stream():
|
|||
assert [part.text for part in added_parts] == [""]
|
||||
|
||||
|
||||
|
||||
def _answer_with_reasoning_refusal_and_custom_tool_call() -> ResponsesAPIResponse:
|
||||
return ResponsesAPIResponse(
|
||||
id="resp_synthetic",
|
||||
|
|
@ -2643,7 +2691,9 @@ async def test_shutdown_drain_waits_for_a_pending_logging_task():
|
|||
|
||||
@pytest.mark.asyncio
|
||||
async def test_shutdown_drain_gives_up_on_a_stuck_logging_task_at_its_timeout():
|
||||
stuck: Final = streaming_module.spawn_logging_task(asyncio.get_running_loop(), asyncio.Event().wait(), task_name="stuck")
|
||||
stuck: Final = streaming_module.spawn_logging_task(
|
||||
asyncio.get_running_loop(), asyncio.Event().wait(), task_name="stuck"
|
||||
)
|
||||
|
||||
await asyncio.wait_for(streaming_module.drain_logging_tasks(timeout=0.05), timeout=1)
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue