From 72ab7368639cffb20dc70b124609d232be40901d Mon Sep 17 00:00:00 2001 From: yuneng-jiang Date: Thu, 8 Oct 2026 21:31:53 -0700 Subject: [PATCH] test(routing): wait on deployment registration instead of boot model-info traffic (#45267) --- .../routing/test_usage_based_routing_redis_reads.py | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) 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 f4801eb3318..f3ee7855395 100644 --- a/tests/integration/routing/test_usage_based_routing_redis_reads.py +++ b/tests/integration/routing/test_usage_based_routing_redis_reads.py @@ -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()