test(integration): repair tests that merged changes left stale (#50782)

Several integration tests on main still assert behavior that later changes replaced on purpose, so they fail on every scheduled run, and one of them stops the whole sdk group at import.

The batch cost case expected cached tokens to bill at the Batch input price when a price row has no Batch cache price. Since #50062 they bill at the Standard cache price, so the case now expects 0.000235 input cost and is renamed to say so.

The OTel v2 attribute test pinned langfuse_otel and langtrace as carrying no request metadata baggage. Since #50373 every OTel v2 destination carries it, so all seven callbacks share one expectation and the Langfuse baggage test that was deferred under LIT-9128 runs again.

The clean-EOF outage test trips its database relay on the text of the spend log insert. The ledger writes that insert without the "public" schema prefix, so the relay never tripped and the test timed out before it checked anything. The trigger now matches the insert on both write paths.

The usage routing tests expected cooldown, tpm and rpm keys in one MGET. Since #50170 the cooldown keys are read in their own MGET, at most once per DEFAULT_COOLDOWN_REDIS_READ_INTERVAL_SECONDS. Both the proxy and the SDK test now expect the two MGETs, and the proxy test waits out that interval after its /model/info poll, which reads the same keys.

The concurrent deployment delete test held its writer inside Router._update_deployment_indices_after_removal, which went away with the Python router core in #50178, so the file failed at import and stopped the whole sdk group. It now holds the writer where the registry reads the model list again after router-core applied the change.
This commit is contained in:
yuneng-jiang 2026-11-09 17:08:35 -08:00 • committed by GitHub
parent 13df479397
commit 71c1a2aff6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 54 additions and 44 deletions

View file

@ -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,
)

View file

@ -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:

View file

@ -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}")

View file

@ -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))

View file

@ -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:

View file

@ -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]]: