litellm/tests/unit/router_strategy/test_router_routing_groups.py

"""
Tests for `routing_groups` — assigns named subsets of `model_name`s to a
per-group routing strategy. Models not claimed by an explicit group fall into
the implicit `"default"` group driven by the router's top-level
`routing_strategy` / `routing_strategy_args`.
"""

import asyncio
import datetime
import time
import uuid
from collections.abc import Callable, Mapping
from typing import Final, Literal
from unittest.mock import patch

import pytest
from pydantic import ValidationError

import litellm
from litellm import Router
from litellm.caching.redis_cache import RedisPipelineIncrementOperation
from litellm.integrations.custom_logger import CustomLogger
from litellm.types.router import DeploymentTypedDict, FallbackAccessCheck, RoutingGroup, RoutingStrategy
from litellm.utils import Rules, function_setup


def _model_list():
    return [
        {
            "model_name": "filtered-model",
            "litellm_params": {
                "model": "openai/gpt-4o",
                "api_key": "sk-test-1",
                "api_base": "https://example.invalid",
            },
            "model_info": {"id": "deploy-1"},
        },
        {
            "model_name": "filtered-model",
            "litellm_params": {
                "model": "openai/gpt-4o-mini",
                "api_key": "sk-test-2",
                "api_base": "https://example.invalid",
            },
            "model_info": {"id": "deploy-2"},
        },
        {
            "model_name": "other-model",
            "litellm_params": {
                "model": "openai/gpt-4o",
                "api_key": "sk-test-3",
                "api_base": "https://example.invalid",
            },
            "model_info": {"id": "deploy-3"},
        },
    ]


def _build_router(routing_strategy="simple-shuffle", routing_groups=None):
    return Router(
        model_list=_model_list(),
        routing_strategy=routing_strategy,
        routing_groups=routing_groups,
    )


def test_no_groups_uses_top_level_strategy_for_all_models():
    router = _build_router(routing_strategy="latency-based-routing")
    assert router._get_routing_context("filtered-model")[0] == "latency-based-routing"
    assert router._get_routing_context("other-model")[0] == "latency-based-routing"


def test_explicit_group_overrides_top_level():
    router = _build_router(
        routing_strategy="simple-shuffle",
        routing_groups=[
            {
                "group_name": "fast",
                "models": ["filtered-model"],
                "routing_strategy": "latency-based-routing",
            }
        ],
    )
    strategy, selector = router._get_routing_context("filtered-model")
    assert strategy == "latency-based-routing"
    assert selector is not None
    # other-model isn't in any explicit group, so it lands in the default fallback
    strategy_other, _ = router._get_routing_context("other-model")
    assert strategy_other == "simple-shuffle"


def test_default_group_is_simple_shuffle_when_top_level_strategy_unset():
    router = _build_router()
    strategy, selector = router._get_routing_context("other-model")
    assert strategy == "simple-shuffle"
    # simple-shuffle does not use a selector
    assert selector is None


def test_two_groups_same_strategy_have_independent_selectors():
    router = _build_router(
        routing_strategy="simple-shuffle",
        routing_groups=[
            {
                "group_name": "group_a",
                "models": ["filtered-model"],
                "routing_strategy": "latency-based-routing",
                "routing_strategy_args": {"ttl": 60},
            },
            {
                "group_name": "group_b",
                "models": ["other-model"],
                "routing_strategy": "latency-based-routing",
                "routing_strategy_args": {"ttl": 600},
            },
        ],
    )
    a_selector = router._group_selectors["group_a"]["latency-based-routing"]
    b_selector = router._group_selectors["group_b"]["latency-based-routing"]
    assert a_selector is not None and b_selector is not None
    assert id(a_selector) != id(b_selector)


def test_overlapping_models_across_groups_raises():
    with pytest.raises(ValueError, match="appears in"):
        _build_router(
            routing_groups=[
                {
                    "group_name": "g1",
                    "models": ["filtered-model"],
                    "routing_strategy": "latency-based-routing",
                },
                {
                    "group_name": "g2",
                    "models": ["filtered-model"],
                    "routing_strategy": "least-busy",
                },
            ],
        )


def test_reserved_default_group_name_raises():
    with pytest.raises(ValueError, match="reserved"):
        _build_router(
            routing_groups=[
                {
                    "group_name": "default",
                    "models": ["filtered-model"],
                    "routing_strategy": "latency-based-routing",
                }
            ],
        )


def test_invalid_strategy_in_group_raises():
    with pytest.raises(ValueError, match="Invalid routing_strategy"):
        _build_router(
            routing_groups=[
                {
                    "group_name": "g1",
                    "models": ["filtered-model"],
                    "routing_strategy": "not-a-real-strategy",
                }
            ],
        )


def test_unknown_model_in_group_warns_but_does_not_raise(caplog):
    import logging

    with caplog.at_level(logging.WARNING, logger="LiteLLM Router"):
        router = _build_router(
            routing_groups=[
                {
                    "group_name": "g1",
                    "models": ["model-not-in-list"],
                    "routing_strategy": "latency-based-routing",
                }
            ],
        )
    assert "model-not-in-list" in caplog.text
    assert router._model_to_group.get("model-not-in-list") == "g1"


def test_duplicate_group_name_raises():
    with pytest.raises(ValueError, match="duplicate"):
        _build_router(
            routing_groups=[
                {
                    "group_name": "g1",
                    "models": ["filtered-model"],
                    "routing_strategy": "latency-based-routing",
                },
                {
                    "group_name": "g1",
                    "models": ["other-model"],
                    "routing_strategy": "least-busy",
                },
            ],
        )


def test_routing_group_object_input_accepted():
    router = _build_router(
        routing_groups=[
            RoutingGroup(
                group_name="g1",
                models=["filtered-model"],
                routing_strategy="latency-based-routing",
            )
        ],
    )
    assert router._model_to_group["filtered-model"] == "g1"


@pytest.mark.asyncio
async def test_async_dispatch_uses_group_strategy_for_grouped_model():
    router = _build_router(
        routing_strategy="simple-shuffle",
        routing_groups=[
            {
                "group_name": "fast",
                "models": ["filtered-model"],
                "routing_strategy": "latency-based-routing",
            }
        ],
    )

    group_selector = router._group_selectors["fast"]["latency-based-routing"]

    with (
        patch.object(
            group_selector,
            "async_get_available_deployments",
            wraps=group_selector.async_get_available_deployments,
        ) as latency_spy,
        patch(
            "litellm.router.simple_shuffle", wraps=litellm.router.simple_shuffle
        ) as shuffle_spy,
    ):
        await router.async_get_available_deployment(
            model="filtered-model", request_kwargs={}
        )

        assert (
            latency_spy.called
        ), "group's latency selector should run for grouped model"
        assert not shuffle_spy.called


@pytest.mark.asyncio
async def test_async_dispatch_falls_back_to_default_for_ungrouped_models():
    router = _build_router(
        routing_strategy="simple-shuffle",
        routing_groups=[
            {
                "group_name": "fast",
                "models": ["filtered-model"],
                "routing_strategy": "latency-based-routing",
            }
        ],
    )

    group_selector = router._group_selectors["fast"]["latency-based-routing"]

    with (
        patch.object(
            group_selector,
            "async_get_available_deployments",
            wraps=group_selector.async_get_available_deployments,
        ) as latency_spy,
        patch(
            "litellm.router.simple_shuffle", wraps=litellm.router.simple_shuffle
        ) as shuffle_spy,
    ):
        await router.async_get_available_deployment(
            model="other-model", request_kwargs={}
        )

        assert shuffle_spy.called, "default group's simple-shuffle should run"
        assert not latency_spy.called


def test_update_settings_round_trip_routing_groups():
    router = _build_router()
    assert router._model_to_group == {}

    router.update_settings(
        routing_groups=[
            {
                "group_name": "fast",
                "models": ["filtered-model"],
                "routing_strategy": "latency-based-routing",
            }
        ]
    )
    assert router._model_to_group == {"filtered-model": "fast"}
    assert router._get_routing_context("filtered-model")[0] == "latency-based-routing"

    settings = router.get_settings()
    assert any(g["group_name"] == "fast" for g in settings["routing_groups"])


def test_default_group_accepts_routing_strategy_enum_for_top_level_strategy():
    """
    The Router constructor accepts both string and RoutingStrategy enum forms.
    The default group must resolve a real selector for either input shape, or
    every ungrouped model raises NoDeploymentAvailable.
    """
    router = _build_router(routing_strategy=RoutingStrategy.LATENCY_BASED)
    strategy, selector = router._get_routing_context("other-model")
    assert strategy == "latency-based-routing"
    assert selector is not None
    assert router.lowestlatency_logger is selector


@pytest.mark.asyncio
async def test_async_dispatch_uses_default_selector_when_constructed_with_enum():
    router = _build_router(routing_strategy=RoutingStrategy.LATENCY_BASED)
    default_selector = router.lowestlatency_logger
    assert default_selector is not None

    # filtered-model has 2 deployments, so the single-deployment short-circuit
    # in async_get_available_deployment doesn't kick in and we actually hit
    # the latency selector.
    with (
        patch.object(
            default_selector,
            "async_get_available_deployments",
            wraps=default_selector.async_get_available_deployments,
        ) as latency_spy,
        patch(
            "litellm.router.simple_shuffle", wraps=litellm.router.simple_shuffle
        ) as shuffle_spy,
    ):
        await router.async_get_available_deployment(
            model="filtered-model", request_kwargs={}
        )

        assert (
            latency_spy.called
        ), "default group's latency selector must run when constructed with the enum"
        assert not shuffle_spy.called


def test_update_settings_does_not_leak_strategy_callbacks(monkeypatch):
    """
    Repeated `update_settings` calls must not accumulate stale selectors in
    `litellm.callbacks` / `litellm.input_callback`. Each rebuild owns the
    previous generation and is responsible for unregistering it by identity
    so the global lists don't grow without bound.
    """
    monkeypatch.setattr(litellm, "callbacks", [])
    monkeypatch.setattr(litellm, "input_callback", [])

    router = _build_router(routing_strategy="latency-based-routing")
    initial_default_selector = router.lowestlatency_logger
    assert initial_default_selector is not None
    assert any(
        c is initial_default_selector for c in litellm.callbacks
    ), "fresh router should register its default selector"

    # Flip strategies back and forth — each transition triggers
    # routing_strategy_init and must replace, not accumulate.
    for _ in range(3):
        router.update_settings(routing_strategy="usage-based-routing")
        router.update_settings(routing_strategy="latency-based-routing")

    assert all(
        c is not initial_default_selector for c in litellm.callbacks
    ), "old default selector instance leaked into litellm.callbacks"

    # The set of router-owned selectors should be bounded — at most one per
    # strategy class. Without the cleanup fix this grows by one per toggle.
    router_owned_classes = (
        "LeastBusyLoggingHandler",
        "LowestTPMLoggingHandler",
        "LowestTPMLoggingHandler_v2",
        "LowestLatencyLoggingHandler",
        "LowestCostLoggingHandler",
    )
    router_owned = [
        c for c in litellm.callbacks if type(c).__name__ in router_owned_classes
    ]
    assert len(router_owned) <= len(
        router_owned_classes
    ), f"router selectors leaked: {[type(c).__name__ for c in router_owned]}"

    # Group selectors: capture the v1 instance, swap, confirm v1 is gone by id.
    router.update_settings(
        routing_groups=[
            {
                "group_name": "fast",
                "models": ["filtered-model"],
                "routing_strategy": "latency-based-routing",
            }
        ]
    )
    group_selector_v1 = router._group_selectors["fast"]["latency-based-routing"]
    assert group_selector_v1 is not None

    router.update_settings(
        routing_groups=[
            {
                "group_name": "fast",
                "models": ["filtered-model"],
                "routing_strategy": "latency-based-routing",
                "routing_strategy_args": {"ttl": 600},
            }
        ]
    )
    assert all(
        c is not group_selector_v1 for c in litellm.callbacks
    ), "old group selector instance leaked into litellm.callbacks"
    group_selector_v2 = router._group_selectors["fast"]["latency-based-routing"]
    assert group_selector_v2 is not group_selector_v1


def test_update_settings_unregisters_group_selectors_when_groups_removed(monkeypatch):
    """
    Setting routing_groups to an empty list (or omitting all groups) must
    unregister every previously-owned group selector.
    """
    monkeypatch.setattr(litellm, "callbacks", [])
    monkeypatch.setattr(litellm, "input_callback", [])

    router = _build_router(
        routing_groups=[
            {
                "group_name": "fast",
                "models": ["filtered-model"],
                "routing_strategy": "least-busy",
            }
        ]
    )
    group_selector = router._group_selectors["fast"]["least-busy"]
    assert group_selector in litellm.callbacks
    assert group_selector in litellm.input_callback

    router.update_settings(routing_groups=[])

    assert all(c is not group_selector for c in litellm.callbacks)
    assert all(c is not group_selector for c in litellm.input_callback)
    assert router._group_selectors == {}


def test_two_least_busy_groups_count_a_request_once(monkeypatch):
    """
    Least-busy counts a request up from the pre-call hooks on `litellm.input_callback` and
    back down from the success hooks on `litellm.callbacks`. The success list drops a second
    selector of the same class, so a pre-call list that kept both counted every request twice
    and released it once, and the deployment's in-flight count climbed until it looked pinned.
    """
    monkeypatch.setattr(litellm, "callbacks", [])
    monkeypatch.setattr(litellm, "input_callback", [])

    router = _build_router(
        routing_strategy="least-busy",
        routing_groups=[
            {
                "group_name": "fast",
                "models": ["filtered-model"],
                "routing_strategy": "least-busy",
            }
        ],
    )
    kwargs = {
        "litellm_params": {
            "metadata": {"model_group": "filtered-model"},
            "model_info": {"id": "deploy-1"},
        }
    }

    for callback in litellm.input_callback:
        if isinstance(callback, CustomLogger):
            callback.log_pre_api_call(model="filtered-model", messages=[], kwargs=kwargs)
    for callback in litellm.callbacks:
        if isinstance(callback, CustomLogger):
            callback.log_success_event(kwargs, None, None, None)

    assert router.cache.get_cache("filtered-model_request_count:deploy-1") == 0


def test_two_routers_in_one_process_each_count_their_own_requests(monkeypatch):
    """
    Least-busy hangs its counting off litellm's global callback lists, and those lists keep one
    logger per class unless the instances differ in a plain attribute. Two routers in one process
    (a second Router, or a per-request `user_config` one) therefore have to register separately:
    a second router whose selector is dropped counts nothing, reads zero for every deployment,
    and sends every request to whichever one is listed first.
    """
    monkeypatch.setattr(litellm, "callbacks", [])
    monkeypatch.setattr(litellm, "input_callback", [])

    first = _build_router(routing_strategy="least-busy")
    second = _build_router(routing_strategy="least-busy")
    kwargs = {
        "litellm_params": {
            "metadata": {"model_group": "filtered-model"},
            "model_info": {"id": "deploy-1"},
        }
    }

    for callback in litellm.input_callback:
        if isinstance(callback, CustomLogger):
            callback.log_pre_api_call(model="filtered-model", messages=[], kwargs=kwargs)

    assert second.cache.get_cache("filtered-model_request_count:deploy-1") == 1
    assert (
        second.get_available_deployment(model="filtered-model", messages=[])["model_info"]["id"]
        == "deploy-2"
    )

    for callback in litellm.callbacks:
        if isinstance(callback, CustomLogger):
            callback.log_success_event(kwargs, None, None, None)

    assert first.cache.get_cache("filtered-model_request_count:deploy-1") == 0
    assert second.cache.get_cache("filtered-model_request_count:deploy-1") == 0


# ---------------------------------------------------------------------------
# Direct helper coverage
# ---------------------------------------------------------------------------


def test_normalize_strategy_handles_string_enum_and_none():
    assert Router._normalize_strategy(None) is None
    assert Router._normalize_strategy("simple-shuffle") == "simple-shuffle"
    assert (
        Router._normalize_strategy(RoutingStrategy.LATENCY_BASED)
        == RoutingStrategy.LATENCY_BASED.value
    )


def test_validate_routing_strategy_accepts_valid_and_rejects_invalid():
    router = _build_router()
    # Valid: string, enum, None
    router._validate_routing_strategy("simple-shuffle")
    router._validate_routing_strategy(RoutingStrategy.LATENCY_BASED)
    router._validate_routing_strategy(None)
    with pytest.raises(ValueError, match="Invalid routing_strategy"):
        router._validate_routing_strategy("not-a-real-strategy")


def test_build_strategy_selector_returns_none_for_simple_shuffle(monkeypatch):
    monkeypatch.setattr(litellm, "callbacks", [])
    monkeypatch.setattr(litellm, "input_callback", [])
    router = _build_router()
    assert (
        router._build_strategy_selector(
            strategy="simple-shuffle",
            routing_strategy_args={},
            register_callbacks=False,
        )
        is None
    )


def test_build_strategy_selector_constructs_for_known_strategies(monkeypatch):
    monkeypatch.setattr(litellm, "callbacks", [])
    monkeypatch.setattr(litellm, "input_callback", [])
    router = _build_router()
    selector = router._build_strategy_selector(
        strategy="latency-based-routing",
        routing_strategy_args={"ttl": 30},
        register_callbacks=False,
    )
    assert selector is not None


def test_unregister_router_selectors_removes_by_identity(monkeypatch):
    monkeypatch.setattr(litellm, "callbacks", [])
    monkeypatch.setattr(litellm, "input_callback", [])
    router = _build_router()
    selector = router._build_strategy_selector(
        strategy="least-busy",
        routing_strategy_args={},
        register_callbacks=True,
    )
    assert selector in litellm.callbacks
    assert selector in litellm.input_callback
    router._unregister_router_selectors([selector])
    assert all(c is not selector for c in litellm.callbacks)
    assert all(c is not selector for c in litellm.input_callback)


def test_init_routing_groups_with_none_clears_state():
    router = _build_router(
        routing_groups=[
            {
                "group_name": "fast",
                "models": ["filtered-model"],
                "routing_strategy": "least-busy",
            }
        ]
    )
    assert router._routing_groups
    router._init_routing_groups(None)
    assert router._routing_groups == {}
    assert router._model_to_group == {}
    assert router._group_selectors == {}


@pytest.mark.asyncio
async def test_select_deployment_async_returns_none_without_selector():
    router = _build_router()
    result = await router._select_deployment_async(
        strategy="simple-shuffle",
        selector=None,
        model="filtered-model",
        healthy_deployments=[],
        messages=None,
        input=None,
        request_kwargs=None,
    )
    assert result is None


def test_select_deployment_sync_returns_none_without_selector():
    router = _build_router()
    result = router._select_deployment_sync(
        strategy="simple-shuffle",
        selector=None,
        model="filtered-model",
        healthy_deployments=[],
        messages=None,
        input=None,
        request_kwargs=None,
    )
    assert result is None


def test_select_deployment_sync_does_not_dispatch_cost_based_routing():
    """
    `LowestCostLoggingHandler` only implements `async_get_available_deployments`,
    so the sync dispatch must fall through to None for `cost-based-routing`
    instead of raising AttributeError.
    """
    router = _build_router(
        routing_strategy="simple-shuffle",
        routing_groups=[
            {
                "group_name": "cheap",
                "models": ["filtered-model"],
                "routing_strategy": "cost-based-routing",
            }
        ],
    )
    cost_selector = router._group_selectors["cheap"]["cost-based-routing"]
    assert not hasattr(cost_selector, "get_available_deployments"), (
        "test premise broken: LowestCostLoggingHandler unexpectedly grew a "
        "sync method — revisit the sync dispatch arm."
    )

    result = router._select_deployment_sync(
        strategy="cost-based-routing",
        selector=cost_selector,
        model="filtered-model",
        healthy_deployments=[],
        messages=None,
        input=None,
        request_kwargs=None,
    )
    assert result is None


@pytest.mark.asyncio
async def test_async_dispatch_falls_back_to_sync_for_usage_based_routing_v1():
    """
    `LowestTPMLoggingHandler` (v1) only implements the sync
    `get_available_deployments`. The async dispatch must call the sync method
    on the v1 selector instead of awaiting a non-existent async one.
    """
    router = _build_router(
        routing_strategy="simple-shuffle",
        routing_groups=[
            {
                "group_name": "v1",
                "models": ["filtered-model"],
                "routing_strategy": "usage-based-routing",
            }
        ],
    )
    v1_selector = router._group_selectors["v1"]["usage-based-routing"]
    assert not hasattr(v1_selector, "async_get_available_deployments"), (
        "test premise broken: LowestTPMLoggingHandler v1 unexpectedly grew an "
        "async method — revisit the async dispatch arm."
    )

    with patch.object(
        v1_selector,
        "get_available_deployments",
        wraps=v1_selector.get_available_deployments,
    ) as v1_spy:
        await router._select_deployment_async(
            strategy="usage-based-routing",
            selector=v1_selector,
            model="filtered-model",
            healthy_deployments=[
                {
                    "model_name": "filtered-model",
                    "litellm_params": {"model": "openai/gpt-4o"},
                    "model_info": {"id": "deploy-1"},
                }
            ],
            messages=[{"role": "user", "content": "hi"}],
            input=None,
            request_kwargs={},
        )

    assert v1_spy.called, "async dispatch must route v1 strategy through sync method"


def test_request_routing_strategy_override_beats_top_level():
    router = _build_router(routing_strategy="least-busy")
    strategy, selector = router._get_routing_context(
        "other-model", {"routing_strategy": "simple-shuffle"}
    )
    assert strategy == "simple-shuffle"
    assert selector is None


def test_request_routing_strategy_override_beats_explicit_group():
    router = _build_router(
        routing_strategy="simple-shuffle",
        routing_groups=[
            {
                "group_name": "fast",
                "models": ["filtered-model"],
                "routing_strategy": "latency-based-routing",
            }
        ],
    )
    strategy, _ = router._get_routing_context(
        "filtered-model", {"routing_strategy": "least-busy"}
    )
    assert strategy == "least-busy"


def test_request_routing_strategy_override_builds_and_caches_selector():
    router = _build_router(routing_strategy="simple-shuffle")
    strategy, selector = router._get_routing_context(
        "other-model", {"routing_strategy": "latency-based-routing"}
    )
    assert strategy == "latency-based-routing"
    assert selector is not None
    _, selector_again = router._get_routing_context(
        "other-model", {"routing_strategy": "latency-based-routing"}
    )
    assert selector_again is selector


def test_request_routing_strategy_override_matching_global_reuses_default_selector():
    router = _build_router(routing_strategy="least-busy")
    _, selector = router._get_routing_context(
        "other-model", {"routing_strategy": "least-busy"}
    )
    assert selector is router.leastbusy_logger
    assert router._override_selectors == {}


def test_invalid_request_routing_strategy_override_falls_back():
    router = _build_router(routing_strategy="least-busy")
    strategy, selector = router._get_routing_context(
        "other-model", {"routing_strategy": "not-a-real-strategy"}
    )
    assert strategy == "least-busy"
    assert selector is router.leastbusy_logger


def test_no_override_key_keeps_existing_behavior():
    router = _build_router(routing_strategy="least-busy")
    strategy, _ = router._get_routing_context("other-model", {"messages": []})
    assert strategy == "least-busy"
    strategy_none_kwargs, _ = router._get_routing_context("other-model", None)
    assert strategy_none_kwargs == "least-busy"


def test_request_routing_strategy_override_helper_validates_directly():
    router = _build_router(routing_strategy="least-busy")
    assert router._get_request_routing_strategy_override({"routing_strategy": "simple-shuffle"}) == "simple-shuffle"
    assert router._get_request_routing_strategy_override({"routing_strategy": RoutingStrategy.LEAST_BUSY}) == "least-busy"
    assert router._get_request_routing_strategy_override({"routing_strategy": "lar1"}) is None
    assert router._get_request_routing_strategy_override({"routing_strategy": {"bad": "type"}}) is None
    assert router._get_request_routing_strategy_override({}) is None
    assert router._get_request_routing_strategy_override(None) is None


def test_override_strategy_selector_helper_builds_per_strategy():
    router = _build_router(routing_strategy="least-busy")
    latency_selector = router._get_override_strategy_selector("latency-based-routing")
    assert latency_selector is not None
    assert router._get_override_strategy_selector("latency-based-routing") is latency_selector
    assert router._get_override_strategy_selector("least-busy") is router.leastbusy_logger
    assert router._get_override_strategy_selector("simple-shuffle") is None


def test_strategy_reinit_unregisters_override_selectors():
    router = _build_router(routing_strategy="least-busy")
    override_selector = router._get_override_strategy_selector("latency-based-routing")
    assert override_selector is not None
    assert not any(cb is override_selector for cb in litellm.callbacks)

    router.update_settings(routing_strategy="latency-based-routing")

    assert router._override_selectors == {}
    assert not any(cb is override_selector for cb in litellm.callbacks)
    assert router._get_override_strategy_selector("latency-based-routing") is router.lowestlatency_logger


def _single_latency_group():
    return [{"group_name": "g1", "models": ["filtered-model"], "routing_strategy": "latency-based-routing"}]


def _assert_still_routes_with_original_group(router, selector):
    assert list(router._routing_groups) == ["g1"]
    assert router._model_to_group == {"filtered-model": "g1"}
    assert router._group_selectors["g1"]["latency-based-routing"] is selector
    assert router._get_routing_context("filtered-model", None) == ("latency-based-routing", selector)
    assert sum(1 for cb in litellm.callbacks if cb is selector) == 1


def test_failed_routing_groups_update_keeps_previous_groups(monkeypatch):
    monkeypatch.setattr(litellm, "callbacks", [])
    monkeypatch.setattr(litellm, "input_callback", [])
    router = _build_router(routing_groups=_single_latency_group())
    selector = router._group_selectors["g1"]["latency-based-routing"]

    with pytest.raises(ValueError, match="appears in"):
        router.update_settings(
            routing_groups=[
                *_single_latency_group(),
                {"group_name": "g2", "models": ["filtered-model"], "routing_strategy": "least-busy"},
            ],
        )

    _assert_still_routes_with_original_group(router, selector)
    assert sum(1 for cb in litellm.callbacks if type(cb) is not type(selector)) == 0
    assert litellm.input_callback == []


def test_failed_routing_groups_update_does_not_poison_later_strategy_changes(monkeypatch):
    monkeypatch.setattr(litellm, "callbacks", [])
    monkeypatch.setattr(litellm, "input_callback", [])
    router = _build_router(routing_groups=_single_latency_group())

    with pytest.raises(ValueError, match="appears in"):
        router.update_settings(
            routing_groups=[
                *_single_latency_group(),
                {"group_name": "g2", "models": ["filtered-model"], "routing_strategy": "least-busy"},
            ],
        )

    router.update_settings(routing_strategy="least-busy")

    assert list(router._routing_groups) == ["g1"]
    assert [g["group_name"] for g in router.get_settings()["routing_groups"]] == ["g1"]


def test_overlap_error_names_every_conflicting_model():
    with pytest.raises(ValueError, match="appears in") as exc_info:
        _build_router(
            routing_groups=[
                {
                    "group_name": "g1",
                    "models": ["filtered-model", "other-model"],
                    "routing_strategy": "latency-based-routing",
                },
                {
                    "group_name": "g2",
                    "models": ["filtered-model", "other-model"],
                    "routing_strategy": "least-busy",
                },
            ],
        )
    message = str(exc_info.value)
    assert "'filtered-model' appears in 'g1' and 'g2'" in message
    assert "'other-model' appears in 'g1' and 'g2'" in message


def test_invalid_group_strategy_keeps_previous_groups(monkeypatch):
    monkeypatch.setattr(litellm, "callbacks", [])
    monkeypatch.setattr(litellm, "input_callback", [])
    router = _build_router(routing_groups=_single_latency_group())
    selector = router._group_selectors["g1"]["latency-based-routing"]

    with pytest.raises(ValueError, match="Invalid routing_strategy"):
        router.update_settings(
            routing_groups=[
                {"group_name": "g2", "models": ["other-model"], "routing_strategy": "not-a-real-strategy"},
            ],
        )

    _assert_still_routes_with_original_group(router, selector)


def test_unbuildable_group_selector_keeps_previous_groups(monkeypatch):
    monkeypatch.setattr(litellm, "callbacks", [])
    monkeypatch.setattr(litellm, "input_callback", [])
    router = _build_router(routing_groups=_single_latency_group())
    selector = router._group_selectors["g1"]["latency-based-routing"]

    with pytest.raises(ValidationError, match="ttl"):
        router.update_settings(
            routing_groups=[
                {"group_name": "g0", "models": ["other-model"], "routing_strategy": "least-busy"},
                *_single_latency_group(),
                {
                    "group_name": "g2",
                    "models": ["other-model-2"],
                    "routing_strategy": "latency-based-routing",
                    "routing_strategy_args": {"ttl": "not-a-number"},
                },
            ],
        )

    _assert_still_routes_with_original_group(router, selector)
    assert litellm.callbacks == [selector]
    assert litellm.input_callback == []


def test_register_router_selector_wires_only_the_hooks_the_strategy_needs(monkeypatch):
    monkeypatch.setattr(litellm, "callbacks", [])
    monkeypatch.setattr(litellm, "input_callback", [])
    router = _build_router()
    least_busy = router._build_strategy_selector(
        strategy="least-busy", routing_strategy_args={}, register_callbacks=False
    )
    latency = router._build_strategy_selector(
        strategy="latency-based-routing", routing_strategy_args={}, register_callbacks=False
    )
    assert least_busy is not None and latency is not None
    assert litellm.callbacks == [] and litellm.input_callback == []

    router._register_router_selector(least_busy)
    router._register_router_selector(latency)

    assert [cb for cb in litellm.callbacks if cb is least_busy or cb is latency] == [least_busy, latency]
    assert litellm.input_callback == [least_busy]


