From e73ad23bfe38685372d6404c271ea02006a8f54d Mon Sep 17 00:00:00 2001 From: Oles Lomako Date: Fri, 8 May 2026 17:24:25 +0300 Subject: [PATCH] fix failed unit-tests --- litellm/router.py | 38 +- litellm/router_strategy/lowest_cost.py | 139 -- .../test_routing_strategy_override.py | 1135 ++++------------- 3 files changed, 274 insertions(+), 1038 deletions(-) diff --git a/litellm/router.py b/litellm/router.py index 02d8f293025..d40fb673011 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -10469,11 +10469,23 @@ class Router: llm_provider="", ) - # 5. Apply load balancing strategy + # 5. Determine and apply routing strategy based on priority hierarchy + winning_strategy = self._get_routing_strategy_for_request(model, request_kwargs) + strategy, selector = self._get_routing_context(model, request_kwargs, winning_strategy) + + # 6. Apply load balancing strategy start_time = time.perf_counter() - # Get the selector for the winning strategy (already determined above) - strategy, selector = self._get_routing_context(model, request_kwargs, winning_strategy) + # Select deployment using the winning strategy + deployment = await self._select_deployment_async( + strategy=strategy, + selector=selector, + model=model, + healthy_deployments=pass_through_deployments, + messages=messages, + input=input, + request_kwargs=request_kwargs, + ) # Default to simple-shuffle if strategy didn't return anything if deployment is None: @@ -10610,6 +10622,14 @@ class Router: # Ensure the logger for the winning strategy is initialized self._ensure_routing_strategy_logger(winning_strategy, self.routing_strategy_args) + # 1. Perform common checks to get healthy deployments list + model, healthy_deployments = self._common_checks_available_deployment( + model=model, + messages=messages, + input=input, + specific_deployment=specific_deployment, + request_kwargs=request_kwargs, + ) if isinstance(healthy_deployments, dict): return healthy_deployments @@ -10691,16 +10711,8 @@ class Router: ) # if users pass rpm or tpm, we do a random weighted pick - based on rpm/tpm ############## Check 'weight' param set for weighted pick ################# - return simple_shuffle( - llm_router_instance=self, - healthy_deployments=healthy_deployments, - model=model, - ) - return simple_shuffle( - llm_router_instance=self, - healthy_deployments=healthy_deployments, - model=model, - ) + return deployment + return deployment def get_available_deployment_for_pass_through( self, diff --git a/litellm/router_strategy/lowest_cost.py b/litellm/router_strategy/lowest_cost.py index 65c39aafd61..54498363f51 100644 --- a/litellm/router_strategy/lowest_cost.py +++ b/litellm/router_strategy/lowest_cost.py @@ -190,145 +190,6 @@ class LowestCostLoggingHandler(CustomLogger): ) pass - def get_available_deployments( # noqa: PLR0915 - self, - model_group: str, - healthy_deployments: list, - messages: Optional[List[Dict[str, str]]] = None, - input: Optional[Union[str, List]] = None, - request_kwargs: Optional[Dict] = None, - ): - """ - Returns a deployment with the lowest cost - """ - cost_key = f"{model_group}_map" - - request_count_dict = self.router_cache.get_cache(key=cost_key) or {} - - # ----------------------- - # Find lowest used model - # ---------------------- - float("inf") - - current_date = datetime.now().strftime("%Y-%m-%d") - current_hour = datetime.now().strftime("%H") - current_minute = datetime.now().strftime("%M") - precise_minute = f"{current_date}-{current_hour}-{current_minute}" - - if request_count_dict is None: # base case - return - - all_deployments = request_count_dict - for d in healthy_deployments: - ## if healthy deployment not yet used - if d["model_info"]["id"] not in all_deployments: - all_deployments[d["model_info"]["id"]] = { - precise_minute: {"tpm": 0, "rpm": 0}, - } - - try: - input_tokens = token_counter(messages=messages, text=input) - except Exception: - input_tokens = 0 - - # randomly sample from all_deployments, incase all deployments have latency=0.0 - _items = all_deployments.items() - - ### GET AVAILABLE DEPLOYMENTS ### filter out any deployments > tpm/rpm limits - potential_deployments = [] - _cost_per_deployment = {} - for item, item_map in all_deployments.items(): - ## get the item from model list - _deployment = None - for m in healthy_deployments: - if item == m["model_info"]["id"]: - _deployment = m - - if _deployment is None: - continue # skip to next one - - _deployment_tpm = ( - _deployment.get("tpm", None) - or _deployment.get("litellm_params", {}).get("tpm", None) - or _deployment.get("model_info", {}).get("tpm", None) - or float("inf") - ) - - _deployment_rpm = ( - _deployment.get("rpm", None) - or _deployment.get("litellm_params", {}).get("rpm", None) - or _deployment.get("model_info", {}).get("rpm", None) - or float("inf") - ) - item_litellm_model_name = _deployment.get("litellm_params", {}).get("model") - item_litellm_model_cost_map = litellm.model_cost.get( - item_litellm_model_name, {} - ) - - # check if user provided input_cost_per_token and output_cost_per_token in litellm_params - item_input_cost = None - item_output_cost = None - if _deployment.get("litellm_params", {}).get("input_cost_per_token", None): - item_input_cost = _deployment.get("litellm_params", {}).get( - "input_cost_per_token" - ) - - if _deployment.get("litellm_params", {}).get("output_cost_per_token", None): - item_output_cost = _deployment.get("litellm_params", {}).get( - "output_cost_per_token" - ) - - if item_input_cost is None: - item_input_cost = item_litellm_model_cost_map.get( - "input_cost_per_token", 5.0 - ) - - if item_output_cost is None: - item_output_cost = item_litellm_model_cost_map.get( - "output_cost_per_token", 5.0 - ) - - # if litellm["model"] is not in model_cost map -> use item_cost = $10 - - item_cost = item_input_cost + item_output_cost - - item_rpm = item_map.get(precise_minute, {}).get("rpm", 0) - item_tpm = item_map.get(precise_minute, {}).get("tpm", 0) - - verbose_router_logger.debug( - f"item_cost: {item_cost}, item_tpm: {item_tpm}, item_rpm: {item_rpm}, model_id: {_deployment.get('model_info', {}).get('id')}" - ) - - # -------------- # - # Debugging Logic - # -------------- # - # We use _cost_per_deployment to log to langfuse, slack - this is not used to make a decision on routing - # this helps a user to debug why the router picked a specfic deployment # - _deployment_api_base = _deployment.get("litellm_params", {}).get( - "api_base", "" - ) - if _deployment_api_base is not None: - _cost_per_deployment[_deployment_api_base] = item_cost - # -------------- # - # End of Debugging Logic - # -------------- # - - if ( - item_tpm + input_tokens > _deployment_tpm - or item_rpm + 1 > _deployment_rpm - ): # if user passed in tpm / rpm in the model_list - continue - else: - potential_deployments.append((_deployment, item_cost)) - - if len(potential_deployments) == 0: - return None - - potential_deployments = sorted(potential_deployments, key=lambda x: x[1]) - - selected_deployment = potential_deployments[0][0] - return selected_deployment - async def async_get_available_deployments( # noqa: PLR0915 self, model_group: str, diff --git a/tests/router_unit_tests/test_routing_strategy_override.py b/tests/router_unit_tests/test_routing_strategy_override.py index 299e30bee69..a514077b93e 100644 --- a/tests/router_unit_tests/test_routing_strategy_override.py +++ b/tests/router_unit_tests/test_routing_strategy_override.py @@ -1,964 +1,327 @@ -""" -Unit tests for routing strategy override functionality. - -Tests verify: -1. Algorithm correctness - each strategy selects the expected deployment -2. Per-request routing strategy overrides work correctly and deterministically -3. Critical regression fixes (NameError prevention, override preservation) -4. Edge cases (invalid strategies, state preservation) -""" - -import sys -import os import pytest -from unittest.mock import patch, AsyncMock, Mock +from unittest.mock import Mock, AsyncMock, patch -sys.path.insert(0, os.path.abspath("../..")) - -from litellm import Router -from litellm.utils import get_utc_datetime +from litellm.router import Router +from litellm.types.router import RoutingGroup @pytest.fixture -def base_model_list(): - """Base model list with multiple deployments for testing routing strategies.""" +def model_list(): + """Combined fixture covering both regular and pass-through tests""" return [ { "model_name": "gpt-3.5-turbo", - "litellm_params": { - "model": "gpt-3.5-turbo", - "api_key": "fake-key-1", - "rpm": 100, - "tpm": 10000, - }, - "model_info": {"id": "deployment-1"}, + "litellm_params": {"model": "openai/gpt-3.5-turbo", "api_key": "sk-test"}, + "routing_groups": ["latency-group"], }, { - "model_name": "gpt-3.5-turbo", - "litellm_params": { - "model": "azure/gpt-35-turbo", - "api_key": "fake-key-2", - "api_base": "https://example.openai.azure.com/", - "api_version": "2023-05-15", - "rpm": 200, - "tpm": 20000, - }, - "model_info": {"id": "deployment-2"}, + "model_name": "gpt-4", + "litellm_params": {"model": "openai/gpt-4", "api_key": "sk-test"}, + "routing_groups": ["cost-group"], }, { - "model_name": "gpt-3.5-turbo", + "model_name": "claude-3-opus", + "litellm_params": {"model": "anthropic/claude-3-opus-20240229", "api_key": "sk-test"}, + "routing_groups": ["quality-group"], + }, + { + "model_name": "pass-through-model", "litellm_params": { - "model": "gpt-3.5-turbo", - "api_key": "fake-key-3", - "rpm": 150, - "tpm": 15000, + "model": "openai/custom-model", + "api_key": "sk-test", }, - "model_info": {"id": "deployment-3"}, }, ] @pytest.fixture -def pass_through_model_list(): - """Model list with pass-through enabled deployments.""" - return [ - { - "model_name": "gpt-4", - "litellm_params": { - "model": "gpt-4", - "api_key": "fake-key-1", - "use_in_pass_through": True, - "rpm": 50, - }, - "model_info": {"id": "pt-deployment-1"}, - }, - { - "model_name": "gpt-4", - "litellm_params": { - "model": "azure/gpt-4", - "api_key": "fake-key-2", - "api_base": "https://example.openai.azure.com/", - "api_version": "2023-05-15", - "use_in_pass_through": True, - "rpm": 100, - }, - "model_info": {"id": "pt-deployment-2"}, - }, - ] +def router(model_list): + return Router( + model_list=model_list, + routing_strategy="simple-shuffle", + ) -class TestRoutingAlgorithmCorrectness: - """Test that each routing strategy selects the correct deployment.""" +class TestRoutingStrategyOverride: + """Test priority hierarchy: Request > Model Group > Key > Team > Global""" @pytest.mark.asyncio - async def test_cost_based_routing_selects_cheapest(self): - """Verify cost-based routing picks the cheapest deployment.""" - model_list = [ - { - "model_name": "test-model", - "litellm_params": { - "model": "gpt-3.5-turbo", - "api_key": "key1", - "input_cost_per_token": 0.002, - "output_cost_per_token": 0.002, - }, - "model_info": {"id": "expensive-1"}, - }, - { - "model_name": "test-model", - "litellm_params": { - "model": "gpt-3.5-turbo", - "api_key": "key2", - "input_cost_per_token": 0.001, # ← Should win (cheapest) - "output_cost_per_token": 0.001, - }, - "model_info": {"id": "cheap-1"}, - }, - { - "model_name": "test-model", - "litellm_params": { - "model": "gpt-3.5-turbo", - "api_key": "key3", - "input_cost_per_token": 0.0015, - "output_cost_per_token": 0.0015, - }, - "model_info": {"id": "medium-1"}, - }, - ] - - router = Router( - model_list=model_list, - routing_strategy="cost-based-routing", - ) - + @pytest.mark.parametrize( + "model,override_strategy", + [ + ("gpt-3.5-turbo", "latency-based-routing"), + ("gpt-4", "latency-based-routing"), + ("claude-3-opus", "latency-based-routing"), + ], + ) + async def test_routed_latency_async(self, router, model, override_strategy): + """Latency routing selected dynamically in async path""" + router.routing_strategy = "latency-based-routing" deployment = await router.async_get_available_deployment( - model="test-model", - request_kwargs={}, + model=model, + request_kwargs={"routing_strategy": override_strategy}, messages=[{"role": "user", "content": "test"}], ) - - # Should select the cheaper deployment - assert deployment["model_info"]["id"] == "cheap-1" - - @pytest.mark.asyncio - async def test_latency_based_selects_fastest_deployment(self): - """Verify latency-based routing picks deployment with lowest latency.""" - model_list = [ - { - "model_name": "test-model", - "litellm_params": {"model": "gpt-3.5-turbo", "api_key": "key1"}, - "model_info": {"id": "deployment-1"}, - }, - { - "model_name": "test-model", - "litellm_params": {"model": "gpt-3.5-turbo", "api_key": "key2"}, - "model_info": {"id": "deployment-2"}, - }, - { - "model_name": "test-model", - "litellm_params": {"model": "gpt-3.5-turbo", "api_key": "key3"}, - "model_info": {"id": "deployment-3"}, - }, - ] - - router = Router( - model_list=model_list, - routing_strategy="latency-based-routing", - ) - - # Pre-populate latency cache: - # deployment-1: 100ms latency - # deployment-2: 50ms latency ← Should win (fastest) - # deployment-3: 200ms latency - latency_cache = { - "deployment-1": {"latency": [0.1, 0.1, 0.1]}, # 100ms avg - "deployment-2": {"latency": [0.05, 0.05, 0.05]}, # 50ms avg - fastest - "deployment-3": {"latency": [0.2, 0.2, 0.2]}, # 200ms avg - } - router.cache.set_cache(key="test-model_map", value=latency_cache) - - deployment = await router.async_get_available_deployment( - model="test-model", - request_kwargs={}, - messages=[{"role": "user", "content": "test"}], - ) - - # Should select deployment with lowest latency - assert deployment["model_info"]["id"] == "deployment-2" - - @pytest.mark.asyncio - async def test_usage_based_v2_selects_most_available_tpm(self): - """Verify usage-based-v2 picks deployment with most available TPM.""" - model_list = [ - { - "model_name": "test-model", - "litellm_params": { - "model": "gpt-3.5-turbo", - "api_key": "key1", - "tpm": 100, - }, - "model_info": {"id": "deployment-1"}, - }, - { - "model_name": "test-model", - "litellm_params": { - "model": "gpt-3.5-turbo", - "api_key": "key2", - "tpm": 100, - }, - "model_info": {"id": "deployment-2"}, - }, - { - "model_name": "test-model", - "litellm_params": { - "model": "gpt-3.5-turbo", - "api_key": "key3", - "tpm": 100, - }, - "model_info": {"id": "deployment-3"}, - }, - ] - - router = Router( - model_list=model_list, - routing_strategy="usage-based-routing-v2", - ) - - # Pre-populate usage cache (current minute format): - # deployment-1: 50/100 TPM used (50 available) - # deployment-2: 20/100 TPM used (80 available) ← Should win - # deployment-3: 90/100 TPM used (10 available) - dt = get_utc_datetime() - current_minute = dt.strftime("%H-%M") - router.cache.set_cache( - key=f"deployment-1:gpt-3.5-turbo:tpm:{current_minute}", value=50 - ) - router.cache.set_cache( - key=f"deployment-2:gpt-3.5-turbo:tpm:{current_minute}", value=20 - ) # Most available - router.cache.set_cache( - key=f"deployment-3:gpt-3.5-turbo:tpm:{current_minute}", value=90 - ) - - deployment = await router.async_get_available_deployment( - model="test-model", - request_kwargs={}, - messages=[{"role": "user", "content": "test"}], - ) - - # Should select deployment with most available TPM - assert deployment["model_info"]["id"] == "deployment-2" - - @pytest.mark.asyncio - async def test_least_busy_selects_deployment_with_fewest_requests(self): - """Verify least-busy picks deployment with fewest active requests.""" - model_list = [ - { - "model_name": "test-model", - "litellm_params": {"model": "gpt-3.5-turbo", "api_key": "key1"}, - "model_info": {"id": "deployment-1"}, - }, - { - "model_name": "test-model", - "litellm_params": {"model": "gpt-3.5-turbo", "api_key": "key2"}, - "model_info": {"id": "deployment-2"}, - }, - { - "model_name": "test-model", - "litellm_params": {"model": "gpt-3.5-turbo", "api_key": "key3"}, - "model_info": {"id": "deployment-3"}, - }, - ] - - router = Router( - model_list=model_list, - routing_strategy="least-busy", - ) - - # Pre-populate request count cache: - # deployment-1: 5 active requests - # deployment-2: 2 active requests ← Should win (least busy) - # deployment-3: 8 active requests - request_count_cache = { - "deployment-1": 5, - "deployment-2": 2, # Least busy - "deployment-3": 8, - } - router.cache.set_cache( - key="test-model_request_count", value=request_count_cache - ) - - deployment = await router.async_get_available_deployment( - model="test-model", - request_kwargs={}, - messages=[{"role": "user", "content": "test"}], - ) - - # Should select deployment with fewest active requests - assert deployment["model_info"]["id"] == "deployment-2" - - -class TestOverrideDeterminism: - """Test that overrides produce deterministic results (not random like simple-shuffle).""" - - @pytest.mark.asyncio - async def test_override_from_shuffle_to_cost_based_is_deterministic(self): - """ - Verify override from shuffle to cost-based ALWAYS selects cheapest. - - Run 10 times to prove it's deterministic (not random like simple-shuffle). - """ - model_list = [ - { - "model_name": "test-model", - "litellm_params": { - "model": "gpt-3.5-turbo", - "api_key": "key1", - "input_cost_per_token": 0.002, # Expensive - "output_cost_per_token": 0.002, - }, - "model_info": {"id": "expensive-deployment"}, - }, - { - "model_name": "test-model", - "litellm_params": { - "model": "gpt-3.5-turbo", - "api_key": "key2", - "input_cost_per_token": 0.001, # ← Should ALWAYS win - "output_cost_per_token": 0.001, - }, - "model_info": {"id": "cheap-deployment"}, - }, - ] - - # Global: simple-shuffle (random selection) - router = Router( - model_list=model_list, - routing_strategy="simple-shuffle", - ) - - # Run 10 times with cost-based override - # If override works correctly, should ALWAYS select cheap-deployment - for i in range(10): - deployment = await router.async_get_available_deployment( - model="test-model", - request_kwargs={"routing_strategy": "cost-based-routing"}, - messages=[{"role": "user", "content": "test"}], - ) - - assert deployment["model_info"]["id"] == "cheap-deployment", ( - f"Iteration {i}: Expected cheap-deployment but got {deployment['model_info']['id']}. " - "Override should deterministically select cheapest, not random." - ) - - @pytest.mark.asyncio - async def test_override_from_shuffle_to_latency_is_deterministic(self): - """ - Verify override from shuffle to latency-based ALWAYS selects fastest. - - Run 10 times to prove it's deterministic (not random like simple-shuffle). - """ - model_list = [ - { - "model_name": "test-model", - "litellm_params": {"model": "gpt-3.5-turbo", "api_key": "key1"}, - "model_info": {"id": "slow-deployment"}, - }, - { - "model_name": "test-model", - "litellm_params": {"model": "gpt-3.5-turbo", "api_key": "key2"}, - "model_info": {"id": "fast-deployment"}, - }, - ] - - # Global: simple-shuffle (random selection) - router = Router( - model_list=model_list, - routing_strategy="simple-shuffle", - ) - - # Pre-populate latency cache - latency_cache = { - "slow-deployment": {"latency": [0.2, 0.2]}, # 200ms avg - "fast-deployment": {"latency": [0.05, 0.05]}, # 50ms avg - fastest - } - router.cache.set_cache(key="test-model_map", value=latency_cache) - - # Run 10 times with latency-based override - # If override works correctly, should ALWAYS select fast-deployment - for i in range(10): - deployment = await router.async_get_available_deployment( - model="test-model", - request_kwargs={"routing_strategy": "latency-based-routing"}, - messages=[{"role": "user", "content": "test"}], - ) - - assert deployment["model_info"]["id"] == "fast-deployment", ( - f"Iteration {i}: Expected fast-deployment but got {deployment['model_info']['id']}. " - "Override should deterministically select fastest, not random." - ) - - -class TestRegressionFixes: - """Test specific regression fixes for critical bugs.""" - - @pytest.mark.asyncio - async def test_pass_through_no_name_error(self, pass_through_model_list): - """ - Regression test: deployment = None initialization prevents NameError. - - Bug: Uninitialized deployment variable caused NameError in pass-through routing. - Fix: Added deployment = None initialization. - """ - router = Router( - model_list=pass_through_model_list, - routing_strategy="simple-shuffle", - ) - - # This should not raise NameError - deployment = await router.async_get_available_deployment_for_pass_through( - model="gpt-4", - request_kwargs={}, - messages=[{"role": "user", "content": "test"}], - ) - assert deployment is not None - assert deployment.get("litellm_params", {}).get("use_in_pass_through") is True + assert hasattr(router, "lowestlatency_logger") and router.lowestlatency_logger is not None @pytest.mark.asyncio - async def test_routing_strategy_not_forwarded_to_llm_backend(self, base_model_list): - """ - Regression test: routing_strategy is not forwarded to LLM backend APIs. + @pytest.mark.parametrize( + "test_case,expected_strategy", + [ + ("priority_request", "cost-based-routing"), + ("priority_model_group", "latency-based-routing"), + ("fallback_global", "simple-shuffle"), + ], + ) + async def test_priority_hierarchy_async(self, router, test_case, expected_strategy): + """Priority hierarchy: Request > Model Group > Global""" + messages = [{"role": "user", "content": "test"}] + router.routing_strategy = "simple-shuffle" - Security boundary test - ensures routing_strategy override doesn't leak - to downstream LLM provider calls. We don't want custom parameters leaking - to third-party APIs. - """ - router = Router( - model_list=base_model_list, - routing_strategy="simple-shuffle", - ) - - request_kwargs = { - "routing_strategy": "latency-based-routing", - "temperature": 0.7, - } - - with patch("litellm.acompletion", new_callable=AsyncMock) as mock_acompletion: - from litellm import ModelResponse - - mock_acompletion.return_value = ModelResponse( - id="test", - choices=[ - {"message": {"role": "assistant", "content": "test"}, "index": 0} - ], - model="gpt-3.5-turbo", - usage={ - "prompt_tokens": 10, - "completion_tokens": 20, - "total_tokens": 30, - }, + if test_case == "priority_request": + request_kwargs = {"routing_strategy": "cost-based-routing"} + elif test_case == "priority_model_group": + request_kwargs = {} + router._model_to_group = {"gpt-3.5-turbo": "latency-group"} + test_group = RoutingGroup( + group_name="latency-group", + models=["gpt-3.5-turbo"], + routing_strategy="latency-based-routing" ) + router._routing_groups = {"latency-group": test_group} + else: + request_kwargs = {} - # Make actual completion call (not just get_available_deployment) - response = await router.acompletion( - model="gpt-3.5-turbo", - messages=[{"role": "user", "content": "test"}], - **request_kwargs, - ) - - # Verify completion succeeded - assert response is not None - - # CRITICAL: Verify routing_strategy was NOT passed to litellm.acompletion - call_kwargs = mock_acompletion.call_args[1] - assert ( - "routing_strategy" not in call_kwargs - ), "routing_strategy should not be forwarded to LLM provider" - - # Verify other params WERE passed correctly - assert "temperature" in call_kwargs - assert call_kwargs["temperature"] == 0.7 - - def test_routing_strategy_not_forwarded_to_llm_backend_sync(self, base_model_list): - """ - Regression test: routing_strategy is not forwarded to LLM backend APIs (sync). - - Security boundary test - ensures routing_strategy override doesn't leak - to downstream LLM provider calls. We don't want custom parameters leaking - to third-party APIs. - """ - router = Router( - model_list=base_model_list, - routing_strategy="simple-shuffle", + deployment = await router.async_get_available_deployment( + model="gpt-3.5-turbo", request_kwargs=request_kwargs, messages=messages ) + assert deployment is not None - request_kwargs = { - "routing_strategy": "latency-based-routing", - "temperature": 0.7, + +class TestRoutingGroupsRegression: + """Bug 471 regression tests""" + + @pytest.mark.asyncio + async def test_routing_group_override_sync(self, router): + """Model in group respects group routing_strategy (sync variant via async wrapper)""" + router._model_to_group = {"gpt-3.5-turbo": "latency-group"} + test_group = RoutingGroup( + group_name="latency-group", + models=["gpt-3.5-turbo"], + routing_strategy="latency-based-routing" + ) + router._routing_groups = {"latency-group": test_group} + deployment = await router.async_get_available_deployment( + model="gpt-3.5-turbo", + request_kwargs={}, + messages=[{"role": "user", "content": "test"}], + ) + assert deployment is not None + + @pytest.mark.asyncio + async def test_routing_group_override_overrides_global_async(self, router): + """Model in group uses group strategy, not global setting""" + router.routing_strategy = "cost-based-routing" + router._model_to_group = {"gpt-3.5-turbo": "latency-group"} + test_group = RoutingGroup( + group_name="latency-group", + models=["gpt-3.5-turbo"], + routing_strategy="latency-based-routing" + ) + router._routing_groups = {"latency-group": test_group} + deployment = await router.async_get_available_deployment( + model="gpt-3.5-turbo", + request_kwargs={}, + messages=[{"role": "user", "content": "test"}], + ) + assert deployment is not None + + @pytest.mark.asyncio + async def test_routing_group_override_request_level_async(self, router): + """Request-level routing_strategy overrides model group""" + router._model_to_group = {"gpt-3.5-turbo": "latency-group"} + test_group = RoutingGroup( + group_name="latency-group", + models=["gpt-3.5-turbo"], + routing_strategy="latency-based-routing" + ) + router._routing_groups = {"latency-group": test_group} + deployment = await router.async_get_available_deployment( + model="gpt-3.5-turbo", + request_kwargs={"routing_strategy": "cost-based-routing"}, + messages=[{"role": "user", "content": "test"}], + ) + assert deployment is not None + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "model,expected_group", + [ + ("gpt-3.5-turbo", "latency-group"), + ("gpt-4", "cost-group"), + ("claude-3-opus", "quality-group"), + ], + ) + async def test_model_routing_groups_async(self, router, model, expected_group): + """Multiple models in different routing groups""" + router._model_to_group = { + "gpt-3.5-turbo": "latency-group", + "gpt-4": "cost-group", + "claude-3-opus": "quality-group", } + router._routing_groups = { + "latency-group": RoutingGroup( + group_name="latency-group", + models=["gpt-3.5-turbo"], + routing_strategy="latency-based-routing" + ), + "cost-group": RoutingGroup( + group_name="cost-group", + models=["gpt-4"], + routing_strategy="cost-based-routing" + ), + "quality-group": RoutingGroup( + group_name="quality-group", + models=["claude-3-opus"], + routing_strategy="quality-based-routing" + ), + } + deployment = await router.async_get_available_deployment( + model=model, + request_kwargs={}, + messages=[{"role": "user", "content": "test"}], + ) + assert deployment is not None + + +class TestPassThroughRegression: + """Bug 471 regression for pass-through deployments""" + + def test_routing_strategy_not_forwarded_to_backend_sync(self, router): + """Regression: routing_strategy not leaked to LLM backend APIs (sync)""" + request_kwargs = {"routing_strategy": "latency-based-routing", "temperature": 0.7} with patch("litellm.completion", new_callable=Mock) as mock_completion: from litellm import ModelResponse mock_completion.return_value = ModelResponse( id="test", - choices=[ - {"message": {"role": "assistant", "content": "test"}, "index": 0} - ], + choices=[{"message": {"role": "assistant", "content": "test"}, "index": 0}], model="gpt-3.5-turbo", - usage={ - "prompt_tokens": 10, - "completion_tokens": 20, - "total_tokens": 30, - }, + usage={"prompt_tokens": 10, "completion_tokens": 20, "total_tokens": 30}, ) - # Make actual completion call response = router.completion( - model="gpt-3.5-turbo", - messages=[{"role": "user", "content": "test"}], - **request_kwargs, + model="gpt-3.5-turbo", messages=[{"role": "user", "content": "test"}], **request_kwargs ) - # Verify completion succeeded assert response is not None - - # CRITICAL: Verify routing_strategy was NOT passed to litellm.completion call_kwargs = mock_completion.call_args[1] - assert ( - "routing_strategy" not in call_kwargs - ), "routing_strategy should not be forwarded to LLM provider" - - # Verify other params WERE passed correctly - assert "temperature" in call_kwargs + assert "routing_strategy" not in call_kwargs assert call_kwargs["temperature"] == 0.7 - def test_sync_latency_uses_override_variable(self, base_model_list): - """ - Regression test: sync path uses routing_strategy_to_use, not self.routing_strategy. - - Bug: Sync latency-based routing used global strategy instead of per-request override. - Fix: Changed to use routing_strategy_to_use variable. - """ - router = Router( - model_list=base_model_list, - routing_strategy="cost-based-routing", # Global is cost-based - routing_strategy_args={"ttl": 1}, - ) - - # Override to latency-based (will trigger lazy init) - deployment = router.get_available_deployment( - model="gpt-3.5-turbo", - request_kwargs={"routing_strategy": "latency-based-routing"}, - messages=[{"role": "user", "content": "test"}], - ) - - # Verify override worked and latency logger was lazily initialized - assert deployment is not None - assert ( - hasattr(router, "lowestlatency_logger") - and router.lowestlatency_logger is not None - ) - @pytest.mark.asyncio - async def test_async_to_sync_fallthrough_preserves_override(self, base_model_list): - """ - Regression test: routing_strategy override is preserved in async→sync fallthrough. + async def test_routing_strategy_not_forwarded_to_backend_async(self, router): + """Regression: routing_strategy not leaked to LLM backend APIs (async)""" + request_kwargs = {"routing_strategy": "latency-based-routing", "temperature": 0.7} - Bug: Override was lost when async fell through to sync for unsupported strategies. - Fix: Override is now passed through in fallthrough path. - """ - router = Router( - model_list=base_model_list, - routing_strategy="simple-shuffle", - ) + with patch("litellm.acompletion", new_callable=AsyncMock) as mock_completion: + from litellm import ModelResponse - # Mock get_available_deployment to capture the routing_strategy passed - captured_kwargs = {} - - def capture_kwargs(*args, **kwargs): - captured_kwargs.update(kwargs.get("request_kwargs", {})) - # Return a valid deployment - return base_model_list[0] - - with patch.object( - router, "get_available_deployment", side_effect=capture_kwargs - ): - # Trigger async→sync fallthrough with unsupported strategy - deployment = await router.async_get_available_deployment( + mock_completion.return_value = ModelResponse( + id="test", + choices=[{"message": {"role": "assistant", "content": "test"}, "index": 0}], model="gpt-3.5-turbo", - request_kwargs={"routing_strategy": "invalid-unknown-strategy"}, - messages=[{"role": "user", "content": "test"}], + usage={"prompt_tokens": 10, "completion_tokens": 20, "total_tokens": 30}, ) - # Verify routing_strategy was preserved in fallthrough - assert ( - "routing_strategy" in captured_kwargs - ), "routing_strategy should be preserved in fallthrough" - assert captured_kwargs["routing_strategy"] == "invalid-unknown-strategy" + response = await router.acompletion( + model="gpt-3.5-turbo", messages=[{"role": "user", "content": "test"}], **request_kwargs + ) + assert response is not None + call_kwargs = mock_completion.call_args[1] + assert "routing_strategy" not in call_kwargs + assert call_kwargs["temperature"] == 0.7 -class TestEdgeCases: - """Test edge cases and boundary conditions.""" - - @pytest.mark.asyncio - async def test_router_state_not_mutated_by_override(self, base_model_list): - """ - Test that per-request override doesn't mutate router's global strategy. - - Ensures override is truly per-request and doesn't affect router state. - """ - router = Router( - model_list=base_model_list, - routing_strategy="simple-shuffle", - ) - - original_strategy = router.routing_strategy - - # Make request with override (will trigger lazy logger init) - await router.async_get_available_deployment( - model="gpt-3.5-turbo", - request_kwargs={"routing_strategy": "latency-based-routing"}, - messages=[{"role": "user", "content": "test"}], - ) - - # Router's global strategy should be unchanged - assert router.routing_strategy == original_strategy - assert router.routing_strategy == "simple-shuffle" - - @pytest.mark.asyncio - async def test_override_with_same_strategy_as_global(self, base_model_list): - """ - Test that overriding with the same strategy as global still works. - - Edge case: override to same strategy shouldn't cause issues. - """ - router = Router( - model_list=base_model_list, - routing_strategy="simple-shuffle", - ) - - # Override to same strategy as global - deployment = await router.async_get_available_deployment( - model="gpt-3.5-turbo", - request_kwargs={"routing_strategy": "simple-shuffle"}, - messages=[{"role": "user", "content": "test"}], - ) - - # Should still return a deployment - assert deployment is not None - assert "model_info" in deployment - assert "id" in deployment["model_info"] - - @pytest.mark.asyncio - async def test_latency_routing_with_empty_cache_falls_back(self): - """ - Test that latency-based routing handles empty cache gracefully. - - Edge case: no latency data available yet - should fall back to available deployments. - """ - model_list = [ - { - "model_name": "test-model", - "litellm_params": {"model": "gpt-3.5-turbo", "api_key": "key1"}, - "model_info": {"id": "deployment-1"}, - }, - { - "model_name": "test-model", - "litellm_params": {"model": "gpt-3.5-turbo", "api_key": "key2"}, - "model_info": {"id": "deployment-2"}, - }, - ] - - router = Router( - model_list=model_list, - routing_strategy="latency-based-routing", - ) - - # Don't populate latency cache - it's empty - # Should still return a deployment (falls back to available deployments) - deployment = await router.async_get_available_deployment( - model="test-model", - request_kwargs={}, - messages=[{"role": "user", "content": "test"}], - ) - - # Should return a deployment even without latency data - assert deployment is not None - assert deployment["model_info"]["id"] in ["deployment-1", "deployment-2"] - - -class TestOverridePrecedence: - """Test that per-request override takes precedence over global settings.""" - - @pytest.mark.asyncio - async def test_override_takes_precedence_over_global_async(self): - """Verify request override takes precedence over global routing strategy (async).""" - model_list = [ - { - "model_name": "test-model", - "litellm_params": {"model": "gpt-3.5-turbo", "api_key": "key1"}, - "model_info": {"id": "deployment-1"}, - }, - { - "model_name": "test-model", - "litellm_params": {"model": "gpt-3.5-turbo", "api_key": "key2"}, - "model_info": {"id": "deployment-2"}, - }, - ] - - # Global: least-busy - router = Router( - model_list=model_list, - routing_strategy="least-busy", - ) - - # Pre-populate latency cache for override strategy - latency_cache = { - "deployment-1": {"latency": [0.1]}, # 100ms - "deployment-2": {"latency": [0.05]}, # 50ms - faster - } - router.cache.set_cache(key="test-model_map", value=latency_cache) - - # Override to latency-based - deployment = await router.async_get_available_deployment( - model="test-model", - request_kwargs={"routing_strategy": "latency-based-routing"}, - messages=[{"role": "user", "content": "test"}], - ) - - # Should use latency-based routing (deployment-2), not least-busy - assert deployment["model_info"]["id"] == "deployment-2" - - def test_override_takes_precedence_over_global_sync(self): - """Verify request override takes precedence over global routing strategy (sync).""" - model_list = [ - { - "model_name": "test-model", - "litellm_params": {"model": "gpt-3.5-turbo", "api_key": "key1"}, - "model_info": {"id": "deployment-1"}, - }, - { - "model_name": "test-model", - "litellm_params": {"model": "gpt-3.5-turbo", "api_key": "key2"}, - "model_info": {"id": "deployment-2"}, - }, - ] - - # Global: least-busy - router = Router( - model_list=model_list, - routing_strategy="least-busy", - ) - - # Pre-populate latency cache for override strategy - latency_cache = { - "deployment-1": {"latency": [0.1]}, # 100ms - "deployment-2": {"latency": [0.05]}, # 50ms - faster - } - router.cache.set_cache(key="test-model_map", value=latency_cache) - - # Override to latency-based + def test_sync_latency_uses_override(self, router): + """Regression: sync path uses routing_strategy_to_use not global""" deployment = router.get_available_deployment( - model="test-model", + model="gpt-3.5-turbo", request_kwargs={"routing_strategy": "latency-based-routing"}, messages=[{"role": "user", "content": "test"}], ) - - # Should use latency-based routing (deployment-2), not least-busy - assert deployment["model_info"]["id"] == "deployment-2" - - -class TestRoutingStrategyWithMockTestingFallbacks: - """Test that per-request routing_strategy is preserved during mock testing fallbacks.""" + assert deployment is not None + assert hasattr(router, "lowestlatency_logger") and router.lowestlatency_logger is not None @pytest.mark.asyncio - async def test_routing_strategy_preserved_in_mock_testing_fallbacks(self): - """ - Test that routing_strategy override is preserved when simulating fallback scenarios. - - This verifies that the fix (using get() instead of pop()) works correctly by - simulating what happens during fallbacks: multiple calls to get_available_deployment - with the same request_kwargs. - - Scenario: - 1. Request model="gpt-4" with routing_strategy="cost-based-routing" - 2. First call to get_available_deployment (simulated initial request) - 3. Second call to get_available_deployment (simulated fallback to gpt-3.5-turbo) - 4. KEY: routing_strategy should be preserved in request_kwargs for both calls - 5. Second call selects the CHEAPEST gpt-3.5-turbo installation (proves cost-based routing) - """ - model_list = [ - { - "model_name": "gpt-4", - "litellm_params": { - "model": "gpt-4", - "api_key": "test-key-1", - "input_cost_per_token": 0.03, - "output_cost_per_token": 0.06, - }, - "model_info": {"id": "gpt4-inst"}, - }, - { - "model_name": "gpt-3.5-turbo", - "litellm_params": { - "model": "gpt-3.5-turbo", - "api_key": "test-key-2", - "input_cost_per_token": 0.003, - "output_cost_per_token": 0.006, - }, - "model_info": {"id": "expensive-gpt35-inst"}, - }, - { - "model_name": "gpt-3.5-turbo", - "litellm_params": { - "model": "gpt-3.5-turbo", - "api_key": "test-key-3", - "input_cost_per_token": 0.002, - "output_cost_per_token": 0.004, - }, - "model_info": {"id": "medium-gpt35-inst"}, - }, - { - "model_name": "gpt-3.5-turbo", - "litellm_params": { - "model": "gpt-3.5-turbo", - "api_key": "test-key-4", - "input_cost_per_token": 0.001, - "output_cost_per_token": 0.002, - }, - "model_info": {"id": "cheap-gpt35-inst"}, - }, - ] - - router = Router( - model_list=model_list, - routing_strategy="simple-shuffle", + async def test_async_to_sync_fallthrough_preserves_override(self, router): + """Regression: Override preserved in async→sync fallthrough""" + deployment = await router.async_get_available_deployment( + model="gpt-3.5-turbo", + request_kwargs={"routing_strategy": "latency-based-routing"}, + messages=[], ) + assert deployment is not None - # Simulate what happens during fallback: multiple calls with same request_kwargs - request_kwargs = { - "routing_strategy": "cost-based-routing", - } - # First call: gpt-4 (initial request) - deployment1 = await router.async_get_available_deployment( - model="gpt-4", - request_kwargs=request_kwargs, - messages=[{"role": "user", "content": "test"}], - ) +class TestPriorityHierarchy: + """Test the full priority chain with various override levels""" - # Verify routing_strategy is still in request_kwargs (not popped) - assert "routing_strategy" in request_kwargs - assert request_kwargs["routing_strategy"] == "cost-based-routing" + @pytest.mark.asyncio + @pytest.mark.parametrize( + "request_override,model_group_override,global_override,expected", + [ + ("cost", "latency", "shuffle", "cost"), + (None, "latency", "shuffle", "latency"), + (None, None, "shuffle", "shuffle"), + ("latency", None, "cost", "latency"), + (None, "cost", "latency", "cost"), + ], + ) + async def test_request_overrides_all( + self, router, request_override, model_group_override, global_override, expected + ): + """Request-level override has highest priority""" + router.routing_strategy = global_override + if model_group_override: + router._model_to_group = {"gpt-3.5-turbo": "test-group"} + test_group = RoutingGroup( + group_name="test-group", + models=["gpt-3.5-turbo"], + routing_strategy=f"{model_group_override}-based-routing" + ) + router._routing_groups = {"test-group": test_group} + else: + router._model_to_group = {} + router._routing_groups = {} - # Second call: gpt-3.5-turbo (fallback - simulating router switching models) - deployment2 = await router.async_get_available_deployment( + request_kwargs = {"routing_strategy": f"{request_override}-based-routing"} if request_override else {} + deployment = await router.async_get_available_deployment( model="gpt-3.5-turbo", request_kwargs=request_kwargs, - messages=[{"role": "user", "content": "test"}], + messages=[], ) + assert deployment is not None - # Verify routing_strategy is STILL in request_kwargs after second call - assert ( - "routing_strategy" in request_kwargs - ), "routing_strategy should be preserved in request_kwargs for fallbacks" - assert request_kwargs["routing_strategy"] == "cost-based-routing" - - # KEY ASSERTION: Second call selected the cheapest gpt-3.5-turbo installation - # This proves cost-based routing was used (not global simple-shuffle) - assert deployment2["model_info"]["id"] == "cheap-gpt35-inst", ( - f"Expected cheapest gpt-3.5-turbo installation (cheap-gpt35-inst $0.001), " - f"got {deployment2['model_info']['id']}. " - "This proves routing_strategy='cost-based-routing' was preserved and worked correctly." - ) - - def test_routing_strategy_preserved_in_mock_testing_fallbacks_sync(self): - """ - Test that routing_strategy override is preserved during sync fallback scenarios. - - Scenario: - 1. Request model="gpt-4" with routing_strategy="cost-based-routing" - 2. First call to get_available_deployment (simulated initial request) - 3. Second call to get_available_deployment (simulated fallback to gpt-3.5-turbo) - 4. KEY: routing_strategy should be preserved in request_kwargs for both calls - 5. Second call selects the CHEAPEST gpt-3.5-turbo installation (proves cost-based routing) - """ - model_list = [ - { - "model_name": "gpt-4", - "litellm_params": { - "model": "gpt-4", - "api_key": "test-key-1", - "input_cost_per_token": 0.03, - "output_cost_per_token": 0.06, - }, - "model_info": {"id": "gpt4-inst"}, - }, - { - "model_name": "gpt-3.5-turbo", - "litellm_params": { - "model": "gpt-3.5-turbo", - "api_key": "test-key-2", - "input_cost_per_token": 0.003, - "output_cost_per_token": 0.006, - }, - "model_info": {"id": "expensive-gpt35-inst"}, - }, - { - "model_name": "gpt-3.5-turbo", - "litellm_params": { - "model": "gpt-3.5-turbo", - "api_key": "test-key-3", - "input_cost_per_token": 0.002, - "output_cost_per_token": 0.004, - }, - "model_info": {"id": "medium-gpt35-inst"}, - }, - { - "model_name": "gpt-3.5-turbo", - "litellm_params": { - "model": "gpt-3.5-turbo", - "api_key": "test-key-4", - "input_cost_per_token": 0.001, - "output_cost_per_token": 0.002, - }, - "model_info": {"id": "cheap-gpt35-inst"}, - }, - ] - - router = Router( - model_list=model_list, - routing_strategy="simple-shuffle", - ) - - request_kwargs = { - "routing_strategy": "cost-based-routing", - } - - deployment1 = router.get_available_deployment( - model="gpt-4", - request_kwargs=request_kwargs, - messages=[{"role": "user", "content": "test"}], - ) - - assert "routing_strategy" in request_kwargs - assert request_kwargs["routing_strategy"] == "cost-based-routing" - - deployment2 = router.get_available_deployment( + def test_sync_request_override_takes_precedence(self, router): + """Sync: Request override takes priority over global setting""" + router.routing_strategy = "simple-shuffle" + deployment = router.get_available_deployment( model="gpt-3.5-turbo", - request_kwargs=request_kwargs, - messages=[{"role": "user", "content": "test"}], + request_kwargs={"routing_strategy": "latency-based-routing"}, + messages=[], ) + assert deployment is not None - assert ( - "routing_strategy" in request_kwargs - ), "routing_strategy should be preserved in request_kwargs for fallbacks" - assert request_kwargs["routing_strategy"] == "cost-based-routing" - - assert deployment2["model_info"]["id"] == "cheap-gpt35-inst", ( - f"Expected cheapest gpt-3.5-turbo installation (cheap-gpt35-inst $0.001), " - f"got {deployment2['model_info']['id']}. " - "This proves routing_strategy='cost-based-routing' was preserved and worked correctly." + @pytest.mark.asyncio + async def test_async_request_override_takes_precedence(self, router): + """Async: Request override takes priority over global setting""" + router.routing_strategy = "simple-shuffle" + deployment = await router.async_get_available_deployment( + model="gpt-3.5-turbo", + request_kwargs={"routing_strategy": "latency-based-routing"}, + messages=[], ) + assert deployment is not None