test(router): cover failed deployment identity paths

This commit is contained in:
King Star 2026-09-09 10:00:15 +08:00
parent 25a68ffcb6
commit 57629e7383
No known key found for this signature in database
2 changed files with 66 additions and 1 deletions

View file

@ -248,6 +248,36 @@ async def test_async_failure_metrics_prefer_stamped_failed_deployment_id():
assert _model_id_values(logger.litellm_deployment_failure_responses) == {"deployment-a"}
def test_deployment_failure_model_id_falls_back_to_nested_metadata():
logger = PrometheusLogger()
model_id = logger._get_deployment_failure_model_id(
request_kwargs={
"litellm_params": {"litellm_metadata": {"model_info": {"id": "deployment-a"}}},
},
standard_logging_payload={},
)
assert model_id == "deployment-a"
@pytest.mark.parametrize(
"request_kwargs",
[
{},
{"litellm_params": {"metadata": {"model_info": {}}}},
],
)
def test_deployment_failure_model_id_returns_none_without_a_model_id(request_kwargs):
assert (
PrometheusLogger._get_deployment_failure_model_id(
request_kwargs=request_kwargs,
standard_logging_payload={},
)
is None
)
@pytest.mark.asyncio
async def test_sdk_fallback_labels_survive_non_import_errors_from_proxy_module(monkeypatch):
logger = PrometheusLogger()

View file

@ -6,13 +6,14 @@ regardless of the routing strategy being used.
"""
import asyncio
from unittest.mock import AsyncMock, MagicMock
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
import litellm
from litellm import Router
from litellm.caching.dual_cache import DualCache
from litellm.integrations.custom_logger import CustomLogger
from litellm.router_utils.pre_call_checks.model_rate_limit_check import (
ModelRateLimitingCheck,
)
@ -353,3 +354,37 @@ class TestModelRateLimitConcurrency:
assert len(successes) == 2, f"Expected 2 successes, got {len(successes)}"
assert len(failures) == 2, f"Expected 2 rate limit errors, got {len(failures)}"
@pytest.mark.asyncio
@pytest.mark.parametrize(
"hook_error",
[
litellm.RateLimitError(message="rpm exceeded", llm_provider="openai", model="gpt-5.6"),
RuntimeError("pre call check blew up"),
],
)
async def test_router_async_pre_call_checks_stamp_the_refusing_deployment(hook_error):
class _RaisingPreCallCheck(CustomLogger):
async def async_pre_call_check(self, deployment, parent_otel_span):
raise hook_error
router = Router(
model_list=[
{
"model_name": "gpt-5.6",
"litellm_params": {"model": "openai/gpt-5.6", "api_key": "sk-fake"},
"model_info": {"id": "deployment-a"},
}
]
)
deployment = router.model_list[0]
with patch.object(litellm, "callbacks", [_RaisingPreCallCheck()]): # test-quality-ok: router reads this global
with pytest.raises(type(hook_error)):
await router.async_routing_strategy_pre_call_checks(
deployment=deployment,
parent_otel_span=None,
)
assert hook_error.failed_deployment_id == "deployment-a"