def test_replace_routing_groups_swaps_state_and_callbacks_in_one_step(monkeypatch):
    monkeypatch.setattr(litellm, "callbacks", [])
    monkeypatch.setattr(litellm, "input_callback", [])
    router = _build_router(routing_groups=_single_latency_group())
    old_selector = router._group_selectors["g1"]["latency-based-routing"]
    new_selector = router._build_strategy_selector(
        strategy="least-busy", routing_strategy_args={}, register_callbacks=False
    )
    assert new_selector is not None

    router._replace_routing_groups(
        (
            (RoutingGroup(group_name="g2", models=["other-model"], routing_strategy="least-busy"), new_selector),
            (RoutingGroup(group_name="g3", models=["other-model-2"], routing_strategy="simple-shuffle"), None),
        )
    )

    assert list(router._routing_groups) == ["g2", "g3"]
    assert router._model_to_group == {"other-model": "g2", "other-model-2": "g3"}
    assert router._group_selectors == {"g2": {"least-busy": new_selector}, "g3": {}}
    assert router._get_routing_context("other-model", None) == ("least-busy", new_selector)
    assert router._get_routing_context("filtered-model", None)[0] == router.routing_strategy
    assert all(cb is not old_selector for cb in litellm.callbacks)
    assert sum(1 for cb in litellm.callbacks if cb is new_selector) == 1
    assert litellm.input_callback == [new_selector]


def test_override_selectors_are_not_registered_process_wide(monkeypatch):
    monkeypatch.setattr(litellm, "callbacks", [])
    monkeypatch.setattr(litellm, "input_callback", [])
    router = _build_router(routing_strategy="simple-shuffle")
    v2_selector = router._get_override_strategy_selector("usage-based-routing-v2")
    least_busy_selector = router._get_override_strategy_selector("least-busy")
    assert v2_selector is not None and least_busy_selector is not None

    assert litellm.callbacks == []
    assert litellm.input_callback == []


def _rpm_limited_model_list():
    return [
        {
            "model_name": "other-model",
            "litellm_params": {
                "model": "openai/gpt-4o",
                "api_key": "sk-test-3",
                "api_base": "https://example.invalid",
                "rpm": 1,
            },
            "model_info": {"id": "deploy-3"},
        },
    ]


async def _mock_completion(router, **override):
    return await router.acompletion(
        model="other-model", messages=[{"role": "user", "content": "hi"}], mock_response="ok", **override
    )


@pytest.mark.asyncio
async def test_usage_based_v2_override_stays_scoped_to_the_request_that_asked_for_it():
    router = Router(model_list=_rpm_limited_model_list(), routing_strategy="simple-shuffle", num_retries=0)

    await _mock_completion(router, routing_strategy="usage-based-routing-v2")
    override_selector = router._override_selectors["usage-based-routing-v2"]
    assert not any(cb is override_selector for cb in litellm.callbacks)

    with patch.object(
        override_selector, "async_pre_call_check", wraps=override_selector.async_pre_call_check
    ) as pre_call_spy:
        for _ in range(2):
            plain = await _mock_completion(router)
            assert plain.choices[0].message.content == "ok"
        assert not pre_call_spy.called

        with pytest.raises(litellm.RateLimitError):
            await _mock_completion(router, routing_strategy="usage-based-routing-v2")


def test_sync_usage_based_v2_override_stays_scoped_to_the_request_that_asked_for_it():
    router = Router(model_list=_rpm_limited_model_list(), routing_strategy="simple-shuffle", num_retries=0)
    messages = [{"role": "user", "content": "hi"}]

    router.completion(
        model="other-model", messages=messages, mock_response="ok", routing_strategy="usage-based-routing-v2"
    )
    override_selector = router._override_selectors["usage-based-routing-v2"]
    assert not any(cb is override_selector for cb in litellm.callbacks)

    for _ in range(2):
        plain = router.completion(model="other-model", messages=messages, mock_response="ok")
        assert plain.choices[0].message.content == "ok"

    with pytest.raises(ValueError, match="No deployments available"):
        router.completion(
            model="other-model", messages=messages, mock_response="ok", routing_strategy="usage-based-routing-v2"
        )


@pytest.mark.asyncio
async def test_override_selector_pre_call_check_only_runs_for_override_selectors(monkeypatch):
    monkeypatch.setattr(litellm, "callbacks", [])
    deployment = _rpm_limited_model_list()[0]

    override_router = Router(model_list=_rpm_limited_model_list(), routing_strategy="simple-shuffle")
    override_selector = override_router._get_override_strategy_selector("usage-based-routing-v2")
    await override_router._async_override_selector_pre_call_check(
        "usage-based-routing-v2", override_selector, deployment, None
    )
    with pytest.raises(litellm.RateLimitError):
        override_router._override_selector_pre_call_check("usage-based-routing-v2", override_selector, deployment)

    default_router = Router(model_list=_rpm_limited_model_list(), routing_strategy="usage-based-routing-v2")
    for _ in range(2):
        await default_router._async_override_selector_pre_call_check(
            "usage-based-routing-v2", default_router.lowesttpm_logger_v2, deployment, None
        )
        default_router._override_selector_pre_call_check(
            "usage-based-routing-v2", default_router.lowesttpm_logger_v2, deployment
        )
    await default_router._async_override_selector_pre_call_check(None, None, deployment, None)
    default_router._override_selector_pre_call_check(None, None, deployment)


@pytest.mark.asyncio
async def test_usage_based_v2_override_enforces_rpm_when_a_specific_deployment_is_requested():
    router = Router(model_list=_rpm_limited_model_list(), routing_strategy="simple-shuffle", num_retries=0)
    kwargs = {"model": "deploy-3", "messages": [{"role": "user", "content": "hi"}], "mock_response": "ok"}

    first = await router.acompletion(**kwargs, routing_strategy="usage-based-routing-v2")
    assert first.choices[0].message.content == "ok"
    with pytest.raises(litellm.RateLimitError):
        await router.acompletion(**kwargs, routing_strategy="usage-based-routing-v2")
    assert (await router.acompletion(**kwargs)).choices[0].message.content == "ok"


def test_sync_usage_based_v2_override_enforces_rpm_when_a_specific_deployment_is_requested():
    router = Router(model_list=_rpm_limited_model_list(), routing_strategy="simple-shuffle", num_retries=0)
    kwargs = {"model": "deploy-3", "messages": [{"role": "user", "content": "hi"}], "mock_response": "ok"}

    first = router.completion(**kwargs, routing_strategy="usage-based-routing-v2")
    assert first.choices[0].message.content == "ok"
    with pytest.raises(litellm.RateLimitError):
        router.completion(**kwargs, routing_strategy="usage-based-routing-v2")
    assert router.completion(**kwargs).choices[0].message.content == "ok"


def _pass_through_rpm_limited_model_list():
    deployment = _rpm_limited_model_list()[0]
    return [{**deployment, "litellm_params": {**deployment["litellm_params"], "use_in_pass_through": True}}]


@pytest.mark.asyncio
async def test_async_early_return_paths_run_the_override_pre_call_check():
    router = Router(model_list=_pass_through_rpm_limited_model_list(), routing_strategy="simple-shuffle")
    override = {"routing_strategy": "usage-based-routing-v2"}

    pinned = await router.async_get_available_deployment(
        model="other-model", request_kwargs={**override, "_encrypted_content_affinity_pinned": True}
    )
    assert pinned["model_info"]["id"] == "deploy-3"
    with pytest.raises(litellm.RateLimitError):
        await router.async_get_available_deployment_for_pass_through(model="deploy-3", request_kwargs=override)
    plain = await router.async_get_available_deployment_for_pass_through(model="deploy-3", request_kwargs={})
    assert plain["model_info"]["id"] == "deploy-3"


def test_sync_pass_through_specific_deployment_runs_the_override_pre_call_check():
    router = Router(model_list=_pass_through_rpm_limited_model_list(), routing_strategy="simple-shuffle")
    override = {"routing_strategy": "usage-based-routing-v2"}

    first = router.get_available_deployment_for_pass_through(model="deploy-3", request_kwargs=override)
    assert first["model_info"]["id"] == "deploy-3"
    with pytest.raises(litellm.RateLimitError):
        router.get_available_deployment_for_pass_through(model="deploy-3", request_kwargs=override)
    plain = router.get_available_deployment_for_pass_through(model="deploy-3", request_kwargs={})
    assert plain["model_info"]["id"] == "deploy-3"


def _two_deployment_model_list(**d1_params: object) -> list[dict[str, object]]:
    return [
        {
            "model_name": "grp",
            "litellm_params": {"model": "openai/gpt-4o", "api_key": "sk-test-1", "mock_response": "ok", **d1_params},
            "model_info": {"id": "d1"},
        },
        {
            "model_name": "grp",
            "litellm_params": {"model": "openai/gpt-4o", "api_key": "sk-test-2", "mock_response": "ok"},
            "model_info": {"id": "d2"},
        },
    ]


def _proxy_shaped_request(**data: object) -> dict[str, object]:
    """The proxy builds the request's `Logging` object before it hands the call to the router."""
    logging_obj, kwargs = function_setup(
        "acompletion",
        Rules(),
        datetime.datetime.now(),
        litellm_call_id=str(uuid.uuid4()),
        messages=[{"role": "user", "content": "hi"}],
        **data,
    )
    return {**kwargs, "litellm_logging_obj": logging_obj}


async def _async_override_pick(router: Router, strategy: str) -> str:
    deployment = await router.async_get_available_deployment(
        "grp", request_kwargs=_proxy_shaped_request(model="grp", routing_strategy=strategy)
    )
    return deployment["model_info"]["id"]


def _sync_override_pick(router: Router, strategy: str) -> str:
    deployment = router.get_available_deployment(
        "grp", request_kwargs=_proxy_shaped_request(model="grp", routing_strategy=strategy)
    )
    return deployment["model_info"]["id"]


def _in_flight(router: Router, deployment_id: str) -> int | None:
    return router.cache.get_cache(f"grp_request_count:{deployment_id}")


async def _async_wait_until(predicate: Callable[[], bool]) -> None:
    for _ in range(100):
        if predicate():
            return
        await asyncio.sleep(0.02)
    raise AssertionError("lifecycle callback never reached the override selector")


def _sync_wait_until(predicate: Callable[[], bool]) -> None:
    for _ in range(100):
        if predicate():
            return
        time.sleep(0.02)
    raise AssertionError("lifecycle callback never reached the override selector")


def _selector_is_not_global(selector: CustomLogger) -> bool:
    global_lists = (
        litellm.callbacks,
        litellm.input_callback,
        litellm.success_callback,
        litellm.failure_callback,
        litellm._async_success_callback,
        litellm._async_failure_callback,
    )
    return not any(cb is selector for cbs in global_lists for cb in cbs)


