scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-07 02:59:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103

test(router): move the sync tpm regression into the existing tpm routing test file

Browse source 
tests/local_testing/test_tpm_rpm_routing_v2.py is the test file that already covers
LowestTPMLoggingHandler_v2, so the regression for the sync get_available_deployments
raise lives there instead of in a new file

Co-authored-by: songkuan-zheng <252822057+songkuan-zheng@users.noreply.github.com>
Co-authored-by: songkuan-zheng <songkuan-zheng@users.noreply.github.com>
This commit is contained in:
songkuan-zheng 2026-09-11 12:06:35 +00:00
parent 455e68bf4a
commit 20e7ac6d8a
2 changed files with 23 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
tests/local_testing/test_tpm_rpm_routing_v2.py
View file

@ -770,3 +770,26 @@ async def test_tpm_rpm_routing_model_name_checks():
standard_logging_payload["hidden_params"]["litellm_model_name"]
== "azure/gpt-4.1-mini"
)
def test_every_deployment_over_its_tpm_limit_raises_a_429():
from litellm.types.router import RouterErrors, RouterNoDeploymentsAvailableError
test_cache = DualCache()
lowest_tpm_logger = LowestTPMLoggingHandler(router_cache=test_cache)
deployment = {
"model_name": "gpt-4o-mini",
"litellm_params": {"model": "openai/gpt-4o-mini", "tpm": 10},
"model_info": {"id": "d1"},
}
minute = get_utc_datetime().strftime("%H-%M")
test_cache.set_cache(key=f"d1:openai/gpt-4o-mini:tpm:{minute}", value=100)
with pytest.raises(RouterNoDeploymentsAvailableError) as raised:
lowest_tpm_logger.get_available_deployments(
model_group="gpt-4o-mini",
healthy_deployments=[deployment],
messages=[{"role": "user", "content": "hi"}],
)
assert raised.value.status_code == 429
assert RouterErrors.no_deployments_available.value in str(raised.value)

27
tests/unit/router_strategy/test_lowest_tpm_rpm_v2.py
View file

@ -1,27 +0,0 @@
import pytest
from litellm.caching.caching import DualCache
from litellm.router_strategy.lowest_tpm_rpm_v2 import LowestTPMLoggingHandler_v2
from litellm.types.router import RouterErrors, RouterNoDeploymentsAvailableError
from litellm.utils import get_utc_datetime
def test_every_deployment_over_its_tpm_limit_raises_a_429():
cache = DualCache()
handler = LowestTPMLoggingHandler_v2(router_cache=cache)
deployment = {
"model_name": "gpt-4o-mini",
"litellm_params": {"model": "openai/gpt-4o-mini", "tpm": 10},
"model_info": {"id": "d1"},
}
minute = get_utc_datetime().strftime("%H-%M")
cache.set_cache(key=f"d1:openai/gpt-4o-mini:tpm:{minute}", value=100)
with pytest.raises(RouterNoDeploymentsAvailableError) as raised:
handler.get_available_deployments(
model_group="gpt-4o-mini",
healthy_deployments=[deployment],
messages=[{"role": "user", "content": "hi"}],
)
assert raised.value.status_code == 429
assert RouterErrors.no_deployments_available.value in str(raised.value)