diff --git a/litellm/cost_calculator.py b/litellm/cost_calculator.py index 06edd198603..41a7ef1ab64 100644 --- a/litellm/cost_calculator.py +++ b/litellm/cost_calculator.py @@ -137,8 +137,6 @@ from litellm.utils import ( TextCompletionResponse, TranscriptionResponse, _cached_get_model_info_helper, - _get_model_info_from_generalization, - _get_potential_model_names, token_counter, ) @@ -926,22 +924,6 @@ def _get_response_model(completion_response: object) -> str | None: return None -def _prices_only_via_capability_rule(model: str | None, custom_llm_provider: str | None) -> bool: - if model is None or model in litellm.model_cost or f"{custom_llm_provider}/{model}" in litellm.model_cost: - return False - try: - return ( - _get_model_info_from_generalization( - model=model, - potential_model_names=_get_potential_model_names(model=model, custom_llm_provider=custom_llm_provider), - custom_llm_provider=custom_llm_provider, - ) - is not None - ) - except Exception: - return False - - _GEMINI_TRAFFIC_TYPE_TO_SERVICE_TIER: Final[dict] = { # ON_DEMAND_PRIORITY maps to "priority" — selects input_cost_per_token_priority, etc. "ON_DEMAND_PRIORITY": "priority", @@ -1458,10 +1440,12 @@ def completion_cost( region_name=region_name, ) - potential_model_names: Final = sorted( - (selected_model, _get_response_model(completion_response), *((model,) if model is not None else ())), - key=lambda candidate: _prices_only_via_capability_rule(candidate, cast(str | None, custom_llm_provider)), - ) + potential_model_names: Final = [ + selected_model, + _get_response_model(completion_response), + ] + if model is not None: + potential_model_names.append(model) for idx, model in enumerate(potential_model_names): try: @@ -1476,7 +1460,7 @@ def completion_cost( else: usage_obj = getattr(completion_response, "usage", {}) if isinstance(usage_obj, BaseModel) and not _is_known_usage_objects(usage_obj=usage_obj): - _usage_for_dump = usage_obj + _usage_for_dump = cast(BaseModel, usage_obj) setattr( completion_response, "usage", @@ -1485,7 +1469,7 @@ def completion_cost( if usage_obj is None: _usage = {} elif isinstance(usage_obj, BaseModel): - _usage = usage_obj.model_dump() + _usage = cast(BaseModel, usage_obj).model_dump() else: _usage = usage_obj @@ -2140,10 +2124,7 @@ def pricing_entry_for_cost_calc( router_model_id=router_model_id, region_name=region_name, ) - candidates: Final = sorted( - (selected_model, _get_response_model(completion_response), model), - key=lambda candidate: _prices_only_via_capability_rule(candidate, custom_llm_provider), - ) + candidates: Final = (selected_model, _get_response_model(completion_response), model) resolved: Final = next( (info for info in (_cost_map_model_info(name, custom_llm_provider) for name in candidates if name) if info), None, diff --git a/tests/integration/spend/test_stream_alias_billing.py b/tests/integration/spend/test_stream_alias_billing.py deleted file mode 100644 index 78f8db8b64b..00000000000 --- a/tests/integration/spend/test_stream_alias_billing.py +++ /dev/null @@ -1,464 +0,0 @@ -"""A model name that only matches a capability rule never zeroes the deployment's price (LIT-9065). - -The proxy restamps every streamed chunk with the client's alias, so end-of-stream cost calculation can see -"claude-opus-4.8-" before the deployment's model. That name is no cost-map key but matches the claude -capability generalization rules, whose model info carries no prices, so the dotted alias must bill exactly what -the plain alias "integration-" bills at the same deployment rates. The same holds for a deployment whose -model_info.base_model only matches a rule, on every endpoint and client -""" - -import asyncio -import json -import threading -from collections.abc import Callable -from concurrent.futures import ThreadPoolExecutor -from hashlib import sha256 -from typing import Final -from uuid import uuid4 - -import anthropic -import httpx -import openai -import pytest -from integration._support.client import Gateway, Scenario, eventually, object_value, string_value -from integration._support.database import read_rows -from integration._support.wire import Reply, Request, wire_server -from pydantic import JsonValue - - -def _sse_event(name: str, payload: dict[str, JsonValue]) -> bytes: - return f"event: {name}\ndata: {json.dumps(payload, separators=(',', ':'))}\n\n".encode() - - -def _anthropic_stream(request: Request) -> Reply: - assert request.target.endswith("/v1/messages"), request.target - body: Final = json.loads(request.body) - assert body["model"] == "claude-opus-4-8" and body["stream"] is True, body - return Reply( - content_type="text/event-stream", - chunks=( - _sse_event( - "message_start", - { - "type": "message_start", - "message": { - "id": f"msg_{uuid4().hex[:12]}", - "type": "message", - "role": "assistant", - "model": "claude-opus-4-8", - "content": [], - "stop_reason": None, - "stop_sequence": None, - "usage": {"input_tokens": 30, "output_tokens": 1}, - }, - }, - ), - _sse_event( - "content_block_start", - {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}, - ), - _sse_event( - "content_block_delta", - {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "hi"}}, - ), - _sse_event("content_block_stop", {"type": "content_block_stop", "index": 0}), - _sse_event( - "message_delta", - { - "type": "message_delta", - "delta": {"stop_reason": "end_turn", "stop_sequence": None}, - "usage": {"output_tokens": 40}, - }, - ), - _sse_event("message_stop", {"type": "message_stop"}), - ), - ) - - -def _deployment( - scenario: Scenario, - model_name: str, - litellm_params: dict[str, JsonValue], - model_info: dict[str, JsonValue] | None = None, -) -> str: - created: Final = scenario.gateway.post( - "/model/new", {"model_name": model_name, "litellm_params": litellm_params, "model_info": model_info or {}} - ) - identity: Final = string_value(object_value(created["model_info"])["id"]) - scenario.cleanups.callback(scenario.delete_model, identity) - return model_name - - -def _streamed_spend(gateway: Gateway, scenario: Scenario, model: str, content: str) -> dict[str, JsonValue]: - key: Final = scenario.key(models=[model]) - response: Final = gateway.request( - "POST", - "/v1/chat/completions", - { - "model": model, - "messages": [{"role": "user", "content": content}], - "stream": True, - "stream_options": {"include_usage": True}, - }, - key=key, - ) - assert response.status_code == 200, response.text - rows: Final = eventually( - lambda: read_rows( - 'SELECT spend, prompt_tokens, completion_tokens FROM "LiteLLM_SpendLogs" WHERE api_key=%s', - (sha256(key.encode()).hexdigest(),), - ), - lambda values: len(values) == 1, - seconds=70, - ) - return rows[0] - - -def _listed_deployments(gateway: Gateway, model_name: str) -> tuple[dict[str, JsonValue], ...]: - entries: Final = gateway.get("/model/info")["data"] - assert isinstance(entries, list) - return tuple(object_value(entry) for entry in entries if object_value(entry)["model_name"] == model_name) - - -def _deployment_pricing(gateway: Gateway, model_name: str) -> dict[str, JsonValue]: - listed: Final = eventually(lambda: _listed_deployments(gateway, model_name), lambda found: len(found) == 1) - return object_value(listed[0]["model_info"]) - - -@pytest.mark.parametrize( - "litellm_params", - ( - pytest.param( - lambda _: {"model": "vertex_ai/claude-opus-4-8@default", "mock_response": "hi"}, - id="vertex-mock-response", - ), - pytest.param( - lambda wire_url: { - "model": "anthropic/claude-opus-4-8", - "api_key": "integration-provider-key", - "api_base": wire_url, - }, - id="anthropic-upstream", - ), - ), -) -@pytest.mark.timeout(180) -def test_streamed_alias_matching_a_capability_rule_bills_the_deployment_price( - gateway: Gateway, litellm_params: Callable[[str], dict[str, JsonValue]] -) -> None: - with wire_server(_anthropic_stream) as wire, gateway.scenario() as scenario: - content: Final = f"alias billing {uuid4().hex}" - plain_alias: Final = f"integration-{uuid4().hex}" - rule_alias: Final = f"claude-opus-4.8-{uuid4().int % 10**8:08d}" - exact_row: Final = _streamed_spend( - gateway, scenario, _deployment(scenario, plain_alias, litellm_params(wire.url)), content - ) - alias_row: Final = _streamed_spend( - gateway, scenario, _deployment(scenario, rule_alias, litellm_params(wire.url)), content - ) - - for model_name, row in ((plain_alias, exact_row), (rule_alias, alias_row)): - pricing: Final = _deployment_pricing(gateway, model_name) - input_rate: Final = float(str(pricing["input_cost_per_token"])) - output_rate: Final = float(str(pricing["output_cost_per_token"])) - uplift: Final = float(str(pricing["regional_endpoint_uplift_multiplier"] or 1)) - assert input_rate > 0 and output_rate > 0, pricing - assert float(str(row["spend"])) == pytest.approx( - uplift - * (float(str(row["prompt_tokens"])) * input_rate + float(str(row["completion_tokens"])) * output_rate) - ), (model_name, row, pricing) - - -INPUT_TOKENS: Final = 30 -OUTPUT_TOKENS: Final = 40 -DEPLOYMENT_MODEL: Final = "anthropic/claude-opus-4-8" -ENDPOINTS: Final = ("/v1/chat/completions", "/v1/messages", "/v1/responses") - - -def _rule_only_name() -> str: - return f"claude-opus-4.8-{uuid4().int % 10**8:08d}" - - -def _anthropic_reply(request: Request) -> Reply: - body: Final = json.loads(request.body) - if body.get("stream") is True: - return _anthropic_stream(request) - assert request.target.endswith("/v1/messages") and body["model"] == "claude-opus-4-8", (request.target, body) - return Reply( - body=json.dumps( - { - "id": f"msg_{uuid4().hex[:12]}", - "type": "message", - "role": "assistant", - "model": "claude-opus-4-8", - "content": [{"type": "text", "text": "hi"}], - "stop_reason": "end_turn", - "stop_sequence": None, - "usage": {"input_tokens": INPUT_TOKENS, "output_tokens": OUTPUT_TOKENS}, - } - ).encode() - ) - - -def _anthropic_params(wire_url: str) -> dict[str, JsonValue]: - return {"model": DEPLOYMENT_MODEL, "api_key": "integration-provider-key", "api_base": wire_url} - - -def _listed_rates(scenario: Scenario, model: str, wire_url: str) -> tuple[float, float]: - model_name: Final = _deployment( - scenario, f"integration-{uuid4().hex}", {**_anthropic_params(wire_url), "model": model} - ) - pricing: Final = _deployment_pricing(scenario.gateway, model_name) - uplift: Final = float(str(pricing.get("regional_endpoint_uplift_multiplier") or 1)) - rates: Final = ( - uplift * float(str(pricing["input_cost_per_token"])), - uplift * float(str(pricing["output_cost_per_token"])), - ) - assert rates[0] > 0 and rates[1] > 0, pricing - return rates - - -def _body(path: str, model: str, content: str, stream: bool) -> dict[str, JsonValue]: - match path: - case "/v1/chat/completions": - return { - "model": model, - "messages": [{"role": "user", "content": content}], - "stream": stream, - **({"stream_options": {"include_usage": True}} if stream else {}), - } - case "/v1/messages": - return { - "model": model, - "max_tokens": 64, - "messages": [{"role": "user", "content": content}], - "stream": stream, - } - case _: - return {"model": model, "input": content, "stream": stream} - - -def _spend_rows(key: str, count: int) -> list[dict[str, JsonValue]]: - return eventually( - lambda: read_rows( - 'SELECT request_id, spend, prompt_tokens, completion_tokens, status, cache_hit FROM "LiteLLM_SpendLogs"' - ' WHERE api_key=%s ORDER BY "startTime"', - (sha256(key.encode()).hexdigest(),), - ), - lambda values: len(values) == count, - seconds=90, - ) - - -def _rule_only_base_model_deployment(scenario: Scenario, wire_url: str) -> str: - return _deployment( - scenario, f"integration-{uuid4().hex}", _anthropic_params(wire_url), {"base_model": _rule_only_name()} - ) - - -@pytest.mark.parametrize("stream", (False, True), ids=("non-streaming", "streaming")) -@pytest.mark.parametrize("path", ENDPOINTS) -@pytest.mark.timeout(180) -def test_rule_only_base_model_bills_the_deployment_price(gateway: Gateway, path: str, stream: bool) -> None: - with wire_server(_anthropic_reply) as wire, gateway.scenario() as scenario: - input_rate, output_rate = _listed_rates(scenario, DEPLOYMENT_MODEL, wire.url) - model: Final = _rule_only_base_model_deployment(scenario, wire.url) - key: Final = scenario.key(models=[model]) - - response: Final = gateway.request( - "POST", path, _body(path, model, f"base model {uuid4().hex}", stream), key=key - ) - - assert response.status_code == 200, response.text - row: Final = _spend_rows(key, 1)[0] - assert (row["prompt_tokens"], row["completion_tokens"]) == (INPUT_TOKENS, OUTPUT_TOKENS), row - assert float(str(row["spend"])) == pytest.approx(INPUT_TOKENS * input_rate + OUTPUT_TOKENS * output_rate), row - - -def _openai_sync_chat_stream(base_url: str, key: str, model: str, content: str) -> None: - with openai.OpenAI(base_url=f"{base_url}/v1", api_key=key, max_retries=0) as client: - chunks: Final = tuple( - client.chat.completions.create( - model=model, - messages=[{"role": "user", "content": content}], - stream=True, - stream_options={"include_usage": True}, - ) - ) - assert "".join(chunk.choices[0].delta.content or "" for chunk in chunks if chunk.choices) == "hi", chunks - - -def _openai_async_responses(base_url: str, key: str, model: str, content: str) -> None: - async def call() -> str: - async with openai.AsyncOpenAI(base_url=f"{base_url}/v1", api_key=key, max_retries=0) as client: - return (await client.responses.create(model=model, input=content)).output_text - - assert asyncio.run(call()) == "hi" - - -def _anthropic_async_messages_stream(base_url: str, key: str, model: str, content: str) -> None: - async def call() -> int: - async with anthropic.AsyncAnthropic(base_url=base_url, api_key=key, max_retries=0) as client: - async with client.messages.stream( - model=model, max_tokens=64, messages=[{"role": "user", "content": content}] - ) as stream: - return (await stream.get_final_message()).usage.output_tokens - - assert asyncio.run(call()) == OUTPUT_TOKENS - - -@pytest.mark.parametrize( - "client_call", - ( - pytest.param(_openai_sync_chat_stream, id="openai-sync-chat-stream"), - pytest.param(_openai_async_responses, id="openai-async-responses"), - pytest.param(_anthropic_async_messages_stream, id="anthropic-async-messages-stream"), - ), -) -@pytest.mark.timeout(180) -def test_rule_only_base_model_bills_the_deployment_price_through_the_sdks( - gateway: Gateway, client_call: Callable[[str, str, str, str], None] -) -> None: - with wire_server(_anthropic_reply) as wire, gateway.scenario() as scenario: - input_rate, output_rate = _listed_rates(scenario, DEPLOYMENT_MODEL, wire.url) - model: Final = _rule_only_base_model_deployment(scenario, wire.url) - key: Final = scenario.key(models=[model]) - - client_call(str(gateway.client.base_url).rstrip("/"), key, model, f"sdk {uuid4().hex}") - - row: Final = _spend_rows(key, 1)[0] - assert float(str(row["spend"])) == pytest.approx(INPUT_TOKENS * input_rate + OUTPUT_TOKENS * output_rate), row - - -@pytest.mark.parametrize("stream", (False, True), ids=("non-streaming", "streaming")) -@pytest.mark.timeout(180) -def test_custom_pricing_still_beats_a_rule_only_base_model(gateway: Gateway, stream: bool) -> None: - with wire_server(_anthropic_reply) as wire, gateway.scenario() as scenario: - model: Final = _deployment( - scenario, - f"integration-{uuid4().hex}", - {**_anthropic_params(wire.url), "input_cost_per_token": 0.001, "output_cost_per_token": 0.002}, - {"base_model": _rule_only_name()}, - ) - key: Final = scenario.key(models=[model]) - path: Final = "/v1/chat/completions" - - response: Final = gateway.request("POST", path, _body(path, model, f"custom {uuid4().hex}", stream), key=key) - - assert response.status_code == 200, response.text - assert float(str(_spend_rows(key, 1)[0]["spend"])) == pytest.approx(30 * 0.001 + 40 * 0.002) - - -@pytest.mark.parametrize("stream", (False, True), ids=("non-streaming", "streaming")) -@pytest.mark.timeout(180) -def test_priced_base_model_still_bills_its_own_price(gateway: Gateway, stream: bool) -> None: - with wire_server(_anthropic_reply) as wire, gateway.scenario() as scenario: - input_rate, output_rate = _listed_rates(scenario, "anthropic/claude-haiku-4-5", wire.url) - model: Final = _deployment( - scenario, f"integration-{uuid4().hex}", _anthropic_params(wire.url), {"base_model": "claude-haiku-4-5"} - ) - key: Final = scenario.key(models=[model]) - path: Final = "/v1/chat/completions" - - response: Final = gateway.request("POST", path, _body(path, model, f"priced {uuid4().hex}", stream), key=key) - - assert response.status_code == 200, response.text - row: Final = _spend_rows(key, 1)[0] - assert float(str(row["spend"])) == pytest.approx(INPUT_TOKENS * input_rate + OUTPUT_TOKENS * output_rate), row - - -@pytest.mark.parametrize( - "base_model", - ( - pytest.param("", id="empty"), - pytest.param(f"claude-opus-4.8-{'9' * 5000}", id="5kb-rule-only"), - pytest.param(f"integration-unmapped-{uuid4().hex}", id="unmapped-no-rule"), - ), -) -@pytest.mark.timeout(180) -def test_odd_base_model_values_bill_the_deployment_price(gateway: Gateway, base_model: str) -> None: - with wire_server(_anthropic_reply) as wire, gateway.scenario() as scenario: - input_rate, output_rate = _listed_rates(scenario, DEPLOYMENT_MODEL, wire.url) - model: Final = _deployment( - scenario, f"integration-{uuid4().hex}", _anthropic_params(wire.url), {"base_model": base_model} - ) - key: Final = scenario.key(models=[model]) - path: Final = "/v1/chat/completions" - - response: Final = gateway.request("POST", path, _body(path, model, f"odd {uuid4().hex}", True), key=key) - - assert response.status_code == 200, response.text - row: Final = _spend_rows(key, 1)[0] - assert float(str(row["spend"])) == pytest.approx(INPUT_TOKENS * input_rate + OUTPUT_TOKENS * output_rate), row - - -@pytest.mark.parametrize("path", ENDPOINTS) -@pytest.mark.timeout(180) -def test_upstream_failure_on_a_rule_only_base_model_logs_a_zero_spend_failure(gateway: Gateway, path: str) -> None: - failure: Final = Reply( - status=500, body=b'{"type":"error","error":{"type":"api_error","message":"integration upstream down"}}' - ) - with wire_server(lambda _: failure) as wire, gateway.scenario() as scenario: - model: Final = _rule_only_base_model_deployment(scenario, wire.url) - key: Final = scenario.key(models=[model]) - - response: Final = gateway.request("POST", path, _body(path, model, f"down {uuid4().hex}", False), key=key) - - assert response.status_code == 500, response.text - row: Final = _spend_rows(key, 1)[0] - assert (row["status"], float(str(row["spend"]))) == ("failure", 0.0), row - - -@pytest.mark.timeout(180) -def test_cache_hit_on_a_rule_only_base_model_bills_only_the_first_call(gateway: Gateway) -> None: - with wire_server(_anthropic_reply) as wire, gateway.scenario() as scenario: - input_rate, output_rate = _listed_rates(scenario, DEPLOYMENT_MODEL, wire.url) - model: Final = _rule_only_base_model_deployment(scenario, wire.url) - key: Final = scenario.key(models=[model]) - path: Final = "/v1/chat/completions" - body: Final = _body(path, model, f"cached {uuid4().hex}", False) - - responses: Final = tuple(gateway.request("POST", path, body, key=key) for _ in range(2)) - - assert [response.status_code for response in responses] == [200, 200], [r.text for r in responses] - rows: Final = _spend_rows(key, 2) - assert [(row["cache_hit"], float(str(row["spend"]))) for row in rows] == [ - ("None", pytest.approx(INPUT_TOKENS * input_rate + OUTPUT_TOKENS * output_rate)), - ("True", 0.0), - ], rows - assert len([request for request in wire.drain() if request.target.endswith("/v1/messages")]) == 1 - - -@pytest.mark.timeout(300) -def test_burst_through_an_upstream_outage_bills_every_recovered_request_once(gateway: Gateway) -> None: - outage: Final = threading.Event() - overloaded: Final = Reply(status=529, body=b'{"type":"error","error":{"type":"overloaded_error","message":"x"}}') - burst: Final = tuple((path, stream) for path in ENDPOINTS for stream in (False, True)) * 4 - with ( - wire_server(lambda request: overloaded if outage.is_set() else _anthropic_reply(request)) as wire, - gateway.scenario() as scenario, - ): - input_rate, output_rate = _listed_rates(scenario, DEPLOYMENT_MODEL, wire.url) - model: Final = _rule_only_base_model_deployment(scenario, wire.url) - key: Final = scenario.key(models=[model]) - - def send(cell: tuple[str, bool]) -> httpx.Response: - return gateway.request("POST", cell[0], _body(cell[0], model, f"burst {uuid4().hex}", cell[1]), key=key) - - outage.set() - with ThreadPoolExecutor(max_workers=len(burst)) as pool: - during: Final = tuple(pool.map(send, burst)) - outage.clear() - with ThreadPoolExecutor(max_workers=len(burst)) as pool: - after: Final = tuple(pool.map(send, burst)) - - assert all(response.status_code != 200 for response in during), [r.status_code for r in during] - assert [response.status_code for response in after] == [200] * len(burst), [r.text for r in after] - rows: Final = _spend_rows(key, 2 * len(burst)) - succeeded: Final = tuple(row for row in rows if row["status"] == "success") - assert len({row["request_id"] for row in succeeded}) == len(succeeded) == len(burst), rows - assert {float(str(row["spend"])) for row in rows if row["status"] != "success"} == {0.0}, rows - assert all( - float(str(row["spend"])) == pytest.approx(INPUT_TOKENS * input_rate + OUTPUT_TOKENS * output_rate) - for row in succeeded - ), succeeded diff --git a/tests/unit/test_cost_calculator.py b/tests/unit/test_cost_calculator.py index 509a697a511..36e188e82d6 100644 --- a/tests/unit/test_cost_calculator.py +++ b/tests/unit/test_cost_calculator.py @@ -15,7 +15,6 @@ from litellm.cost_calculator import ( completion_cost, cost_per_token, handle_realtime_stream_cost_calculation, - pricing_entry_for_cost_calc, response_cost_calculator, ) from litellm.litellm_core_utils.litellm_logging import Logging @@ -5640,56 +5639,3 @@ def test_completion_cost_bills_base_when_gemini_serves_on_demand( ) assert cost == pytest.approx(100 * 0.001 + 50 * 0.002) - - -@pytest.mark.parametrize( - "custom_llm_provider,deployment_model,cost_map_key", - [ - ("vertex_ai", "claude-opus-4-8@default", "vertex_ai/claude-opus-4-8@default"), - ("anthropic", "claude-opus-4-8", "claude-opus-4-8"), - ], -) -def test_completion_cost_prices_capability_rule_alias_from_the_deployment( - _local_model_cost_map: None, custom_llm_provider: str, deployment_model: str, cost_map_key: str -) -> None: - """Streamed proxy chunks carry the client's alias, so the first cost candidate is the - provider-prefixed alias. That name matches a claude capability generalization rule (unpriced) - and must fall through to the deployment's priced model instead of stopping at $0.""" - response: Final = ModelResponse( - id="chatcmpl_x", - choices=[{"index": 0, "message": {"role": "assistant", "content": "hi"}, "finish_reason": "stop"}], - model="claude-opus-4.8", - usage=Usage(prompt_tokens=30, completion_tokens=40, total_tokens=70), - ) - row: Final = litellm.model_cost[cost_map_key] - expected: Final = 30 * row["input_cost_per_token"] + 40 * row["output_cost_per_token"] - assert expected > 0 - - assert completion_cost( - completion_response=response, - model=deployment_model, - custom_llm_provider=custom_llm_provider, - ) == pytest.approx(expected) - - -def test_pricing_entry_for_cost_calc_skips_capability_rule_alias(_local_model_cost_map: None) -> None: - response: Final = ModelResponse( - id="chatcmpl_x", - choices=[{"index": 0, "message": {"role": "assistant", "content": "hi"}, "finish_reason": "stop"}], - model="claude-opus-4.8", - usage=Usage(prompt_tokens=30, completion_tokens=40, total_tokens=70), - ) - - resolved: Final = pricing_entry_for_cost_calc( - model="claude-opus-4-8@default", - completion_response=response, - custom_llm_provider="vertex_ai", - custom_pricing=None, - base_model=None, - router_model_id=None, - region_name=None, - litellm_logging_obj=None, - ) - - assert resolved is not None - assert resolved[0] == "vertex_ai/claude-opus-4-8@default" diff --git a/tests/unit/test_main.py b/tests/unit/test_main.py index 3866828d4ba..59c907d60c6 100644 --- a/tests/unit/test_main.py +++ b/tests/unit/test_main.py @@ -4092,7 +4092,7 @@ def test_stream_chunk_builder_skips_stamp_when_cost_is_unpriceable(): from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging logging_obj: Final = LiteLLMLogging( - model="unmapped-deployment-without-cost-map-entry", + model="us.anthropic.claude-opus-5", messages=[{"role": "user", "content": "hi"}], stream=True, call_type="completion",