@pytest.mark.asyncio
async def test_least_busy_override_sees_the_overriding_request_in_flight():
    router = Router(model_list=_two_deployment_model_list(), routing_strategy="simple-shuffle", num_retries=0)

    stream = await router.acompletion(**_proxy_shaped_request(model="grp", routing_strategy="least-busy", stream=True))
    busy = stream._hidden_params["model_id"]
    idle = "d2" if busy == "d1" else "d1"
    assert [await _async_override_pick(router, "least-busy") for _ in range(3)] == [idle, idle, idle]

    async for _ in stream:
        pass
    await _async_wait_until(lambda: _in_flight(router, busy) == 0)
    assert await _async_override_pick(router, "least-busy") == "d1"
    assert _selector_is_not_global(router._override_selectors["least-busy"])


def test_sync_least_busy_override_sees_the_overriding_request_in_flight():
    router = Router(model_list=_two_deployment_model_list(), routing_strategy="simple-shuffle", num_retries=0)

    stream = router.completion(**_proxy_shaped_request(model="grp", routing_strategy="least-busy", stream=True))
    busy = stream._hidden_params["model_id"]
    idle = "d2" if busy == "d1" else "d1"
    assert [_sync_override_pick(router, "least-busy") for _ in range(3)] == [idle, idle, idle]

    for _ in stream:
        pass
    _sync_wait_until(lambda: _in_flight(router, busy) == 0)
    assert _sync_override_pick(router, "least-busy") == "d1"
    assert _selector_is_not_global(router._override_selectors["least-busy"])


@pytest.mark.asyncio
async def test_least_busy_override_releases_the_slot_when_the_overriding_request_fails():
    router = Router(
        model_list=_two_deployment_model_list(mock_response="litellm.InternalServerError"),
        routing_strategy="simple-shuffle",
        num_retries=0,
    )

    with pytest.raises(litellm.InternalServerError):
        await router.acompletion(**_proxy_shaped_request(model="grp", routing_strategy="least-busy"))

    await _async_wait_until(lambda: _in_flight(router, "d1") == 0)
    assert await _async_override_pick(router, "least-busy") == "d1"
    assert _selector_is_not_global(router._override_selectors["least-busy"])


@pytest.mark.asyncio
async def test_latency_based_override_learns_from_the_overriding_requests():
    router = Router(
        model_list=_two_deployment_model_list(mock_delay=0.05), routing_strategy="simple-shuffle", num_retries=0
    )

    def samples(deployment_id: str) -> list[float]:
        recorded = (router.cache.get_cache("grp_map") or {}).get(deployment_id, {}).get("latency", [])
        return [latency for latency in recorded if latency > 0]

    async def overriding_call() -> str:
        sampled_before = {"d1": len(samples("d1")), "d2": len(samples("d2"))}
        response = await router.acompletion(
            **_proxy_shaped_request(model="grp", routing_strategy="latency-based-routing")
        )
        deployment_id = response._hidden_params["model_id"]
        await _async_wait_until(lambda: len(samples(deployment_id)) > sampled_before[deployment_id])
        return deployment_id

    served = [await overriding_call() for _ in range(6)]

    assert "d1" in served
    assert served[2:] == ["d2"] * 4
    assert _selector_is_not_global(router._override_selectors["latency-based-routing"])


def test_override_selector_is_bound_only_to_the_request_that_asked_for_it():
    router = Router(model_list=_two_deployment_model_list(), routing_strategy="simple-shuffle")
    overriding = _proxy_shaped_request(model="grp", routing_strategy="least-busy")
    plain = _proxy_shaped_request(model="grp")

    router.get_available_deployment("grp", request_kwargs=overriding)
    router.get_available_deployment("grp", request_kwargs=overriding)
    router.get_available_deployment("grp", request_kwargs=plain)

    selector = router._override_selectors["least-busy"]
    bound = overriding["litellm_logging_obj"]
    for callbacks in (
        bound.dynamic_input_callbacks,
        bound.dynamic_success_callbacks,
        bound.dynamic_async_success_callbacks,
        bound.dynamic_failure_callbacks,
        bound.dynamic_async_failure_callbacks,
    ):
        assert callbacks == [selector]
    unbound = plain["litellm_logging_obj"]
    assert unbound.dynamic_input_callbacks is None and unbound.dynamic_success_callbacks is None
    assert unbound.dynamic_failure_callbacks is None and unbound.dynamic_async_failure_callbacks is None


def test_override_matching_the_router_strategy_is_not_bound_twice():
    router = Router(model_list=_two_deployment_model_list(), routing_strategy="least-busy")
    request = _proxy_shaped_request(model="grp", routing_strategy="least-busy")

    router.get_available_deployment("grp", request_kwargs=request)

    assert request["litellm_logging_obj"].dynamic_input_callbacks is None


@pytest.mark.asyncio
async def test_override_matching_a_routing_group_strategy_records_each_request_once():
    router = Router(
        model_list=_two_deployment_model_list(),
        routing_strategy="simple-shuffle",
        routing_groups=[RoutingGroup(group_name="lat", models=["grp"], routing_strategy="latency-based-routing")],
        num_retries=0,
    )
    request = _proxy_shaped_request(model="grp", routing_strategy="latency-based-routing")
    assert router._globally_registered_strategies() == {"simple-shuffle", "latency-based-routing"}

    response = await router.acompletion(**request)
    deployment_id = response._hidden_params["model_id"]
    await _async_wait_until(lambda: (router.cache.get_cache("grp_map") or {}).get(deployment_id) is not None)

    assert len(router.cache.get_cache("grp_map")[deployment_id]["latency"]) == 1
    assert request["litellm_logging_obj"].dynamic_success_callbacks is None


def test_bind_override_selector_to_request_binds_once_and_ignores_requests_without_logging():
    router = Router(model_list=_two_deployment_model_list(), routing_strategy="simple-shuffle")
    selector = router._get_override_strategy_selector("least-busy")
    request = _proxy_shaped_request(model="grp", routing_strategy="least-busy")
    request["litellm_logging_obj"].dynamic_success_callbacks = ["langfuse"]

    router._bind_override_selector_to_request("least-busy", selector, request)
    router._bind_override_selector_to_request("least-busy", selector, request)
    router._bind_override_selector_to_request("least-busy", selector, None)
    router._bind_override_selector_to_request("least-busy", selector, {"model": "grp"})

    logging_obj = request["litellm_logging_obj"]
    assert logging_obj.dynamic_success_callbacks == ["langfuse", selector]
    assert logging_obj.dynamic_input_callbacks == [selector]
    assert logging_obj.dynamic_async_failure_callbacks == [selector]
    assert _selector_is_not_global(selector)


def _quality_group(strategy="latency-based-routing"):
    return [{"group_name": "quality", "models": ["filtered-model", "other-model"], "routing_strategy": strategy}]


def test_group_name_is_callable_and_unions_member_deployments():
    router = _build_router(routing_groups=_quality_group())
    model, deployments = router._common_checks_available_deployment(model="quality")
    assert model == "quality"
    assert sorted(d["model_info"]["id"] for d in deployments) == ["deploy-1", "deploy-2", "deploy-3"]


def test_group_name_appears_in_model_names_and_model_list():
    router = _build_router(routing_groups=_quality_group())
    assert "quality" in router.get_model_names()
    rows = router.get_model_list(model_name="quality")
    assert {r["model_name"] for r in rows} == {"quality"}
    assert sorted(r["model_info"]["id"] for r in rows) == ["deploy-1", "deploy-2", "deploy-3"]


def test_get_routing_context_for_group_name_uses_group_strategy():
    router = _build_router(routing_groups=_quality_group())
    strategy, selector = router._get_routing_context("quality")
    assert strategy == "latency-based-routing"
    assert selector is router._group_selectors["quality"]["latency-based-routing"]


@pytest.mark.asyncio
async def test_group_call_dispatches_via_group_selector():
    router = _build_router(routing_groups=_quality_group())
    group_selector = router._group_selectors["quality"]["latency-based-routing"]

    with (
        patch.object(
            group_selector,
            "async_get_available_deployments",
            wraps=group_selector.async_get_available_deployments,
        ) as latency_spy,
        patch("litellm.router.simple_shuffle", wraps=litellm.router.simple_shuffle) as shuffle_spy,
    ):
        deployment = await router.async_get_available_deployment(model="quality", request_kwargs={})

        assert latency_spy.called
        assert not shuffle_spy.called
        assert deployment["model_name"] in {"filtered-model", "other-model"}


def test_group_name_colliding_with_model_name_is_shadowed_with_warning(caplog):
    import logging

    with caplog.at_level(logging.WARNING, logger="LiteLLM Router"):
        router = _build_router(
            routing_groups=[
                {"group_name": "filtered-model", "models": ["other-model"], "routing_strategy": "latency-based-routing"}
            ]
        )
    assert any("shadowed" in record.getMessage() for record in caplog.records)
    assert router.get_routing_group("filtered-model") is None
    assert router._get_routing_context("other-model")[0] == "latency-based-routing"

    model, deployments = router._common_checks_available_deployment(model="filtered-model")
    assert sorted(d["model_info"]["id"] for d in deployments) == ["deploy-1", "deploy-2"]


def test_group_name_colliding_with_model_group_alias_is_shadowed_with_warning(caplog):
    import logging

    with caplog.at_level(logging.WARNING, logger="LiteLLM Router"):
        router = Router(
            model_list=_model_list(),
            model_group_alias={"quality": "filtered-model"},
            routing_groups=_quality_group(),
        )
    assert any("shadowed" in record.getMessage() for record in caplog.records)
    assert router.get_routing_group("quality") is None

    model, deployments = router._common_checks_available_deployment(model="quality")
    assert model == "filtered-model"
    assert sorted(d["model_info"]["id"] for d in deployments) == ["deploy-1", "deploy-2"]


def test_real_model_added_later_shadows_group():
    router = _build_router(routing_groups=_quality_group())
    assert router.get_routing_group("quality") is not None

    from litellm.types.router import Deployment

    router.add_deployment(
        Deployment(
            model_name="quality",
            litellm_params={"model": "openai/gpt-4o", "api_key": "sk-test-4", "api_base": "https://example.invalid"},
            model_info={"id": "deploy-shadow"},
        )
    )
    assert router.get_routing_group("quality") is None
    model, deployments = router._common_checks_available_deployment(model="quality")
    assert [d["model_info"]["id"] for d in deployments] == ["deploy-shadow"]

    router.delete_deployment(id="deploy-shadow")
    assert "quality" not in router.model_names
    assert router.get_routing_group("quality") is not None
    _, restored = router._common_checks_available_deployment(model="quality")
    assert sorted(d["model_info"]["id"] for d in restored) == ["deploy-1", "deploy-2", "deploy-3"]


def test_group_with_no_member_deployments_raises_no_healthy():
    router = Router(
        model_list=_model_list(),
        routing_groups=[{"group_name": "empty-group", "models": ["ghost-model"], "routing_strategy": "simple-shuffle"}],
    )
    with pytest.raises(litellm.BadRequestError):
        router._common_checks_available_deployment(model="empty-group")


def test_alias_pointing_at_group_composes():
    router = Router(
        model_list=_model_list(),
        model_group_alias={"quality-alias": "quality"},
        routing_groups=_quality_group(),
    )
    model, deployments = router._common_checks_available_deployment(model="quality-alias")
    assert model == "quality"
    assert sorted(d["model_info"]["id"] for d in deployments) == ["deploy-1", "deploy-2", "deploy-3"]


def test_model_group_info_reports_group():
    router = _build_router(routing_groups=_quality_group())
    info = router.get_model_group_info("quality")
    assert info is not None
    assert info.model_group == "quality"
    assert "openai" in info.providers


def test_routing_group_has_alternatives():
    router = _build_router(routing_groups=_quality_group())
    assert router.routing_group_has_alternatives("quality") is True
    assert router.routing_group_has_alternatives("filtered-model") is False
    assert router.routing_group_has_alternatives(None) is False

    solo_router = Router(
        model_list=_model_list(),
        routing_groups=[{"group_name": "solo-group", "models": ["other-model"], "routing_strategy": "simple-shuffle"}],
    )
    assert solo_router.routing_group_has_alternatives("solo-group") is False


def test_member_direct_call_unchanged_by_callable_groups():
    router = _build_router(routing_groups=_quality_group())
    model, deployments = router._common_checks_available_deployment(model="other-model")
    assert model == "other-model"
    assert [d["model_info"]["id"] for d in deployments] == ["deploy-3"]


def test_update_settings_group_change_invalidates_model_group_info():
    router = _build_router(routing_groups=_quality_group())
    assert router.get_model_group_info("quality") is not None
    assert router.get_model_group_info("renamed-group") is None

    router.update_settings(
        routing_groups=[
            {"group_name": "renamed-group", "models": ["filtered-model"], "routing_strategy": "simple-shuffle"}
        ]
    )
    assert router.get_model_group_info("quality") is None
    info = router.get_model_group_info("renamed-group")
    assert info is not None
    assert info.model_group == "renamed-group"


def test_is_recognized_model_covers_every_virtual_model_kind():
    router = Router(
        model_list=_model_list(),
        model_group_alias={"my-alias": "filtered-model"},
        routing_groups=_quality_group(),
    )
    assert router.is_recognized_model("filtered-model") is True
    assert router.is_recognized_model("deploy-1") is True
    assert router.is_recognized_model("my-alias") is True
    assert router.is_recognized_model("quality") is True
    assert router.is_recognized_model("ghost") is False


def test_routing_group_has_alternatives_resolves_aliases():
    router = Router(
        model_list=_model_list(),
        model_group_alias={"quality-alias": "quality"},
        routing_groups=_quality_group(),
    )
    assert router.routing_group_has_alternatives("quality-alias") is True
    assert router.routing_group_has_alternatives("quality") is True


def test_group_rows_cache_invalidated_on_model_list_change():
    from litellm.types.router import Deployment

    router = _build_router(routing_groups=_quality_group())
    assert sum(1 for row in router.get_model_list() if row["model_name"] == "quality") == 3

    router.add_deployment(
        Deployment(
            model_name="filtered-model",
            litellm_params={"model": "openai/gpt-4o", "api_key": "sk-test-5", "api_base": "https://example.invalid"},
            model_info={"id": "deploy-4"},
        )
    )
    assert sum(1 for row in router.get_model_list() if row["model_name"] == "quality") == 4


def _pin_choice_to(deployment_id):
    def _pick(seq):
        for candidate in seq:
            if candidate["model_info"]["id"] == deployment_id:
                return candidate
        return seq[0]

    return _pick


async def _call_and_get_cooldowns(router, model):
    from litellm.router_utils.cooldown_handlers import _async_get_cooldown_deployments

    with (
        patch("litellm.router_strategy.simple_shuffle.random.choice", side_effect=_pin_choice_to("deploy-3")),
        pytest.raises(litellm.RateLimitError),
    ):
        await router.acompletion(
            model=model,
            messages=[{"role": "user", "content": "hi"}],
            mock_response="litellm.RateLimitError",
        )
    return await _async_get_cooldown_deployments(litellm_router_instance=router, parent_otel_span=None)


@pytest.mark.asyncio
async def test_group_call_429_registers_cooldown_end_to_end():
    router = Router(
        model_list=_model_list(),
        routing_groups=_quality_group("simple-shuffle"),
        num_retries=0,
        cooldown_time=60,
    )
    cooldown_ids = await _call_and_get_cooldowns(router, "quality")
    assert "deploy-3" in cooldown_ids


@pytest.mark.asyncio
async def test_alias_to_group_429_registers_cooldown_end_to_end():
    router = Router(
        model_list=_model_list(),
        model_group_alias={"quality-alias": "quality"},
        routing_groups=_quality_group("simple-shuffle"),
        num_retries=0,
        cooldown_time=60,
    )
    cooldown_ids = await _call_and_get_cooldowns(router, "quality-alias")
    assert "deploy-3" in cooldown_ids


@pytest.mark.asyncio
async def test_direct_single_deployment_member_429_keeps_exemption_end_to_end():
    router = Router(
        model_list=_model_list(),
        routing_groups=_quality_group("simple-shuffle"),
        num_retries=0,
        cooldown_time=60,
    )
    cooldown_ids = await _call_and_get_cooldowns(router, "other-model")
    assert "deploy-3" not in cooldown_ids


def test_group_rows_do_not_inherit_member_access_groups():
    router = Router(
        model_list=[
            {
                "model_name": "gated-member",
                "litellm_params": {"model": "openai/gpt-4o", "api_key": "sk-test"},
                "model_info": {"id": "gated-1", "access_groups": ["restricted-team"]},
            }
        ],
        routing_groups=[
            {"group_name": "gated-group", "models": ["gated-member"], "routing_strategy": "simple-shuffle"}
        ],
    )
    access_groups = router.get_model_access_groups()
    assert "gated-group" not in access_groups.get("restricted-team", [])
    assert all("access_groups" not in (row.get("model_info") or {}) for row in router.get_model_list(model_name="gated-group"))
    assert "access_groups" in router.get_model_list(model_name="gated-member")[0]["model_info"]


def test_group_rebuild_invalidates_access_groups_cache():
    router = _build_router(routing_groups=_quality_group())
    router.get_model_access_groups()
    assert router._access_groups_cache is not None

    router.update_settings(routing_groups=[])
    assert router._access_groups_cache is None


def test_get_model_list_from_routing_groups_materializes_rows():
    router = _build_router(routing_groups=_quality_group())
    rows = router.get_model_list_from_routing_groups()
    assert {row["model_name"] for row in rows} == {"quality"}
    assert router.get_model_list_from_routing_groups() is rows

    named = router.get_model_list_from_routing_groups(model_name="quality")
    assert sorted(row["model_info"]["id"] for row in named) == ["deploy-1", "deploy-2", "deploy-3"]
    assert router.get_model_list_from_routing_groups(model_name="filtered-model") == ()


def test_get_routing_group_deployments_unions_members():
    router = _build_router(routing_groups=_quality_group())
    union = router._get_routing_group_deployments("quality")
    assert sorted(d["model_info"]["id"] for d in union) == ["deploy-1", "deploy-2", "deploy-3"]
    assert router._get_routing_group_deployments("filtered-model") is None


def test_materialize_routing_group_rows_labels_members_with_group_name():
    router = _build_router(routing_groups=_quality_group())
    group = router.get_routing_group("quality")
    rows = router._materialize_routing_group_rows((group,))
    assert {row["model_name"] for row in rows} == {"quality"}
    assert len(rows) == 3


def test_as_routing_group_row_strips_access_groups():
    source = {"model_name": "member", "model_info": {"id": "d1", "access_groups": ["restricted"]}}
    row = Router._as_routing_group_row(source)
    assert row["model_info"] == {"id": "d1"}
    assert source["model_info"]["access_groups"] == ["restricted"]


@pytest.mark.asyncio
async def test_group_call_429_cools_down_member_across_retries():
    router = Router(
        model_list=_model_list(),
        routing_groups=_quality_group("simple-shuffle"),
        num_retries=1,
        cooldown_time=60,
    )
    cooldown_ids = await _call_and_get_cooldowns(router, "quality")
    assert "deploy-3" in cooldown_ids


def _priority_group(
    name: str = "priority-group", primary: int = 1, backup: int = 2
) -> RoutingGroup:
    return RoutingGroup.model_validate({
        "group_name": name,
        "models": ["filtered-model", "other-model"],
        "routing_strategy": "priority",
        "model_priorities": {"filtered-model": primary, "other-model": backup},
    })


def _priority_deployments(
    primary_response: str = "primary", primary_blocked: bool = False, backup_response: str = "backup"
) -> list[DeploymentTypedDict]:
    return [
        {
            **deployment,
            "litellm_params": {
                **deployment["litellm_params"],
                "mock_response": (
                    primary_response if deployment["model_name"] == "filtered-model" else backup_response
                ),
                "order": 10 if deployment["model_name"] == "filtered-model" else 1,
            },
            "model_info": {
                **deployment["model_info"],
                "blocked": primary_blocked and deployment["model_name"] == "filtered-model",
            },
        }
        for deployment in _model_list()
    ]


def test_priority_group_affinity_scope_follows_group_aliases_and_settings_reload() -> None:
    router: Final = Router(
        model_list=_priority_deployments(),
        routing_groups=[
            _priority_group(),
            RoutingGroup(group_name="legacy", models=["filtered-model"], routing_strategy="simple-shuffle"),
        ],
        model_group_alias={"priority-alias": "priority-group"},
    )
    requested_models: Final = (
        "priority-group", "priority-alias", "filtered-model", "other-model", "legacy", "missing"
    )
    assert tuple(router._is_priority_routing_group(model) for model in requested_models) == (
        True, True, False, False, False, False
    )

    router.update_settings(routing_groups=[{
        "group_name": "priority-group",
        "models": ["filtered-model", "other-model"],
        "routing_strategy": "simple-shuffle",
    }])
    assert tuple(router._is_priority_routing_group(model) for model in requested_models) == (False,) * 6


@pytest.mark.parametrize("asynchronous", [False, True])
@pytest.mark.parametrize("requested_model", ["priority-group", "priority-alias"])
@pytest.mark.asyncio
async def test_priority_group_always_uses_primary_when_healthy(
    asynchronous: bool, requested_model: str
) -> None:
    router: Final = Router(
        model_list=_priority_deployments(),
        routing_groups=[_priority_group()],
        model_group_alias={"priority-alias": "priority-group"},
        num_retries=0,
    )
    request: Final = {"model": requested_model, "messages": [{"role": "user", "content": "hi"}]}
    response: Final = (
        await router.acompletion(**request) if asynchronous else router.completion(**request)
    )
    assert response.choices[0].message.content == "primary"


@pytest.mark.parametrize("requested_model", ["priority-group", "priority-alias"])
@pytest.mark.asyncio
async def test_priority_group_fails_over_without_retries_and_leaves_direct_calls_unchanged(
    requested_model: str,
) -> None:
    router: Final = Router(
        model_list=_priority_deployments(primary_response="litellm.RateLimitError"),
        routing_groups=[_priority_group()],
        model_group_alias={"priority-alias": "priority-group"},
        num_retries=0,
        disable_cooldowns=True,
    )
    response: Final = await router.acompletion(
        model=requested_model, messages=[{"role": "user", "content": "hi"}]
    )
    assert response.choices[0].message.content == "backup"
    with pytest.raises(litellm.RateLimitError):
        await router.acompletion(model="filtered-model", messages=[{"role": "user", "content": "hi"}])


@pytest.mark.asyncio
async def test_opposite_priority_groups_preserve_deployments_and_legacy_member_policy() -> None:
    router: Final = Router(
        model_list=_priority_deployments(),
        routing_strategy="latency-based-routing",
        routing_groups=[
            RoutingGroup(group_name="legacy", models=["filtered-model"], routing_strategy="least-busy"),
            _priority_group(),
            _priority_group("reverse-group", primary=2, backup=1),
        ],
    )
    before: Final = router.get_model_list(model_name="filtered-model")
    forward: Final = await router.acompletion(
        model="priority-group", messages=[{"role": "user", "content": "hi"}]
    )
    reverse: Final = await router.acompletion(
        model="reverse-group", messages=[{"role": "user", "content": "hi"}]
    )
    assert (forward.choices[0].message.content, reverse.choices[0].message.content) == ("primary", "backup")
    assert router._get_routing_context("filtered-model")[0] == "least-busy"
    assert router._get_routing_context("other-model")[0] == "latency-based-routing"
    assert router.get_model_list(model_name="filtered-model") == before
    assert all(deployment["litellm_params"]["order"] == 10 for deployment in before)


@pytest.mark.parametrize("backup_priority, expected", [(1, "backup"), (2, "primary")])
@pytest.mark.asyncio
async def test_priority_group_weights_select_only_within_the_first_eligible_level(
    backup_priority: int, expected: str
) -> None:
    router: Final = Router(
        model_list=_priority_deployments(), routing_groups=[_priority_group(backup=backup_priority)]
    )
    response: Final = await router.acompletion(
        model="priority-group",
        messages=[{"role": "user", "content": "hi"}],
        _router_weights={"priority-group": {"deploy-1": 0, "deploy-2": 0, "deploy-3": 1}},
    )
    assert response.choices[0].message.content == expected


@pytest.mark.asyncio
async def test_priority_group_skips_paused_primary_and_returns_to_it_after_recovery() -> None:
    router: Final = Router(
        model_list=_priority_deployments(primary_blocked=True), routing_groups=[_priority_group()]
    )
    paused: Final = await router.acompletion(
        model="priority-group", messages=[{"role": "user", "content": "hi"}]
    )
    router.set_model_list(_priority_deployments())
    recovered: Final = await router.acompletion(
        model="priority-group", messages=[{"role": "user", "content": "hi"}]
    )
    assert (paused.choices[0].message.content, recovered.choices[0].message.content) == ("backup", "primary")


@pytest.mark.parametrize("controls", [{"disable_fallbacks": True}, {"max_fallbacks": 0}])
@pytest.mark.asyncio
async def test_priority_group_respects_request_fallback_controls(controls: dict[str, object]) -> None:
    router: Final = Router(
        model_list=_priority_deployments(primary_response="litellm.RateLimitError"),
        routing_groups=[_priority_group()],
        num_retries=0,
        disable_cooldowns=True,
    )
    with pytest.raises(litellm.RateLimitError):
        await router.acompletion(
            model="priority-group", messages=[{"role": "user", "content": "hi"}], **controls
        )


@pytest.mark.parametrize(
    "priorities",
    [None, {}, {"filtered-model": 1}, {"filtered-model": 1, "other-model": 2, "extra": 3}],
)
def test_priority_group_requires_exact_member_priorities(priorities: object) -> None:
    with pytest.raises(ValidationError):
        RoutingGroup.model_validate({
            "group_name": "priority-group",
            "models": ["filtered-model", "other-model"],
            "routing_strategy": "priority",
            "model_priorities": priorities,
        })


@pytest.mark.parametrize("priority", [True, 0, -1, 1.5, "1", 9007199254740992])
def test_priority_group_rejects_invalid_priority_values(priority: object) -> None:
    with pytest.raises(ValidationError):
        RoutingGroup.model_validate({
            "group_name": "priority-group",
            "models": ["filtered-model"],
            "routing_strategy": "priority",
            "model_priorities": {"filtered-model": priority},
        })


@pytest.mark.parametrize("models", [[], ["filtered-model", "filtered-model"]])
def test_priority_group_requires_nonempty_unique_members(models: list[str]) -> None:
    with pytest.raises(ValidationError):
        RoutingGroup.model_validate({
            "group_name": "priority-group",
            "models": models,
            "routing_strategy": "priority",
            "model_priorities": {model: 1 for model in models},
        })


@pytest.mark.parametrize(
    "changes", [{"routing_strategy": "simple-shuffle"}, {"routing_strategy_args": {"ttl": 60}}]
)
def test_priority_group_rejects_conflicting_strategy_settings(changes: dict[str, object]) -> None:
    with pytest.raises(ValidationError):
        RoutingGroup.model_validate({
            "group_name": "priority-group",
            "models": ["filtered-model"],
            "routing_strategy": "priority",
            "model_priorities": {"filtered-model": 1},
            **changes,
        })


@pytest.mark.parametrize("group_name", ["filtered-model", "priority-alias"])
def test_priority_group_rejects_names_shadowed_by_a_model_or_alias(group_name: str) -> None:
    with pytest.raises(ValueError, match=r"shadow|collid|conflict"):
        Router(
            model_list=_priority_deployments(),
            routing_groups=[_priority_group(name=group_name)],
            model_group_alias={"priority-alias": "filtered-model"},
        )


def test_priority_is_rejected_as_a_top_level_strategy() -> None:
    with pytest.raises(ValueError, match="routing_strategy"):
        _build_router(routing_strategy="priority")


@pytest.mark.asyncio
async def test_priority_group_settings_roundtrip_replaces_the_order() -> None:
    router: Final = Router(model_list=_priority_deployments(), routing_groups=[_priority_group()])
    replacement: Final = _priority_group(primary=9007199254740991, backup=1)
    router.update_settings(routing_groups=[replacement.model_dump()])
    stored: Final = router.get_settings()["routing_groups"]
    assert stored == [replacement.model_dump()]
    response: Final = await router.acompletion(
        model="priority-group", messages=[{"role": "user", "content": "hi"}]
    )
    assert response.choices[0].message.content == "backup"


def _priority_auto_router_deployment() -> DeploymentTypedDict:
    return {
        "model_name": "smart-router",
        "litellm_params": {
            "model": "auto_router/complexity_router",
            "complexity_router_config": {
                "classifier_type": "heuristic",
                "adaptive": False,
                "tiers": {
                    tier: "priority-group"
                    for tier in ("SIMPLE", "MEDIUM", "COMPLEX", "REASONING")
                },
            },
            "complexity_router_default_model": "filtered-model",
        },
    }


@pytest.mark.asyncio
async def test_auto_router_selected_priority_group_fails_over_inside_the_group() -> None:
    router: Final = Router(
        model_list=[
            *_priority_deployments(primary_response="litellm.RateLimitError"),
            _priority_auto_router_deployment(),
        ],
        routing_groups=[_priority_group()],
        num_retries=0,
        disable_cooldowns=True,
    )
    response: Final = await router.acompletion(
        model="smart-router", messages=[{"role": "user", "content": "hi"}]
    )
    assert response.choices[0].message.content == "backup"


@pytest.mark.parametrize("endpoint_request", [{"input": "hi"}, {"messages": [{"role": "user", "content": "hi"}]}])
@pytest.mark.asyncio
async def test_priority_generic_fallback_keeps_routing_controls_out_of_provider_kwargs(
    endpoint_request: dict[str, object],
) -> None:
    async def provider(model: str, **provider_kwargs: object) -> str:
        assert "_target_order" not in provider_kwargs
        assert "model_priorities" not in provider_kwargs
        model_info: Final = provider_kwargs["model_info"]
        assert isinstance(model_info, dict)
        if model_info["id"] != "deploy-3":
            raise litellm.RateLimitError(message="primary refused", model=model, llm_provider="openai")
        return "backup"

    router: Final = Router(
        model_list=_priority_deployments(),
        routing_groups=[_priority_group()],
        num_retries=0,
        disable_cooldowns=True,
    )
    response: Final = await router._ageneric_api_call_with_fallbacks(
        model="priority-group", original_function=provider, **endpoint_request
    )
    assert response == "backup"


@pytest.mark.parametrize("provider_model_id", [None, "resolved-downstream-id"])
@pytest.mark.asyncio
async def test_priority_generic_fallback_dict_preserves_served_model_id(provider_model_id: str | None) -> None:
    async def provider(model: str, **provider_kwargs: object) -> dict[str, object]:
        model_info: Final = provider_kwargs["model_info"]
        assert isinstance(model_info, dict)
        if model_info["id"] == "deploy-1":
            raise litellm.NotFoundError(message="primary missing", model=model, llm_provider="openai")
        assert model_info["id"] == "deploy-3"
        return {
            "content": "backup",
            **({"_hidden_params": {"model_id": provider_model_id}} if provider_model_id is not None else {}),
        }

    router: Final = Router(
        model_list=_priority_deployments()[::2],
        routing_groups=[_priority_group()],
        num_retries=0,
        disable_cooldowns=True,
    )
    outer_metadata: Final[dict[str, object]] = {}
    response: Final = await router._ageneric_api_call_with_fallbacks(
        model="priority-group",
        original_function=provider,
        messages=[{"role": "user", "content": "hi"}],
        litellm_metadata=outer_metadata,
    )
    primary_info: Final = outer_metadata["model_info"]
    assert isinstance(primary_info, dict)
    assert primary_info["id"] == "deploy-1"
    assert response["content"] == "backup"
    assert response["_hidden_params"]["model_id"] == (provider_model_id or "deploy-3")
    assert response["_hidden_params"]["additional_headers"]["x-litellm-attempted-fallbacks"] == 1


@pytest.mark.asyncio
async def test_priority_group_ignores_cached_backup_affinity() -> None:
    from litellm.router_utils.prompt_caching_cache import PromptCachingCache

    router: Final = Router(
        model_list=_priority_deployments(),
        routing_groups=[_priority_group()],
        optional_pre_call_checks=["prompt_caching"],
    )
    messages: Final = [{
        "role": "user",
        "content": [{"type": "text", "text": "word " * 5000, "cache_control": {"type": "ephemeral"}}],
    }]
    cache: Final = PromptCachingCache(cache=router.cache)
    await cache.async_add_model_id(model_id="deploy-3", messages=messages, tools=None)
    assert await cache.async_get_model_id(messages=messages, tools=None) == {"model_id": "deploy-3"}
    response: Final = await router.acompletion(model="priority-group", messages=messages)
    assert response.choices[0].message.content == "primary"


@pytest.mark.asyncio
async def test_priority_group_preserves_responses_continuity_on_a_backup() -> None:
    from litellm.responses.utils import ResponsesAPIRequestUtils

    router: Final = Router(
        model_list=_priority_deployments(),
        routing_groups=[_priority_group()],
        optional_pre_call_checks=["responses_api_deployment_check"],
    )
    previous_response_id: Final = ResponsesAPIRequestUtils._build_responses_api_response_id(
        custom_llm_provider="openai", model_id="deploy-3", response_id="resp-prior"
    )
    deployment: Final = await router.async_get_available_deployment(
        model="priority-group", input="continue", request_kwargs={"previous_response_id": previous_response_id}
    )
    assert deployment["model_info"]["id"] == "deploy-3"


def _recording_fallback_gate(allowed: frozenset[str], checked: list[str]) -> FallbackAccessCheck:
    async def check(*, model: str, request_kwargs: Mapping[str, object], llm_router: Router) -> bool:
        checked.append(model)
        return model in allowed

    return check


@pytest.mark.parametrize("budget_admits_group", [False, True])
@pytest.mark.asyncio
async def test_auto_priority_advance_preserves_access_scope_and_checks_paid_group_budget(
    budget_admits_group: bool,
) -> None:
    access_checks: Final[list[str]] = []
    budget_checks: Final[list[str]] = []
    router: Final = Router(
        model_list=[
            *_priority_deployments(primary_response="litellm.RateLimitError"),
            _priority_auto_router_deployment(),
        ],
        routing_groups=[_priority_group()],
        num_retries=0,
        disable_cooldowns=True,
        fallback_access_check=_recording_fallback_gate(frozenset({"smart-router"}), access_checks),
        fallback_budget_check=_recording_fallback_gate(
            frozenset({"priority-group"}) if budget_admits_group else frozenset(), budget_checks
        ),
    )
    if budget_admits_group:
        response: Final = await router.acompletion(
            model="smart-router", messages=[{"role": "user", "content": "hi"}]
        )
        assert response.choices[0].message.content == "backup"
    else:
        with pytest.raises(litellm.RateLimitError):
            await router.acompletion(model="smart-router", messages=[{"role": "user", "content": "hi"}])
    assert access_checks == []
    assert budget_checks == ["priority-group"]


@pytest.mark.asyncio
async def test_auto_priority_group_exhaustion_still_checks_external_fallback_access() -> None:
    access_checks: Final[list[str]] = []
    router: Final = Router(
        model_list=[
            *_priority_deployments(
                primary_response="litellm.RateLimitError", backup_response="litellm.RateLimitError"
            ),
            _priority_auto_router_deployment(),
            {
                **_model_list()[2],
                "model_name": "external",
                "litellm_params": {**_model_list()[2]["litellm_params"], "mock_response": "external"},
                "model_info": {"id": "external-deployment"},
            },
        ],
        routing_groups=[_priority_group()],
        fallbacks=[{"priority-group": ["external"]}],
        num_retries=0,
        disable_cooldowns=True,
        fallback_access_check=_recording_fallback_gate(frozenset({"smart-router"}), access_checks),
    )
    with pytest.raises(litellm.RateLimitError):
        await router.acompletion(model="smart-router", messages=[{"role": "user", "content": "hi"}])
    assert access_checks and set(access_checks) == {"external"}


@pytest.mark.parametrize("check_kind", ["access", "budget"])
@pytest.mark.parametrize("metadata_bucket", ["metadata", "litellm_metadata"])
@pytest.mark.asyncio
async def test_caller_cannot_spoof_a_priority_group_to_bypass_fallback_gates(
    check_kind: Literal["access", "budget"], metadata_bucket: str,
) -> None:
    checked: Final[list[str]] = []
    check: Final = _recording_fallback_gate(frozenset({"filtered-model"}), checked)
    router: Final = Router(
        model_list=_priority_deployments(primary_response="litellm.RateLimitError"),
        routing_groups=[_priority_group()],
        fallbacks=[{"filtered-model": ["priority-group"]}],
        num_retries=0,
        disable_cooldowns=True,
        fallback_access_check=check if check_kind == "access" else None,
        fallback_budget_check=check if check_kind == "budget" else None,
    )
    with pytest.raises(litellm.RateLimitError):
        await router.acompletion(
            model="filtered-model",
            messages=[{"role": "user", "content": "hi"}],
            **{metadata_bucket: {"pre_routing_selected_model": "priority-group"}},
        )
    assert checked == ["priority-group"]


def _sync_task_count() -> int:
    return sum(
        1
        for t in asyncio.all_tasks()
        if t.get_coro() is not None
        and t.get_coro().__qualname__ == "BaseRoutingStrategy.periodic_sync_in_memory_spend_with_redis"
    )


@pytest.mark.asyncio
async def test_update_settings_same_routing_strategy_args_does_not_leak_sync_tasks(monkeypatch) -> None:
    monkeypatch.setattr(litellm, "callbacks", [])
    monkeypatch.setattr(litellm, "input_callback", [])
    router: Final = Router(
        model_list=_model_list(),
        routing_strategy="usage-based-routing-v2",
        routing_strategy_args={"ttl": 60},
    )
    try:
        assert _sync_task_count() == 1
        selector_before: Final = router.lowesttpm_logger_v2

        for _ in range(5):
            router.update_settings(routing_strategy_args={"ttl": 60})
        await asyncio.sleep(0)
        assert _sync_task_count() == 1
        assert router.lowesttpm_logger_v2 is selector_before, "same routing_strategy_args must not rebuild the selector"

        router.update_settings(routing_strategy_args={"ttl": 120})
        await asyncio.sleep(0)
        assert _sync_task_count() == 1
        assert router.lowesttpm_logger_v2.routing_args.ttl == 120
    finally:
        for t in [
            t
            for t in asyncio.all_tasks()
            if t.get_coro() is not None
            and t.get_coro().__qualname__ == "BaseRoutingStrategy.periodic_sync_in_memory_spend_with_redis"
        ]:
            t.cancel()
            try:
                await t
            except asyncio.CancelledError:
                pass


class _RecordingRedisCache:
    def __init__(self) -> None:
        self.increment_lists: list[list[RedisPipelineIncrementOperation]] = []

    async def async_increment_pipeline(self, increment_list: list[RedisPipelineIncrementOperation]) -> list[float]:
        self.increment_lists.append(list(increment_list))
        return [float(op["increment_value"]) for op in increment_list]


@pytest.mark.asyncio
async def test_update_settings_changed_routing_strategy_args_flushes_replaced_selector_queue(
    monkeypatch,
) -> None:
    monkeypatch.setattr(litellm, "callbacks", [])
    monkeypatch.setattr(litellm, "input_callback", [])
    router: Final = Router(
        model_list=_model_list(),
        routing_strategy="usage-based-routing-v2",
        routing_strategy_args={"ttl": 60},
    )
    try:
        redis_cache: Final = _RecordingRedisCache()
        replaced: Final = router.lowesttpm_logger_v2
        replaced.dual_cache.redis_cache = redis_cache
        replaced.redis_increment_operation_queue.append(
            RedisPipelineIncrementOperation(key="rpm-key", increment_value=3, ttl=60)
        )

        router.update_settings(routing_strategy_args={"ttl": 120})
        await asyncio.sleep(0)
        await asyncio.gather(
            *(
                t
                for t in asyncio.all_tasks()
                if t.get_coro() is not None
                and t.get_coro().__qualname__ == "BaseRoutingStrategy._push_in_memory_increments_to_redis"
            )
        )

        assert router.lowesttpm_logger_v2 is not replaced
        assert redis_cache.increment_lists == [
            [RedisPipelineIncrementOperation(key="rpm-key", increment_value=3, ttl=60)]
        ]
        assert replaced.redis_increment_operation_queue == []
    finally:
        for t in [
            t
            for t in asyncio.all_tasks()
            if t.get_coro() is not None
            and t.get_coro().__qualname__ == "BaseRoutingStrategy.periodic_sync_in_memory_spend_with_redis"
        ]:
            t.cancel()
            try:
                await t
            except asyncio.CancelledError:
                pass