scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-10 03:28:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 85

test(routing): wait on deployment registration instead of boot model-info traffic (#45267)

Browse source
This commit is contained in:
yuneng-jiang 2026-10-08 21:31:53 -07:00 • committed by GitHub
parent 4b975e6f49
commit 72ab736863
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 10 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
tests/integration/routing/test_usage_based_routing_redis_reads.py
View file

@ -15,7 +15,7 @@ from typing import Final
import httpx
import pytest
import yaml
from integration._support.client import Gateway, eventually
from integration._support.client import Gateway, eventually, object_value, string_value
from integration._support.process import owned_proxy
from integration._support.redis_process import owned_redis
from integration._support.wire import Reply, Request, wire_server
@ -24,6 +24,7 @@ from redis import Redis
from redis.exceptions import TimeoutError as RedisTimeoutError
JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
MODEL_INFO_ENTRIES: Final = TypeAdapter(tuple[dict[str, JsonValue], ...])
MONITOR_COMMAND: Final = TypeAdapter(dict[str, JsonValue])
OPENAI_MODEL: Final = "gpt-4o-mini"
MASTER_KEY: Final = "sk-integration-usage-routing-redis-reads"
@ -161,6 +162,12 @@ def _capture_redis_commands(host: str, port: int) -> Iterator[SimpleQueue[str]]:
assert not thread.is_alive(), "Redis MONITOR thread survived cleanup"
def _registered_deployment_ids(candidate: Gateway, model_name: str) -> frozenset[str]:
entries: Final = MODEL_INFO_ENTRIES.validate_python(candidate.get("/model/info")["data"])
group: Final = tuple(entry for entry in entries if entry["model_name"] == model_name)
return frozenset(string_value(object_value(entry["model_info"])["id"]) for entry in group)
def _drain_mgets(commands: SimpleQueue[str]) -> tuple[tuple[str, tuple[str, ...]], ...]:
captured: Final = tuple(commands.get_nowait() for _ in range(commands.qsize()))
parsed: Final = tuple((line, tuple(shlex.split(line))) for line in captured)
@ -202,8 +209,8 @@ def test_proxy_usage_routing_reads_cooldown_tpm_then_rpm_from_redis(
config=config_path,
) as candidate:
eventually(
lambda: wire.received.qsize(),
lambda received: received >= len(deployment_ids),
lambda: _registered_deployment_ids(candidate, model_name),
lambda registered: registered == frozenset(deployment_ids),
seconds=15,
)
wire.drain()