diff --git a/tests/integration/cost_calculation/cases/batch_realtime.py b/tests/integration/cost_calculation/cases/batch_realtime.py index 0284036f31f..52e2021dc7d 100644 --- a/tests/integration/cost_calculation/cases/batch_realtime.py +++ b/tests/integration/cost_calculation/cases/batch_realtime.py @@ -46,8 +46,8 @@ GPT_5_6_BATCH_CACHED_INPUT_HALVED: Final = BatchCostCase( ), ) -GPT_5_4_BATCH_EXPLICIT_BATCH_RATES_BILL_CACHED_AT_BATCH_INPUT_RATE: Final = BatchCostCase( - name="gpt-5.4-batch-explicit_batch_rates_bill_cached_at_batch_input_rate", +GPT_5_4_BATCH_NO_BATCH_CACHE_RATE_BILLS_CACHED_AT_STANDARD_CACHE_RATE: Final = BatchCostCase( + name="gpt-5.4-batch-no_batch_cache_rate_bills_cached_at_standard_cache_rate", covers="quota_management.spend_tracking.batch_costs.explicit_rates", model="gpt-5.4", litellm_model="openai/gpt-5.4", @@ -56,8 +56,8 @@ GPT_5_4_BATCH_EXPLICIT_BATCH_RATES_BILL_CACHED_AT_BATCH_INPUT_RATE: Final = Batc BatchOutputLine(status_code=200, prompt_tokens=120, completion_tokens=30), ), expected=ExactExpected( - spend=0.000875, - input_cost=0.000275, + spend=0.000835, + input_cost=0.000235, output_cost=0.0006, prompt_tokens=220, completion_tokens=80, @@ -188,7 +188,7 @@ GPT_REALTIME_MINI_2025_12_15_REALTIME_SESSION_WITHOUT_TURNS_ZERO_SPEND: Final = BATCH_CASES: Final[tuple[BatchCostCase, ...]] = ( GPT_5_6_BATCH_HALVED_RATES_WHEN_MAP_HAS_NO_BATCH_KEYS, GPT_5_6_BATCH_CACHED_INPUT_HALVED, - GPT_5_4_BATCH_EXPLICIT_BATCH_RATES_BILL_CACHED_AT_BATCH_INPUT_RATE, + GPT_5_4_BATCH_NO_BATCH_CACHE_RATE_BILLS_CACHED_AT_STANDARD_CACHE_RATE, GPT_5_6_BATCH_ALL_REQUESTS_FAILED_ZERO_SPEND, ) diff --git a/tests/integration/observability/test_arize_otel_v2_openinference_family.py b/tests/integration/observability/test_arize_otel_v2_openinference_family.py index f638a7360be..881a65c687e 100644 --- a/tests/integration/observability/test_arize_otel_v2_openinference_family.py +++ b/tests/integration/observability/test_arize_otel_v2_openinference_family.py @@ -23,7 +23,7 @@ from integration._support.client import Gateway from integration._support.wire import Reply, Request from pydantic import JsonValue -_GENAI_B3_KEYS_WITHOUT_BAGGAGE: Final = frozenset( +_GENAI_B3_KEYS: Final = frozenset( { "gen_ai.input.messages", "gen_ai.operation.name", @@ -56,6 +56,7 @@ _GENAI_B3_KEYS_WITHOUT_BAGGAGE: Final = frozenset( "litellm.cost.output", "litellm.cost.tool_usage", "litellm.cost.total", + "litellm.metadata.trace_marker", "litellm.provider.model", "litellm.request.route", "litellm.request.tools.declared", @@ -66,7 +67,6 @@ _GENAI_B3_KEYS_WITHOUT_BAGGAGE: Final = frozenset( "server.port", } ) -_GENAI_B3_KEYS_WITH_BAGGAGE: Final = _GENAI_B3_KEYS_WITHOUT_BAGGAGE | frozenset({"litellm.metadata.trace_marker"}) _LANGFUSE_B3_KEYS: Final = frozenset( { "gen_ai.input.messages", @@ -109,6 +109,7 @@ _LANGFUSE_B3_KEYS: Final = frozenset( "litellm.cost.output", "litellm.cost.tool_usage", "litellm.cost.total", + "litellm.metadata.trace_marker", "litellm.provider.model", "litellm.request.route", "litellm.request.tools.declared", @@ -122,15 +123,14 @@ _LANGFUSE_B3_KEYS: Final = frozenset( _B3_ATTRIBUTE_KEYS: Final[Mapping[str, frozenset[str]]] = MappingProxyType( { "langfuse_otel": _LANGFUSE_B3_KEYS, - "langtrace": _GENAI_B3_KEYS_WITHOUT_BAGGAGE, - "signoz": _GENAI_B3_KEYS_WITH_BAGGAGE, - "newrelic": _GENAI_B3_KEYS_WITH_BAGGAGE, - "levo": _GENAI_B3_KEYS_WITH_BAGGAGE, - "agentops": _GENAI_B3_KEYS_WITH_BAGGAGE, - "otel": _GENAI_B3_KEYS_WITH_BAGGAGE, + "langtrace": _GENAI_B3_KEYS, + "signoz": _GENAI_B3_KEYS, + "newrelic": _GENAI_B3_KEYS, + "levo": _GENAI_B3_KEYS, + "agentops": _GENAI_B3_KEYS, + "otel": _GENAI_B3_KEYS, } ) -_B3_BAGGAGE_CALLBACKS: Final = frozenset({"signoz", "newrelic", "levo", "agentops", "otel"}) _B4_ATTRIBUTE_KEYS: Final = frozenset( { "input.value", @@ -305,14 +305,10 @@ def test_arize_otel_v2_b3_non_openinference_callback_family(callback: str, gatew baggage: Final = tuple( sorted((key, value) for key, value in attributes.items() if key.startswith("litellm.metadata.")) ) - expected_baggage: Final = ( - (("litellm.metadata.trace_marker", marker),) if callback in _B3_BAGGAGE_CALLBACKS else () - ) - assert baggage == expected_baggage, attributes + assert baggage == (("litellm.metadata.trace_marker", marker),), attributes def test_arize_otel_v2_b3_langfuse_carries_metadata_baggage(gateway: Gateway, tmp_path: Path) -> None: - pytest.skip("BUG: LIT-9128 Langfuse OTel v2 preset omits request-metadata baggage") marker: Final = "b3-langfuse-" + uuid.uuid4().hex def upstream(request: Request) -> Reply: diff --git a/tests/integration/routing/test_usage_based_routing_redis_reads.py b/tests/integration/routing/test_usage_based_routing_redis_reads.py index 8a31eac7f91..edb24cc7930 100644 --- a/tests/integration/routing/test_usage_based_routing_redis_reads.py +++ b/tests/integration/routing/test_usage_based_routing_redis_reads.py @@ -3,6 +3,7 @@ from __future__ import annotations import json import shlex import threading +import time import uuid from collections.abc import Iterator from contextlib import contextmanager @@ -19,6 +20,7 @@ from integration._support.client import Gateway, eventually, object_value, strin from integration._support.process import owned_proxy from integration._support.redis_process import owned_redis from integration._support.wire import Reply, Request, wire_server +from litellm.constants import DEFAULT_COOLDOWN_REDIS_READ_INTERVAL_SECONDS from pydantic import JsonValue, TypeAdapter from redis import Redis from redis.exceptions import TimeoutError as RedisTimeoutError @@ -217,11 +219,12 @@ def test_proxy_usage_routing_reads_cooldown_tpm_then_rpm_from_redis( lambda registered: registered == frozenset(deployment_ids), seconds=15, ) + cooldown_read_throttle_ends: Final = time.monotonic() + DEFAULT_COOLDOWN_REDIS_READ_INTERVAL_SECONDS wire.drain() eventually( - lambda: datetime.now(UTC), - lambda current: current.second < 40, - seconds=65, + lambda: (datetime.now(UTC), time.monotonic()), + lambda clock: clock[0].second < 40 and clock[1] >= cooldown_read_throttle_ends, + seconds=65 + DEFAULT_COOLDOWN_REDIS_READ_INTERVAL_SECONDS, ) minute: Final = datetime.now(UTC).strftime("%H-%M") markers: Final = tuple(f"{run_id}-{index}" for index in range(3)) @@ -256,18 +259,18 @@ def test_proxy_usage_routing_reads_cooldown_tpm_then_rpm_from_redis( observed_bodies: Final = tuple(_request_object(request.body) for request in received) expected_bodies: Final = tuple(_expected_wire_body(endpoint, marker) for marker in markers[:2]) assert observed_bodies == expected_bodies, observed_bodies - expected_mget: Final = ( + cooldown_mget: Final = ( "MGET", - f"deployment:{deployment_ids[0]}:cooldown", - f"deployment:{deployment_ids[1]}:cooldown", - f"{deployment_ids[0]}:openai/{OPENAI_MODEL}:tpm:{minute}", - f"{deployment_ids[1]}:openai/{OPENAI_MODEL}:tpm:{minute}", - *( - f"{deployment_id}:openai/{OPENAI_MODEL}:rpm:{minute}" - for deployment_id in deployment_ids - ), + *(f"deployment:{deployment_id}:cooldown" for deployment_id in deployment_ids), + ) + usage_mget: Final = ( + "MGET", + *(f"{deployment_id}:openai/{OPENAI_MODEL}:tpm:{minute}" for deployment_id in deployment_ids), + *(f"{deployment_id}:openai/{OPENAI_MODEL}:rpm:{minute}" for deployment_id in deployment_ids), ) mgets: Final = _drain_mgets(commands) - assert any(arguments == expected_mget for _, arguments in mgets), mgets + mget_arguments: Final = tuple(arguments for _, arguments in mgets) + assert cooldown_mget in mget_arguments, mgets + assert usage_mget in mget_arguments, mgets raw_mgets: Final = tuple(line for line, _ in mgets) print(f"proxy {endpoint} MGETs: {raw_mgets}") diff --git a/tests/integration/sdk/test_router_concurrent_deployment_delete_sdk.py b/tests/integration/sdk/test_router_concurrent_deployment_delete_sdk.py index 3867808aa5f..9f8aa966335 100644 --- a/tests/integration/sdk/test_router_concurrent_deployment_delete_sdk.py +++ b/tests/integration/sdk/test_router_concurrent_deployment_delete_sdk.py @@ -19,6 +19,8 @@ from pydantic import JsonValue import litellm from litellm import Router +from litellm.router_rust.surface.model_list import HostedModelList +from litellm.router_rust.surface.registry import HostRegistry _DEPLOYMENTS: Final = 4 _TARGET: Final = _DEPLOYMENTS - 1 @@ -28,12 +30,13 @@ _BURST_THREADS: Final = 8 _BURST_CALLS: Final = 24 _CHURN_UPSERTS: Final = 120 _UNREPORTED_MODEL_ID: Final = frozenset({"aresponses-stream", "aanthropic-messages"}) -_REINDEX: Final = Router._update_deployment_indices_after_removal.__code__ +_LIST_REFRESH: Final = HostedModelList.refresh.__code__ +_HAND_CHANGES: Final = HostRegistry._hand_changes.__code__ def _router_source_files() -> frozenset[str]: facade_files: Final = (sys.modules[cls.__module__].__file__ for cls in Router.__mro__ if cls is not object) - return frozenset({_REINDEX.co_filename, *(path for path in facade_files if path is not None)}) + return frozenset({_HAND_CHANGES.co_filename, *(path for path in facade_files if path is not None)}) def _lock_lines(path: str) -> frozenset[int]: @@ -88,9 +91,9 @@ def _router(wire: Wire, team: bool = False) -> Router: def _read_during_held_write(read: Callable[[], T], write: Callable[[], object]) -> T: - """Run `read` on its own thread while `write` waits where the list has shrunk but the index maps have not caught - up. The write resumes once `read` returns or reaches the router's index lock, so a read that skips the lock meets - the half-updated list and a read that takes the lock waits for the write""" + """Run `read` on its own thread while `write` waits where router-core has applied the change but the Python model + list has not been read again. The write resumes once `read` returns or reaches the router's index lock, so a read + that skips the lock meets the half-updated list and a read that takes the lock waits for the write""" held: Final = threading.Event() release: Final = threading.Event() settled: Final = threading.Event() @@ -99,7 +102,11 @@ def _read_during_held_write(read: Callable[[], T], write: Callable[[], object]) failure: Final[list[BaseException]] = [] def hold_at_reindex(frame: FrameType, event: str, _arg: object) -> None: - if event == "call" and frame.f_code is _REINDEX and not held.is_set(): + caller: Final = frame.f_back + refreshes_after_a_write: Final = ( + frame.f_code is _LIST_REFRESH and caller is not None and caller.f_code is _HAND_CHANGES + ) + if event == "call" and refreshes_after_a_write and not held.is_set(): held.set() released_in_time.append(release.wait(timeout=_SYNC_SECONDS)) diff --git a/tests/integration/sdk/test_usage_based_routing_sdk_redis_reads.py b/tests/integration/sdk/test_usage_based_routing_sdk_redis_reads.py index da9a8309c8d..ff4761b1da3 100644 --- a/tests/integration/sdk/test_usage_based_routing_sdk_redis_reads.py +++ b/tests/integration/sdk/test_usage_based_routing_sdk_redis_reads.py @@ -140,7 +140,7 @@ async def _exercise_router(router: Router, model_name: str, markers: tuple[str, return _model_id(first), _model_id(second) -def test_sdk_usage_routing_reads_tpm_then_rpm_from_redis(tmp_path: Path) -> None: +def test_sdk_usage_routing_reads_cooldown_tpm_then_rpm_from_redis(tmp_path: Path) -> None: with owned_redis(tmp_path) as cache, wire_server(_reply) as wire: run_id: Final = uuid.uuid4().hex model_name: Final = f"usage-redis-{run_id}" @@ -176,15 +176,19 @@ def test_sdk_usage_routing_reads_tpm_then_rpm_from_redis(tmp_path: Path) -> None for marker in markers[:2] ) assert observed_bodies == expected_bodies, observed_bodies - expected_mget: Final = ( + cooldown_mget: Final = ( + "MGET", + *(f"deployment:{deployment_id}:cooldown" for deployment_id in deployment_ids), + ) + usage_mget: Final = ( "MGET", - f"deployment:{deployment_ids[0]}:cooldown", - f"deployment:{deployment_ids[1]}:cooldown", *(f"{deployment_id}:openai/{OPENAI_MODEL}:tpm:{minute}" for deployment_id in deployment_ids), *(f"{deployment_id}:openai/{OPENAI_MODEL}:rpm:{minute}" for deployment_id in deployment_ids), ) mgets: Final = _drain_mgets(commands) - assert any(arguments == expected_mget for _, arguments in mgets), mgets + mget_arguments: Final = tuple(arguments for _, arguments in mgets) + assert cooldown_mget in mget_arguments, mgets + assert usage_mget in mget_arguments, mgets raw_mgets: Final = tuple(line for line, _ in mgets) print(f"sdk MGETs: {raw_mgets}") finally: diff --git a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py index f3b427d91c4..933c152eb26 100644 --- a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py +++ b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py @@ -464,7 +464,7 @@ def test_concurrent_burst_ends_each_stream_by_whether_its_provider_finished_it(g ) -_SPEND_LOG_INSERT: Final = b'INSERT INTO "public"."LiteLLM_SpendLogs"' +_SPEND_LOG_INSERT: Final = b'"LiteLLM_SpendLogs" (' def _spend_statuses(database_url: str | None) -> Counter[tuple[str, str]]: