mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-18 04:44:37 +00:00
fix(proxy): bill failed WebSocket turns and refund only the reserved window on disconnect
A failed turn's reported usage now counts toward the key's TPM and the project ITPM/OTPM scopes the same way a success does, and closing the socket hands back only the estimate still unused in the window that holds it, so a window that already rolled is left alone. An id-less error event is booked against the newest pending turn, the one the backend rejected at parse time, and a turn that nests its input under response logs its user message. A connection's one budget reservation goes to the first turn that settles it, every later turn charging its cost directly.
This commit is contained in:
mateo-berri
parent
d2a342f6ab
commit
3a8953462b
9 changed files with 593 additions and 30 deletions
|
|
@ -658,6 +658,7 @@ class RequestRateLimiterStash:
|
|||
reserved_tokens: int = 0
|
||||
reserved_model: RateLimitedModel | None = None
|
||||
reserved_scopes: frozenset[tuple[str, str]] = field(default_factory=frozenset)
|
||||
reserved_window_identities: frozenset[tuple[str, str, Literal["redis", "local"]]] = field(default_factory=frozenset)
|
||||
itpm_reserved_tokens: int = 0
|
||||
itpm_reserved_scopes: frozenset[tuple[str, str]] = field(default_factory=frozenset)
|
||||
itpm_reserved_window_identities: frozenset[tuple[str, str, Literal["redis", "local"]]] = field(
|
||||
|
|
@ -671,6 +672,7 @@ class RequestRateLimiterStash:
|
|||
batch_enqueued_reservation: BatchEnqueuedTokenReservation | None = None
|
||||
batch_tpd_refund_ops: tuple[ReservationAwareIncrementOperation, ...] = ()
|
||||
reservation_released: bool = False
|
||||
websocket_connection: bool = False
|
||||
tpm_limited_tags: frozenset[str] = field(default_factory=frozenset)
|
||||
|
||||
|
||||
|
|
@ -4043,6 +4045,7 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
|
|||
verbose_proxy_logger.debug("Inside Rate Limit Pre-Call Hook")
|
||||
|
||||
stash: Final = claim_request_stash_for_data(data)
|
||||
stash.websocket_connection = call_type == CallTypes.aresponses_websocket.value
|
||||
|
||||
#########################################################
|
||||
# Check if the call type has a specific rate limiter
|
||||
|
|
@ -4225,6 +4228,7 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
|
|||
if d["key"] not in (PROJECT_ITPM_DESCRIPTOR_KEY, PROJECT_OTPM_DESCRIPTOR_KEY)
|
||||
and (d.get("rate_limit") or {}).get("tokens_per_unit") is not None
|
||||
)
|
||||
stash.reserved_window_identities = frozenset(tpm_response.get("reservation_windows", frozenset()))
|
||||
tpm_reservation_scopes = tuple( # rebind-ok: record successful reservation scopes
|
||||
stash.reserved_scopes
|
||||
)
|
||||
|
|
@ -4719,7 +4723,7 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
|
|||
return ()
|
||||
stash: Final = get_request_stash_for_call(_call_id_from_callback_kwargs(kwargs))
|
||||
if stash is None:
|
||||
return ()
|
||||
return self._build_websocket_turn_io_ops(kwargs, response_obj)
|
||||
|
||||
itpm_reserved: Final = stash.itpm_reserved_tokens
|
||||
otpm_reserved: Final = stash.otpm_reserved_tokens
|
||||
|
|
@ -4784,9 +4788,68 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
|
|||
)
|
||||
return tuple((*itpm_ops, *otpm_ops))
|
||||
|
||||
def _websocket_turn_usage(self, kwargs: Mapping[str, object], response_obj: object) -> tuple[int, int, bool]:
|
||||
response_usage: Final = self._resolve_io_token_reconcile_usage(response_obj)
|
||||
if response_usage[2]:
|
||||
return response_usage
|
||||
return self._resolve_io_token_reconcile_usage(kwargs.get("combined_usage_object"))
|
||||
|
||||
def _build_websocket_turn_io_ops(
|
||||
self,
|
||||
kwargs: Mapping[str, object],
|
||||
response_obj: object,
|
||||
) -> Sequence[RedisPipelineIncrementOperation]:
|
||||
"""
|
||||
A native Responses WebSocket turn logs under its own call id, so no
|
||||
stash is its own. The connection's stash, inherited through the
|
||||
context, names the project ITPM/OTPM scopes the turn's usage counts
|
||||
toward; the connection's reservation itself is refunded at disconnect.
|
||||
"""
|
||||
connection: Final = get_request_stash()
|
||||
if connection is None or not connection.websocket_connection:
|
||||
return []
|
||||
billable_input, completion_tokens, usage_resolved = self._websocket_turn_usage(kwargs, response_obj)
|
||||
if not usage_resolved:
|
||||
return []
|
||||
return self._build_reservation_aware_tpm_ops(
|
||||
targets=tuple(connection.itpm_reserved_scopes),
|
||||
reserved_scopes=frozenset(),
|
||||
actual_tokens=billable_input,
|
||||
reserved_tokens=0,
|
||||
) + self._build_reservation_aware_tpm_ops(
|
||||
targets=tuple(connection.otpm_reserved_scopes),
|
||||
reserved_scopes=frozenset(),
|
||||
actual_tokens=completion_tokens,
|
||||
reserved_tokens=0,
|
||||
)
|
||||
|
||||
async def _charge_websocket_turn_usage(
|
||||
self,
|
||||
kwargs: Mapping[str, object],
|
||||
response_obj: object,
|
||||
parent_otel_span: Span | None,
|
||||
) -> None:
|
||||
"""A failed turn still consumed the usage its failure event reports; count it the way a success does."""
|
||||
connection: Final = get_request_stash()
|
||||
if connection is None or not connection.websocket_connection:
|
||||
return
|
||||
pipeline_operations: Final = [
|
||||
*self._build_success_event_pipeline_operations(
|
||||
kwargs=dict(kwargs),
|
||||
response_obj=response_obj,
|
||||
rate_limit_type=self.get_rate_limit_type(),
|
||||
),
|
||||
*self._build_websocket_turn_io_ops(kwargs, response_obj),
|
||||
]
|
||||
if pipeline_operations:
|
||||
await self.async_increment_tokens_with_ttl_preservation(
|
||||
pipeline_operations=pipeline_operations,
|
||||
parent_otel_span=parent_otel_span,
|
||||
)
|
||||
|
||||
def _collect_tpm_scope_targets(
|
||||
self,
|
||||
standard_logging_metadata: dict[str, Any],
|
||||
standard_logging_metadata: Mapping[str, Any],
|
||||
kwargs: object,
|
||||
model_group: str | None,
|
||||
tpm_limited_tags: Set[str] = frozenset(),
|
||||
|
|
@ -5172,7 +5235,9 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
|
|||
)
|
||||
|
||||
@with_service_target("rate_limits")
|
||||
async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time) -> None:
|
||||
async def async_log_failure_event(
|
||||
self, kwargs: object, response_obj: object, start_time: object, end_time: object
|
||||
) -> None:
|
||||
"""
|
||||
On failure: decrement max_parallel_requests and refund the upfront
|
||||
TPM reservation only against the scopes the reservation actually
|
||||
|
|
@ -5186,12 +5251,15 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
|
|||
)
|
||||
|
||||
try:
|
||||
litellm_parent_otel_span: Final[Span | None] = get_parent_otel_span_from_kwargs(kwargs)
|
||||
failure_kwargs: Final = dict(_REQUEST_RATE_LIMIT_DATA.validate_python(kwargs))
|
||||
litellm_parent_otel_span: Final[Span | None] = get_parent_otel_span_from_kwargs(failure_kwargs)
|
||||
|
||||
pipeline_operations: Final[list[RedisPipelineIncrementOperation]] = []
|
||||
|
||||
stash: Final = get_request_stash_for_call(_call_id_from_callback_kwargs(kwargs))
|
||||
stash: Final = get_request_stash_for_call(_call_id_from_callback_kwargs(failure_kwargs))
|
||||
await self._release_stashed_parallel_slot(stash, litellm_parent_otel_span, in_logging_callback=True)
|
||||
if stash is None:
|
||||
await self._charge_websocket_turn_usage(failure_kwargs, response_obj, litellm_parent_otel_span)
|
||||
|
||||
# Skip the reservation refund if async_post_call_failure_hook
|
||||
# already released it (proxy-level rejection that also bubbles up
|
||||
|
|
@ -5202,7 +5270,7 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
|
|||
if stash is None or stash.reservation_released
|
||||
else (stash.reserved_tokens, stash.itpm_reserved_tokens, stash.otpm_reserved_tokens)
|
||||
)
|
||||
tpm_actual, itpm_actual, otpm_actual = self._recovered_partial_usage_tokens(kwargs)
|
||||
tpm_actual, itpm_actual, otpm_actual = self._recovered_partial_usage_tokens(failure_kwargs)
|
||||
|
||||
if stash is not None and reserved_tokens > 0:
|
||||
verbose_proxy_logger.debug(
|
||||
|
|
@ -5297,19 +5365,26 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
|
|||
TTL prunes it. The stashed acquisition's presence (not the key
|
||||
object's current max_parallel_requests configuration, which can
|
||||
change mid-request) decides whether there is anything to release.
|
||||
A TPM, ITPM or OTPM reservation no callback settled is refunded the
|
||||
same way, so a native Responses WebSocket connection, whose turns log
|
||||
under their own call ids, hands back the estimate it reserved.
|
||||
A native Responses WebSocket connection logs its turns under their own
|
||||
call ids, so no callback of its own ever settles the TPM, ITPM or OTPM
|
||||
reservation it made at pre-call; its disconnect refunds that estimate.
|
||||
Every other request keeps its reservation for the success or failure
|
||||
event that settles it, even when the client went away mid-stream.
|
||||
"""
|
||||
stash: Final = get_request_stash()
|
||||
await self._release_stashed_parallel_slot(stash, None)
|
||||
await self._refund_unsettled_reservation(stash)
|
||||
if stash is not None and stash.websocket_connection:
|
||||
await self._refund_unsettled_reservation(stash)
|
||||
|
||||
async def _refund_unsettled_reservation(self, stash: RequestRateLimiterStash | None) -> None:
|
||||
if stash is None or stash.reservation_released:
|
||||
async def _refund_unsettled_reservation(self, stash: RequestRateLimiterStash) -> None:
|
||||
if stash.reservation_released:
|
||||
return
|
||||
stash.reservation_released = True
|
||||
await self._refund_reserved_tokens(scopes=tuple(stash.reserved_scopes), amount=stash.reserved_tokens)
|
||||
await self._refund_reserved_tokens(
|
||||
scopes=tuple(stash.reserved_scopes),
|
||||
amount=stash.reserved_tokens,
|
||||
reservation_windows=stash.reserved_window_identities,
|
||||
)
|
||||
await self._refund_reserved_tokens(
|
||||
scopes=tuple(stash.itpm_reserved_scopes),
|
||||
amount=stash.itpm_reserved_tokens,
|
||||
|
|
|
|||
|
|
@ -4,6 +4,8 @@ from collections.abc import Callable, Mapping, Sequence
|
|||
from datetime import datetime
|
||||
from typing import TYPE_CHECKING, Final, Protocol, cast
|
||||
|
||||
from pydantic import BaseModel, ConfigDict
|
||||
|
||||
import litellm
|
||||
from litellm._logging import verbose_proxy_logger
|
||||
from litellm.batches.batch_utils import batch_cost_is_final
|
||||
|
|
@ -166,14 +168,18 @@ class _ProxyDBLogger(CustomLogger):
|
|||
user_api_key_dict: UserAPIKeyAuth,
|
||||
traceback_str: str | None = None,
|
||||
) -> None:
|
||||
budget_reservation: Final = _claim_budget_reservation(
|
||||
budget_reservation=user_api_key_dict.budget_reservation,
|
||||
litellm_call_id=_RequestCallId.model_validate(request_data).litellm_call_id,
|
||||
)
|
||||
try:
|
||||
await _release_budget_reservation(budget_reservation=user_api_key_dict.budget_reservation)
|
||||
await _release_budget_reservation(budget_reservation=budget_reservation)
|
||||
except Exception:
|
||||
verbose_proxy_logger.exception("Failed to release budget reservation during failure handling")
|
||||
try:
|
||||
await _invalidate_budget_reservation_counters(budget_reservation=user_api_key_dict.budget_reservation)
|
||||
if user_api_key_dict.budget_reservation is not None:
|
||||
user_api_key_dict.budget_reservation["finalized"] = True
|
||||
await _invalidate_budget_reservation_counters(budget_reservation=budget_reservation)
|
||||
if budget_reservation is not None:
|
||||
budget_reservation["finalized"] = True
|
||||
except Exception:
|
||||
verbose_proxy_logger.exception(
|
||||
"Failed to invalidate budget reservation counters after failure release failed"
|
||||
|
|
@ -343,7 +349,10 @@ class _ProxyDBLogger(CustomLogger):
|
|||
resolve_missing_key_identity=str(kwargs.get("call_type")) not in _CAPTURED_IDENTITY_CALL_TYPES,
|
||||
)
|
||||
_write_spend_metadata_to_kwargs(kwargs=kwargs, metadata=metadata)
|
||||
budget_reservation: Final = _get_budget_reservation_from_metadata(metadata=metadata)
|
||||
budget_reservation: Final = _claim_budget_reservation(
|
||||
budget_reservation=_get_budget_reservation_from_metadata(metadata=metadata),
|
||||
litellm_call_id=_RequestCallId.model_validate(kwargs).litellm_call_id,
|
||||
)
|
||||
if (
|
||||
isinstance(completion_response, LiteLLMBatch)
|
||||
and kwargs.get("call_type") == CallTypes.aretrieve_batch.value
|
||||
|
|
@ -615,7 +624,7 @@ class _ProxyDBLogger(CustomLogger):
|
|||
ProxyDBLogger: Final = _ProxyDBLogger
|
||||
|
||||
|
||||
def _write_spend_metadata_to_kwargs(kwargs: dict, metadata: dict) -> None:
|
||||
def _write_spend_metadata_to_kwargs(kwargs: dict, metadata: Mapping[str, object]) -> None:
|
||||
patch = {k: v for k, v in metadata.items() if (k.startswith("user_api_key") or k == "tags") and v is not None}
|
||||
if not patch:
|
||||
return
|
||||
|
|
@ -691,13 +700,27 @@ def _metadata_keys(metadata: object) -> tuple[str, ...]:
|
|||
return tuple(sorted(str(key) for key in metadata))
|
||||
|
||||
|
||||
def _get_budget_reservation_from_metadata(metadata: dict) -> dict | None:
|
||||
def _get_budget_reservation_from_metadata(metadata: Mapping[str, object]) -> dict[str, object] | None:
|
||||
return budget_reservation_from_metadata(metadata)
|
||||
|
||||
|
||||
class _RequestCallId(BaseModel):
|
||||
model_config = ConfigDict(frozen=True)
|
||||
|
||||
litellm_call_id: str | None = None
|
||||
|
||||
|
||||
def _claim_budget_reservation(
|
||||
budget_reservation: dict[str, object] | None, litellm_call_id: str | None
|
||||
) -> dict[str, object] | None:
|
||||
from litellm.proxy.spend_tracking.budget_reservation import claim_budget_reservation
|
||||
|
||||
return claim_budget_reservation(budget_reservation=budget_reservation, litellm_call_id=litellm_call_id)
|
||||
|
||||
|
||||
def _get_request_tags_for_cost_tracking(
|
||||
sl_object: StandardLoggingPayload | None,
|
||||
metadata: dict,
|
||||
metadata: Mapping[str, object],
|
||||
) -> list[str] | None:
|
||||
if sl_object is not None:
|
||||
request_tags: Final = sl_object.get("request_tags", None)
|
||||
|
|
|
|||
|
|
@ -378,6 +378,20 @@ async def reconcile_budget_reservation(
|
|||
return pending
|
||||
|
||||
|
||||
def claim_budget_reservation(
|
||||
budget_reservation: dict[str, object] | None, litellm_call_id: str | None
|
||||
) -> dict[str, object] | None:
|
||||
"""Hand the reservation to the first call that settles it.
|
||||
|
||||
A native Responses WebSocket connection shares its one reservation with every turn it logs, each under its
|
||||
own call id. Two turns settling it at once would both move the counter from the estimate to their own cost,
|
||||
so the first claimant keeps the reservation and every later call sees none and charges its cost directly."""
|
||||
if budget_reservation is None or litellm_call_id is None:
|
||||
return budget_reservation
|
||||
settling_call_id: Final = budget_reservation.setdefault("settling_call_id", litellm_call_id)
|
||||
return budget_reservation if settling_call_id == litellm_call_id else None
|
||||
|
||||
|
||||
def stamp_budget_reservation_actual_cost(budget_reservation: dict | None, actual_cost: float | None) -> None:
|
||||
"""Record that every reserved counter now holds ``actual_cost``, once the adjustments handed back by
|
||||
``reconcile_budget_reservation(apply_consistent=False)`` have been written."""
|
||||
|
|
|
|||
|
|
@ -1871,18 +1871,31 @@ def _logged_user_messages(client_frame: object) -> tuple[dict[str, object], ...]
|
|||
frame: Final = _json_object_or_none(client_frame)
|
||||
if frame is None or frame.get("type") != "response.create":
|
||||
return ()
|
||||
input_items: Final = frame.get("input", [])
|
||||
nested_response: Final = frame.get("response")
|
||||
input_items: Final = frame.get(
|
||||
"input", nested_response.get("input", []) if _is_json_object(nested_response) else []
|
||||
)
|
||||
if isinstance(input_items, str):
|
||||
return ({"role": "user", "content": input_items},)
|
||||
texts: Final = chain.from_iterable(_user_texts_in_input_item(item) for item in _json_array_or_empty(input_items))
|
||||
return tuple({"role": "user", "content": text} for text in texts)
|
||||
|
||||
|
||||
def _ws_event_response_id(event: object) -> str | None:
|
||||
response: Final = event.get("response") if _is_json_object(event) else None
|
||||
response_id: Final = response.get("id") if _is_json_object(response) else None
|
||||
return response_id if isinstance(response_id, str) else None
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _ResponsesWsTurn:
|
||||
logging_obj: LiteLLMLoggingObj
|
||||
events: tuple[_MutableJsonObject, ...] = ()
|
||||
|
||||
@property
|
||||
def response_id(self) -> str | None:
|
||||
return next(filter(None, map(_ws_event_response_id, self.events)), None)
|
||||
|
||||
def with_event(self, event: _MutableJsonObject) -> _ResponsesWsTurn:
|
||||
return _ResponsesWsTurn(logging_obj=self.logging_obj, events=(*self.events, event))
|
||||
|
||||
|
|
@ -2010,12 +2023,33 @@ class ResponsesWebSocketStreaming:
|
|||
return
|
||||
if not self._open_turns:
|
||||
self._start_turn(None)
|
||||
turn: Final = self._open_turns[0].with_event(event_obj)
|
||||
index: Final = self._turn_index_for_event(event_obj)
|
||||
turn: Final = self._open_turns[index].with_event(event_obj)
|
||||
other_turns: Final = (*self._open_turns[:index], *self._open_turns[index + 1 :])
|
||||
if event_obj.get("type") in _RESPONSES_WS_TERMINAL_EVENT_TYPES:
|
||||
self._open_turns = self._open_turns[1:]
|
||||
self._open_turns = other_turns
|
||||
self._log_turn(turn)
|
||||
return
|
||||
self._open_turns = (turn, *self._open_turns[1:])
|
||||
self._open_turns = (*self._open_turns[:index], turn, *self._open_turns[index + 1 :])
|
||||
|
||||
def _turn_index_for_event(self, event_obj: Mapping[str, object]) -> int:
|
||||
"""
|
||||
An event naming a response id belongs to the turn whose ``response.created``
|
||||
carried it. The backend answers accepted turns strictly in order, so an
|
||||
id-less event goes to the oldest turn still waiting for its ``response.created``,
|
||||
except an id-less ``error``: that is the backend rejecting the newest pending
|
||||
``response.create`` at parse time while an earlier turn is still streaming.
|
||||
"""
|
||||
response_id: Final = _ws_event_response_id(event_obj)
|
||||
turn_ids: Final = tuple(turn.response_id for turn in self._open_turns)
|
||||
if response_id is not None and response_id in turn_ids:
|
||||
return turn_ids.index(response_id)
|
||||
pending: Final = tuple(index for index, turn_id in enumerate(turn_ids) if turn_id is None)
|
||||
if not pending:
|
||||
return 0
|
||||
if response_id is None and event_obj.get("type") == "error":
|
||||
return pending[-1]
|
||||
return pending[0]
|
||||
|
||||
def _log_open_turns(self) -> None:
|
||||
open_turns: Final = self._open_turns
|
||||
|
|
|
|||
|
|
@ -46,6 +46,7 @@ from litellm.types.llms.openai import ResponsesAPIResponse
|
|||
from litellm.types.mcp import MCPPreCallRequestObject, MCPTransport
|
||||
from litellm.types.mcp_server.mcp_server_manager import MCPServer
|
||||
from litellm.types.utils import (
|
||||
CallTypes,
|
||||
EmbeddingResponse,
|
||||
ModelResponse,
|
||||
TextCompletionResponse,
|
||||
|
|
@ -4237,7 +4238,10 @@ async def test_release_on_disconnect_works_when_key_config_changed_v3():
|
|||
|
||||
|
||||
async def _admit_with_tpm_reservation(
|
||||
handler: _PROXY_MaxParallelRequestsHandler, local_cache: DualCache, user_api_key_dict: UserAPIKeyAuth
|
||||
handler: _PROXY_MaxParallelRequestsHandler,
|
||||
local_cache: DualCache,
|
||||
user_api_key_dict: UserAPIKeyAuth,
|
||||
call_type: str = "completion",
|
||||
) -> int:
|
||||
await handler.async_pre_call_hook(
|
||||
user_api_key_dict=user_api_key_dict,
|
||||
|
|
@ -4248,7 +4252,7 @@ async def _admit_with_tpm_reservation(
|
|||
"max_tokens": 50,
|
||||
"litellm_call_id": "owner-call-id",
|
||||
},
|
||||
call_type="completion",
|
||||
call_type=call_type,
|
||||
)
|
||||
stash = get_request_stash()
|
||||
assert stash is not None and stash.reserved_tokens > 0
|
||||
|
|
@ -4268,7 +4272,9 @@ async def test_release_on_disconnect_refunds_an_unsettled_tpm_reservation_v3(mon
|
|||
handler = _PROXY_MaxParallelRequestsHandler(internal_usage_cache=InternalUsageCache(local_cache))
|
||||
user_api_key_dict = UserAPIKeyAuth(api_key=_api_key, tpm_limit=10_000, max_parallel_requests=2)
|
||||
tokens_key = handler.create_rate_limit_keys(key="api_key", value=_api_key, rate_limit_type="tokens")
|
||||
reserved = await _admit_with_tpm_reservation(handler, local_cache, user_api_key_dict)
|
||||
reserved = await _admit_with_tpm_reservation(
|
||||
handler, local_cache, user_api_key_dict, CallTypes.aresponses_websocket.value
|
||||
)
|
||||
assert int(await local_cache.async_get_cache(key=tokens_key) or 0) == reserved
|
||||
|
||||
await handler.async_release_max_parallel_requests_on_disconnect(user_api_key_dict)
|
||||
|
|
@ -4289,7 +4295,7 @@ async def test_release_on_disconnect_keeps_a_settled_tpm_reservation_v3(monkeypa
|
|||
handler = _PROXY_MaxParallelRequestsHandler(internal_usage_cache=InternalUsageCache(local_cache))
|
||||
user_api_key_dict = UserAPIKeyAuth(api_key=_api_key, tpm_limit=10_000, max_parallel_requests=2)
|
||||
tokens_key = handler.create_rate_limit_keys(key="api_key", value=_api_key, rate_limit_type="tokens")
|
||||
await _admit_with_tpm_reservation(handler, local_cache, user_api_key_dict)
|
||||
await _admit_with_tpm_reservation(handler, local_cache, user_api_key_dict, CallTypes.aresponses_websocket.value)
|
||||
|
||||
await handler.async_log_success_event(
|
||||
kwargs={
|
||||
|
|
@ -7938,3 +7944,117 @@ async def test_a2a_url_target_owns_invocation_fee_and_request_limit(
|
|||
await _rpm_request(limiter, cache, auth, "a2a/cheap")
|
||||
assert denied.value.status_code == 429
|
||||
assert "expensive" in str(denied.value.detail)
|
||||
|
||||
|
||||
def _websocket_turn_kwargs(call_id: str, api_key_hash: str, usage: Usage) -> dict[str, object]:
|
||||
return {
|
||||
"litellm_call_id": call_id,
|
||||
"model": "gpt-4o-mini",
|
||||
"combined_usage_object": usage,
|
||||
"litellm_params": {"metadata": {"user_api_key": api_key_hash}},
|
||||
"standard_logging_object": {"metadata": {"user_api_key_hash": api_key_hash}},
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_release_on_disconnect_keeps_a_plain_request_tpm_reservation_v3(monkeypatch):
|
||||
"""
|
||||
A cancelled HTTP stream keeps its reservation for the success or failure event
|
||||
that still settles it; only a WebSocket connection, whose turns log under their
|
||||
own call ids, hands its estimate back at disconnect.
|
||||
"""
|
||||
monkeypatch.delenv("LITELLM_TPM_TOKEN_RESERVATION_ENABLED", raising=False)
|
||||
_api_key = hash_token("sk-http-cancelled-stream")
|
||||
local_cache = DualCache()
|
||||
handler = _PROXY_MaxParallelRequestsHandler(internal_usage_cache=InternalUsageCache(local_cache))
|
||||
user_api_key_dict = UserAPIKeyAuth(api_key=_api_key, tpm_limit=10_000, max_parallel_requests=2)
|
||||
tokens_key = handler.create_rate_limit_keys(key="api_key", value=_api_key, rate_limit_type="tokens")
|
||||
reserved = await _admit_with_tpm_reservation(handler, local_cache, user_api_key_dict, "completion")
|
||||
|
||||
await handler.async_release_max_parallel_requests_on_disconnect(user_api_key_dict)
|
||||
|
||||
assert int(await local_cache.async_get_cache(key=tokens_key) or 0) == reserved
|
||||
assert get_request_stash().reservation_released is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_release_on_disconnect_skips_a_tpm_window_that_rolled_v3(monkeypatch):
|
||||
"""
|
||||
The TPM window can roll while a WebSocket connection stays open. The disconnect
|
||||
refund belongs to the window the connection reserved against, so it must leave
|
||||
the window that is live now untouched instead of handing it tokens it never took.
|
||||
"""
|
||||
monkeypatch.delenv("LITELLM_TPM_TOKEN_RESERVATION_ENABLED", raising=False)
|
||||
_api_key = hash_token("sk-ws-rolled-window")
|
||||
local_cache = DualCache()
|
||||
handler = _PROXY_MaxParallelRequestsHandler(internal_usage_cache=InternalUsageCache(local_cache))
|
||||
user_api_key_dict = UserAPIKeyAuth(api_key=_api_key, tpm_limit=10_000, max_parallel_requests=2)
|
||||
tokens_key = handler.create_rate_limit_keys(key="api_key", value=_api_key, rate_limit_type="tokens")
|
||||
window_key = handler.create_rate_limit_keys(key="api_key", value=_api_key, rate_limit_type="window")
|
||||
await _admit_with_tpm_reservation(handler, local_cache, user_api_key_dict, CallTypes.aresponses_websocket.value)
|
||||
reserved_window_start = await local_cache.async_get_cache(key=window_key)
|
||||
assert reserved_window_start is not None
|
||||
await local_cache.async_set_cache(key=window_key, value=int(reserved_window_start) + handler.window_size)
|
||||
await local_cache.async_set_cache(key=tokens_key, value=500)
|
||||
|
||||
await handler.async_release_max_parallel_requests_on_disconnect(user_api_key_dict)
|
||||
|
||||
assert int(await local_cache.async_get_cache(key=tokens_key) or 0) == 500
|
||||
assert get_request_stash().reservation_released is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_websocket_turn_failure_counts_its_usage_toward_tpm_v3(monkeypatch):
|
||||
"""
|
||||
A turn that ends in ``response.failed`` still consumed the usage the event
|
||||
reports. It logs under its own call id, so the failure event owns no stash; it
|
||||
must still charge the connection's TPM scopes the way a successful turn does.
|
||||
"""
|
||||
monkeypatch.delenv("LITELLM_TPM_TOKEN_RESERVATION_ENABLED", raising=False)
|
||||
_api_key = hash_token("sk-ws-failed-turn")
|
||||
local_cache = DualCache()
|
||||
handler = _PROXY_MaxParallelRequestsHandler(internal_usage_cache=InternalUsageCache(local_cache))
|
||||
user_api_key_dict = UserAPIKeyAuth(api_key=_api_key, tpm_limit=10_000, max_parallel_requests=2)
|
||||
tokens_key = handler.create_rate_limit_keys(key="api_key", value=_api_key, rate_limit_type="tokens")
|
||||
reserved = await _admit_with_tpm_reservation(
|
||||
handler, local_cache, user_api_key_dict, CallTypes.aresponses_websocket.value
|
||||
)
|
||||
|
||||
await handler.async_log_failure_event(
|
||||
kwargs=_websocket_turn_kwargs(
|
||||
"turn-1", _api_key, Usage(prompt_tokens=12, completion_tokens=5, total_tokens=17)
|
||||
),
|
||||
response_obj=None,
|
||||
start_time=datetime.now(),
|
||||
end_time=datetime.now(),
|
||||
)
|
||||
|
||||
assert int(await local_cache.async_get_cache(key=tokens_key) or 0) == reserved + 17
|
||||
assert get_request_stash().reservation_released is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_failure_event_without_a_stash_charges_nothing_outside_a_websocket_connection_v3(monkeypatch):
|
||||
"""
|
||||
A nested call made inside a plain request (an LLM-judge guardrail) logs under a
|
||||
fresh call id too. Its failure owns no stash and must not touch the request's
|
||||
counters: only a WebSocket connection's turns are charged that way.
|
||||
"""
|
||||
monkeypatch.delenv("LITELLM_TPM_TOKEN_RESERVATION_ENABLED", raising=False)
|
||||
_api_key = hash_token("sk-nested-failed-call")
|
||||
local_cache = DualCache()
|
||||
handler = _PROXY_MaxParallelRequestsHandler(internal_usage_cache=InternalUsageCache(local_cache))
|
||||
user_api_key_dict = UserAPIKeyAuth(api_key=_api_key, tpm_limit=10_000, max_parallel_requests=2)
|
||||
tokens_key = handler.create_rate_limit_keys(key="api_key", value=_api_key, rate_limit_type="tokens")
|
||||
reserved = await _admit_with_tpm_reservation(handler, local_cache, user_api_key_dict, "completion")
|
||||
|
||||
await handler.async_log_failure_event(
|
||||
kwargs=_websocket_turn_kwargs(
|
||||
"nested-call-id", _api_key, Usage(prompt_tokens=12, completion_tokens=5, total_tokens=17)
|
||||
),
|
||||
response_obj=None,
|
||||
start_time=datetime.now(),
|
||||
end_time=datetime.now(),
|
||||
)
|
||||
|
||||
assert int(await local_cache.async_get_cache(key=tokens_key) or 0) == reserved
|
||||
|
|
|
|||
|
|
@ -2972,3 +2972,94 @@ async def test_track_cost_callback_enqueue_emits_no_service_span(): # test-qual
|
|||
|
||||
assert update_database.await_count == 1, "the spend enqueue itself must still run"
|
||||
assert success_hook.await_count == 0, [call.kwargs for call in success_hook.await_args_list]
|
||||
|
||||
|
||||
def _shared_reservation_turn_kwargs(call_id: str, reservation: dict) -> dict:
|
||||
return {
|
||||
"model": "gpt-5.4-mini",
|
||||
"litellm_call_id": call_id,
|
||||
"litellm_params": {
|
||||
"metadata": {"user_api_key": "test-key-hash", "user_api_key_budget_reservation": reservation}
|
||||
},
|
||||
"stream": False,
|
||||
"standard_logging_object": {"response_cost": 0.25, "request_tags": None},
|
||||
}
|
||||
|
||||
|
||||
async def _reservation_handed_to_the_counters(kwargs) -> dict | None:
|
||||
"""The reservation the callback hands the spend counters for one logged call."""
|
||||
logger = ProxyDBLogger()
|
||||
with (
|
||||
patch( # test-quality-ok: callback imports proxy_logging_obj off proxy_server in its body, no seam
|
||||
"litellm.proxy.proxy_server.proxy_logging_obj"
|
||||
) as mock_proxy_logging,
|
||||
patch( # test-quality-ok: the arguments to this call are the boundary under test
|
||||
"litellm.proxy.hooks.proxy_track_cost_callback._update_database_and_spend_counters",
|
||||
new=AsyncMock(),
|
||||
) as mock_update,
|
||||
patch( # test-quality-ok: llm_router is a proxy_server global the callback reads lazily, no seam
|
||||
"litellm.proxy.proxy_server.llm_router", new=_FakeDeploymentLookup({})
|
||||
),
|
||||
):
|
||||
mock_proxy_logging.failed_tracking_alert = AsyncMock()
|
||||
mock_proxy_logging.slack_alerting_instance.customer_spend_alert = AsyncMock()
|
||||
mock_proxy_logging.db_spend_update_writer = MagicMock()
|
||||
mock_proxy_logging.db_spend_update_writer.update_database = AsyncMock()
|
||||
|
||||
await logger._PROXY_track_cost_callback(
|
||||
kwargs=kwargs,
|
||||
completion_response=None,
|
||||
start_time=datetime.now(),
|
||||
end_time=datetime.now(),
|
||||
)
|
||||
|
||||
return mock_update.await_args.kwargs["budget_reservation"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_track_cost_callback_hands_a_shared_reservation_to_the_first_turn_only():
|
||||
"""Every turn of a Responses WebSocket connection carries the connection's one reservation under its
|
||||
own call id. The first turn settles it; a later turn reconciling the same estimate again would
|
||||
under-count the key, so it must charge its cost without the reservation."""
|
||||
reservation = {"reserved_cost": 0.5, "entries": []}
|
||||
|
||||
first_turn = await _reservation_handed_to_the_counters(_shared_reservation_turn_kwargs("turn-1", reservation))
|
||||
second_turn = await _reservation_handed_to_the_counters(_shared_reservation_turn_kwargs("turn-2", reservation))
|
||||
|
||||
assert first_turn is reservation
|
||||
assert second_turn is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_post_call_failure_hook_releases_a_shared_reservation_for_the_first_turn_only():
|
||||
logger = ProxyDBLogger()
|
||||
budget_reservation = {"reserved_cost": 0.5, "entries": []}
|
||||
user_api_key_dict = UserAPIKeyAuth(
|
||||
api_key="test_api_key",
|
||||
request_route="/custom/route",
|
||||
budget_reservation=budget_reservation,
|
||||
)
|
||||
|
||||
with (
|
||||
patch(
|
||||
"litellm.proxy.spend_tracking.budget_reservation.release_budget_reservation",
|
||||
new_callable=AsyncMock,
|
||||
) as mock_release_budget_reservation,
|
||||
patch(
|
||||
"litellm.proxy.db.db_spend_update_writer.DBSpendUpdateWriter.update_database",
|
||||
new_callable=AsyncMock,
|
||||
),
|
||||
):
|
||||
for call_id in ("turn-1", "turn-2"):
|
||||
await logger.async_post_call_failure_hook(
|
||||
request_data={"litellm_call_id": call_id},
|
||||
original_exception=Exception("Test exception"),
|
||||
user_api_key_dict=user_api_key_dict,
|
||||
)
|
||||
|
||||
assert mock_release_budget_reservation.await_count == 1
|
||||
assert (
|
||||
mock_release_budget_reservation.await_args.kwargs["budget_reservation"]
|
||||
is user_api_key_dict.budget_reservation
|
||||
)
|
||||
assert user_api_key_dict.budget_reservation["settling_call_id"] == "turn-1"
|
||||
|
|
|
|||
|
|
@ -42,7 +42,7 @@ from litellm.types.llms.openai import (
|
|||
ResponsesAPIResponse,
|
||||
)
|
||||
from litellm.types.rerank import RerankResponse
|
||||
from litellm.types.utils import ModelResponse, PromptTokensDetailsWrapper, Usage
|
||||
from litellm.types.utils import CallTypes, ModelResponse, PromptTokensDetailsWrapper, Usage
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -3710,3 +3710,85 @@ async def test_the_project_itpm_reservation_counts_the_request_off_the_event_loo
|
|||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v", "-s"])
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_websocket_turn_usage_counts_toward_project_itpm_and_otpm(rate_limiter):
|
||||
"""
|
||||
A native Responses WebSocket connection reserves project ITPM/OTPM at pre-call and
|
||||
its turns log under their own call ids. Every turn's real usage must reach the
|
||||
project counters, success and failure alike, and the disconnect must hand back
|
||||
only the connection's estimate, never the turns' usage.
|
||||
"""
|
||||
handler, cache = rate_limiter
|
||||
api_key = hash_token("sk-ws-project-turns")
|
||||
user_api_key_dict = UserAPIKeyAuth(
|
||||
api_key=api_key,
|
||||
project_id="proj-ws-turns",
|
||||
project_metadata={
|
||||
"model_itpm_limit": {"bedrock_mantle/claude-opus": 1000},
|
||||
"model_otpm_limit": {"bedrock_mantle/claude-opus": 500},
|
||||
},
|
||||
)
|
||||
await handler.async_pre_call_hook(
|
||||
user_api_key_dict=user_api_key_dict,
|
||||
cache=cache,
|
||||
data={
|
||||
"model": "bedrock_mantle/claude-opus",
|
||||
"messages": [{"role": "user", "content": "hello"}],
|
||||
"max_tokens": 50,
|
||||
"litellm_call_id": "connection-call-id",
|
||||
},
|
||||
call_type=CallTypes.aresponses_websocket.value,
|
||||
)
|
||||
stash = get_request_stash()
|
||||
assert stash is not None and stash.itpm_reserved_tokens > 0 and stash.otpm_reserved_tokens > 0
|
||||
(itpm_scope,) = stash.itpm_reserved_scopes
|
||||
(otpm_scope,) = stash.otpm_reserved_scopes
|
||||
itpm_key = handler.create_rate_limit_keys(itpm_scope[0], itpm_scope[1], "tokens")
|
||||
otpm_key = handler.create_rate_limit_keys(otpm_scope[0], otpm_scope[1], "tokens")
|
||||
|
||||
async def counters() -> tuple[int, int]:
|
||||
return (
|
||||
int(await cache.async_get_cache(key=itpm_key) or 0),
|
||||
int(await cache.async_get_cache(key=otpm_key) or 0),
|
||||
)
|
||||
|
||||
def turn_kwargs(call_id: str, usage: Usage | None = None) -> dict[str, object]:
|
||||
return {
|
||||
"litellm_call_id": call_id,
|
||||
"model": "bedrock_mantle/claude-opus",
|
||||
"combined_usage_object": usage,
|
||||
"litellm_params": {"metadata": {"user_api_key": api_key}},
|
||||
"standard_logging_object": {
|
||||
"metadata": {"user_api_key_hash": api_key, "user_api_key_project_id": "proj-ws-turns"}
|
||||
},
|
||||
}
|
||||
|
||||
assert await counters() == (stash.itpm_reserved_tokens, stash.otpm_reserved_tokens)
|
||||
|
||||
await handler.async_log_success_event(
|
||||
kwargs=turn_kwargs("turn-1"),
|
||||
response_obj=ResponsesAPIResponse(
|
||||
id="resp_1",
|
||||
created_at=int(datetime.now().timestamp()),
|
||||
output=[],
|
||||
usage=ResponseAPIUsage(input_tokens=30, output_tokens=7, total_tokens=37),
|
||||
),
|
||||
start_time=datetime.now(),
|
||||
end_time=datetime.now(),
|
||||
)
|
||||
assert await counters() == (stash.itpm_reserved_tokens + 30, stash.otpm_reserved_tokens + 7)
|
||||
|
||||
await handler.async_log_failure_event(
|
||||
kwargs=turn_kwargs("turn-2", Usage(prompt_tokens=10, completion_tokens=3, total_tokens=13)),
|
||||
response_obj=None,
|
||||
start_time=datetime.now(),
|
||||
end_time=datetime.now(),
|
||||
)
|
||||
assert await counters() == (stash.itpm_reserved_tokens + 40, stash.otpm_reserved_tokens + 10)
|
||||
|
||||
await handler.async_release_max_parallel_requests_on_disconnect(user_api_key_dict=user_api_key_dict)
|
||||
|
||||
assert await counters() == (40, 10)
|
||||
assert stash.reservation_released is True
|
||||
|
|
|
|||
|
|
@ -40,6 +40,7 @@ from litellm.proxy.common_utils.user_api_key_cache import (
|
|||
)
|
||||
from litellm.proxy.spend_tracking.budget_reservation import (
|
||||
_get_model_access_group_budget_counters,
|
||||
claim_budget_reservation,
|
||||
estimate_request_max_cost,
|
||||
get_budget_window_start,
|
||||
invalidate_budget_reservation_counters,
|
||||
|
|
@ -3720,3 +3721,63 @@ async def test_unreserved_model_access_group_is_charged_alongside_a_reserved_one
|
|||
assert counter_cache.in_memory_cache.get_cache(
|
||||
key=model_access_group_spend_counter_key("starter")
|
||||
) == pytest.approx(4.2)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_should_charge_both_turns_when_two_calls_settle_one_reservation_at_once(
|
||||
spend_counter_state,
|
||||
):
|
||||
"""A Responses WebSocket connection's turns share its reservation under their own call ids. Only the
|
||||
first claim keeps it; the other turn adds its cost on top instead of settling the same estimate again,
|
||||
which would leave the key counter short by the estimate."""
|
||||
counter_cache, key_cache = spend_counter_state
|
||||
proxy_logging_obj = ProxyLogging(user_api_key_cache=key_cache)
|
||||
valid_token = UserAPIKeyAuth(
|
||||
token="key-budget-shared-turns",
|
||||
spend=0.0,
|
||||
max_budget=1.0,
|
||||
)
|
||||
|
||||
with patch(
|
||||
"litellm.proxy.spend_tracking.budget_reservation.estimate_request_max_cost",
|
||||
return_value=0.6,
|
||||
):
|
||||
reservation = await reserve_budget_for_request(
|
||||
request_body=_request_body(),
|
||||
route="/responses",
|
||||
llm_router=None,
|
||||
valid_token=valid_token,
|
||||
team_object=None,
|
||||
user_object=None,
|
||||
prisma_client=None,
|
||||
user_api_key_cache=key_cache,
|
||||
proxy_logging_obj=proxy_logging_obj,
|
||||
)
|
||||
|
||||
from litellm.proxy.proxy_server import increment_spend_counters
|
||||
|
||||
first_turn = claim_budget_reservation(budget_reservation=reservation, litellm_call_id="turn-1")
|
||||
second_turn = claim_budget_reservation(budget_reservation=reservation, litellm_call_id="turn-2")
|
||||
assert first_turn is reservation
|
||||
assert second_turn is None
|
||||
assert claim_budget_reservation(budget_reservation=reservation, litellm_call_id="turn-1") is reservation
|
||||
|
||||
await asyncio.gather(
|
||||
increment_spend_counters(
|
||||
token="key-budget-shared-turns",
|
||||
team_id="team-without-budget",
|
||||
user_id=None,
|
||||
response_cost=0.2,
|
||||
budget_reservation=first_turn,
|
||||
),
|
||||
increment_spend_counters(
|
||||
token="key-budget-shared-turns",
|
||||
team_id="team-without-budget",
|
||||
user_id=None,
|
||||
response_cost=0.3,
|
||||
budget_reservation=second_turn,
|
||||
),
|
||||
)
|
||||
|
||||
assert counter_cache.in_memory_cache.get_cache(key="spend:key:key-budget-shared-turns") == pytest.approx(0.5)
|
||||
assert counter_cache.in_memory_cache.get_cache(key="spend:team:team-without-budget") == pytest.approx(0.5)
|
||||
|
|
|
|||
|
|
@ -3339,6 +3339,69 @@ class TestNativeWebSocketPerTurnLogging:
|
|||
for turn in recorder.turns:
|
||||
turn.dispatch_failure_handlers.assert_not_awaited()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_turn_whose_input_is_nested_under_response_logs_its_user_message(self):
|
||||
import websockets.exceptions # noqa: F401 (lazy submodule must be importable)
|
||||
|
||||
recorder = _TurnLoggingRecorder()
|
||||
nested_frame = json.dumps(
|
||||
{"type": "response.create", "response": {"model": "gpt-5.4-mini", "input": "nested"}}
|
||||
)
|
||||
client = _ScriptedClient([nested_frame], recorder.logged_turns)
|
||||
backend = _ScriptedBackend([_completed_turn("resp_1", 10, 1)])
|
||||
handler = _make_streaming(
|
||||
websocket=client, backend_ws=backend, logging_obj=recorder.connection, request_data={}
|
||||
)
|
||||
|
||||
await handler.bidirectional_forward()
|
||||
|
||||
assert [turn.model_call_details["messages"] for turn in recorder.turns] == [
|
||||
[{"role": "user", "content": "nested"}]
|
||||
]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_overlapped_turn_rejected_at_parse_time_fails_alone_and_the_streaming_turn_keeps_its_usage(
|
||||
self,
|
||||
):
|
||||
"""OpenAI rejects an unparseable response.create with an id-less error while the earlier turn is
|
||||
still streaming. That error is the second turn's failure, and the first turn's completed event with
|
||||
its usage stays on the first turn."""
|
||||
import websockets.exceptions # noqa: F401 (lazy submodule must be importable)
|
||||
|
||||
recorder = _TurnLoggingRecorder()
|
||||
handler = _make_streaming(
|
||||
websocket=MagicMock(),
|
||||
backend_ws=MagicMock(),
|
||||
logging_obj=recorder.connection,
|
||||
request_data={},
|
||||
authorized_model="gpt-5.4-mini",
|
||||
custom_llm_provider="openai",
|
||||
)
|
||||
created, _delta, completed = _completed_turn("resp_1", 300, 23)
|
||||
rejection = {
|
||||
"type": "error",
|
||||
"status": None,
|
||||
"error": {"type": "invalid_request_error", "code": None, "message": "Invalid 'max_output_tokens'"},
|
||||
}
|
||||
|
||||
handler._record_client_frame(_response_create("first"))
|
||||
handler._record_client_frame(_response_create("second"))
|
||||
for event in (created, rejection, completed):
|
||||
handler._record_backend_event(json.dumps(event))
|
||||
assert await _until(lambda: recorder.logged_turns() == 2)
|
||||
|
||||
first, second = recorder.turns
|
||||
first.dispatch_failure_handlers.assert_not_awaited()
|
||||
first.dispatch_success_handlers.assert_awaited_once()
|
||||
assert [event["type"] for event in first.dispatch_success_handlers.await_args[0][0]] == [
|
||||
"response.created",
|
||||
"response.completed",
|
||||
]
|
||||
second.dispatch_success_handlers.assert_not_awaited()
|
||||
second.dispatch_failure_handlers.assert_awaited_once()
|
||||
assert "max_output_tokens" in str(second.dispatch_failure_handlers.await_args[0][0])
|
||||
assert second.record_partial_usage_for_failure.call_args[0][0].total_tokens == 0
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_failed_turn_is_booked_alone_and_leaves_the_other_turns_successful(self):
|
||||
from unittest.mock import AsyncMock
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue