litellm/tests/unit/conftest.py
Nate Armstrong 9752869d84
feat(router): add the Rust router core with the host as its only boundary (#49513)
Adds the litellm-router-core crate, its step-driven PyO3 bridge
(RouterCoreHandle / RouterCall) and RustCore (litellm/router_rust/),
served behind LITELLM_RUST through the ROUTER route rule. The host
call-out is the only boundary: every attempt goes through Host::invoke,
there is no native executor path, and RustCore never builds, wraps or
forwards to PythonCore, so router_python_core.py stays deletable.
Attempt preparation lives in litellm/router_attempt.py so it outlives
router_python_core.py.

RustCore matches PythonCore's behavior, including its known bugs (fixed
in follow-up PRs). Routing decisions (strategy picks, routing groups,
PTU, pre-routing, pre-call checks, fallback walk, health routing, access
groups, weights, cooldowns, allowed fails, usage) live in the crate;
work that needs live Python objects (request shaping, endpoints,
streaming, model info facts, callbacks, alerting, prompt management)
lives in the host-side litellm/router_rust/ package (steps/, invoke/,
surface/) driven by Rust steps.

The crate owns the registry. It runs the deployment build loop (secret
references, ids, organization fan-out, install order) and asks the host
for what needs Python, one deployment at a time so errors surface where
Python raised them: secrets, validate, provider and added steps. It
holds the model list, the index maps and the pattern routers with the
Python core's history rules (host_index.rs), and answers the registry
lookups (lookup.rs). The Python views of that state in
litellm/router_rust/surface/ (model_list, index(), names(), patterns())
read the handle, and every write to them becomes a changeset. Model
info, deployment model info, configured limits and model group info are
merged in the crate (model_info.rs, group_info.rs) from catalog rows and
facts the host answers on a model_info step asked at query time, so
price reloads and register_model stay visible. The usage cache key
format is defined once, in the crate.

The crate records each attempt's outcome itself rather than through
litellm callbacks. A streamed attempt is deferred: it counts its
request when the stream opens, and the host's stream wrappers report
the success with the final usage, or a failure after the first chunk,
when the stream ends, as the Python core's callbacks do.

tests/unit/test_router/core characterizes PythonCore's public behavior
and runs every test on both cores, with nothing pending; only the names
of the Python core's own outcome callbacks are asserted on Python alone.

litellm.Router is now final: subclassing it raises TypeError at the class statement, and the
machinery that kept subclass overrides working is gone. The Rust core never honoured subclass
overrides, so they would have silently stopped running there. Custom routing goes through
register_plugin(), settings through apply() and outcomes through observe(). This is a breaking
change for anyone subclassing Router.

Fuzz parity fixes on top of the rebase: the Rust core keeps num_retries,
max_fallbacks, allowed_fails, retry_after, timeout, the tag settings,
routing_strategy_args and the fallback lists as given (the crate reads
them as numbers, update_settings casts as Python does); a deprecated
`model_list = [...]` write reaches the Rust core; a failed set_model_list
is all or nothing on both cores (the deployments served before the call
are put back and the error raised); config_deployments() is a generator and `del` of a plain
attribute works on the Rust core. Least-busy pick reservations from
main now work on the Rust core: its selectors share one reservation
store, routed calls reserve and free their unstarted picks, and
select() and batch cancel do not reserve. On the Python core a dropped
upsert takes its name back off deployment_names, and
get_deployment_order reads every numeric spelling as one int so mixed
orders no longer raise TypeError.

Rulings on the remaining differences, the Python core changed where it
was the odd one out: sync picks and picks by deployment id apply the
team filter as async picks do, so another team's deployment is refused;
the fallback lists and default_litellm_params in RouterSettings are
copies, so an in-place edit is ignored and settings change only through
apply(); one health record with an unreadable timestamp drops out alone;
an empty fallback entry is skipped; a bool cooldown_time is refused at
construction, a bool or string cooldown_time on a deployment or a call
falls back to the router default, and a failure status that no HTTP
status can be is recorded as 500; an infinite or NaN cost and a whole
number wider than 64 bits (ptu_shares) are refused on both cores with
the same error (litellm/router_utils/deployment_numbers.py); and a
/v1/messages stream whose provider iterator raises cools the deployment
and retries like a non-stream call, with the Rust core waiting for that
cooldown before its retry picks.

Differences the proxy parity fuzz and its regression tests found on the
Rust core: a file upload or a batch on an uploaded file falls back only
within its own model group, as on the Python core, since the file lives
with one provider (the host sends same_group_fallbacks_only and the walk
skips targets in other groups); Decisions calls drop deployments whose
provider has no Decisions config, and deployments that refuse the audio
or videos a call carries, unless that would leave none, as the Python
core does, while the retry count still reads the whole group; a file
upload that fails on one order level is created on the next level's
deployments rather than the first level again; and a response that
cannot be converted (a text_completion body without a message) reports
only its own traceback, so a rate limit handled by the attempt before it
no longer turns the error into a 429, and both cores return 500.
2026-11-04 21:47:58 -08:00

500 lines
18 KiB
Python

import asyncio
import base64
import importlib
import os
import shutil
import tempfile
from collections.abc import Callable, Coroutine, Iterator
from dataclasses import dataclass, field
from pathlib import Path
from types import MappingProxyType
from typing import Final, Literal, TypeAlias
import boto3
import httpx
import pytest
from pytest_socket import enable_socket, socket_allow_hosts
HOST_ENVIRONMENT_ALLOWLIST: Final = frozenset(
(
"PATH",
"HOME",
"USER",
"LOGNAME",
"TMPDIR",
"TEMP",
"TMP",
"LANG",
"LC_ALL",
"LC_CTYPE",
"TZ",
"VIRTUAL_ENV",
"LITELLM_LOCAL_MODEL_COST_MAP",
"TIKTOKEN_CACHE_DIR",
)
)
HOST_ENVIRONMENT_ALLOWED_PREFIXES: Final = ("PYTEST_", "PYTHON", "COV_CORE_", "COVERAGE_")
HOST_ONLY_ENVIRONMENT: Final = frozenset(
name
for name in os.environ
if name not in HOST_ENVIRONMENT_ALLOWLIST and not name.startswith(HOST_ENVIRONMENT_ALLOWED_PREFIXES)
)
os.environ["PYTHON_DOTENV_DISABLED"] = "1"
os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
import litellm # noqa: E402 # litellm reads LITELLM_LOCAL_MODEL_COST_MAP at import
import litellm.router as litellm_router_module # noqa: E402 # same import-time dependency
import litellm.utils as litellm_utils_module # noqa: E402 # same import-time dependency
from litellm._logging import ALL_LOGGERS # noqa: E402 # same import-time dependency
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER # noqa: E402 # same import-time dependency
from litellm.anthropic_beta_headers_manager import reload_beta_headers_config # noqa: E402 # same import-time dependency
from litellm.litellm_core_utils.prompt_templates import factory as prompt_factory_module # noqa: E402 # same import-time dependency
from litellm.litellm_core_utils.prompt_templates import ( # noqa: E402 # same import-time dependency
image_handling as image_handling_module,
)
from litellm.llms.gemini.chat import transformation as gemini_chat_transformation_module # noqa: E402 # same import-time dependency
from litellm.llms.custom_httpx.async_client_cleanup import ( # noqa: E402 # same import-time dependency
close_litellm_async_clients,
)
from litellm.proxy.db import tool_registry_writer as tool_registry_writer_module # noqa: E402 # same import-time dependency
LOOPBACK_HOSTS: Final = ["127.0.0.1", "::1", "localhost"]
AMBIENT_AZURE_CREDENTIAL_ENV_VARS: Final = (
"AZURE_AD_TOKEN",
"AZURE_TENANT_ID",
"AZURE_CLIENT_ID",
"AZURE_CLIENT_SECRET",
"AZURE_USERNAME",
"AZURE_PASSWORD",
)
AMBIENT_AWS_ENV_VARS: Final = (
"AWS_PROFILE",
"AWS_DEFAULT_PROFILE",
"AWS_CONTAINER_CREDENTIALS_FULL_URI",
"AWS_CONTAINER_CREDENTIALS_RELATIVE_URI",
"AWS_SESSION_TOKEN",
"AWS_ROLE_ARN",
"AWS_WEB_IDENTITY_TOKEN_FILE",
"AWS_BEARER_TOKEN_BEDROCK",
"AWS_REGION_NAME",
"AWS_DEFAULT_REGION",
)
MODULES_WITH_AWS_AUTH_HANDLERS: Final = (
"litellm.main",
"litellm.files.main",
"litellm.rerank_api.main",
"litellm.realtime_api.main",
)
CALLBACK_LISTS: Final = (
"callbacks",
"success_callback",
"failure_callback",
"input_callback",
"_async_success_callback",
"_async_failure_callback",
"_async_input_callback",
)
RESET_TO_NONE_GLOBALS: Final = (
"model_fallbacks",
"cache",
"max_fallbacks",
"default_fallbacks",
"fallbacks",
"context_window_fallbacks",
"content_policy_fallbacks",
)
RESTORED_GLOBALS: Final = (
"disable_aiohttp_transport",
"force_ipv4",
"drop_params",
"secret_manager_client",
"_key_management_system",
"_key_management_settings",
"api_base",
"num_retries",
"modify_params",
"ssl_verify",
"credential_list",
"model_group_settings",
"default_internal_user_params",
"default_team_params",
"prometheus_emit_stream_label",
"vector_store_registry",
"model_cost",
"cost_margin_config",
"cost_discount_config",
"disable_hf_tokenizer_download",
"disable_copilot_system_to_assistant",
"cohere_models",
"anthropic_models",
"token_counter",
"initialized_langfuse_clients",
)
MODULE_LEVEL_CLIENTS: Final = ("module_level_client", "module_level_aclient")
SESSION_CLIENTS: Final = ("base_llm_aiohttp_handler", "httpx_client", "aclient", "client")
ONE_PIXEL_PNG: Final = base64.b64decode(
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg=="
)
def _allow_loopback_only() -> None:
socket_allow_hosts(LOOPBACK_HOSTS, allow_unix_socket=True)
_allow_loopback_only()
def pytest_collectstart() -> None:
_allow_loopback_only()
@pytest.hookimpl(trylast=True)
def pytest_runtest_setup() -> None:
_allow_loopback_only()
def _run_coroutine_if_needed(result: object) -> None:
if not asyncio.iscoroutine(result):
return
coroutine: Final[Coroutine[object, object, object]] = result
try:
asyncio.run(coroutine)
except RuntimeError:
try:
loop: Final = asyncio.get_running_loop()
except RuntimeError:
coroutine.close()
return
loop.create_task(coroutine)
async def _complete_test_logging() -> None:
await GLOBAL_LOGGING_WORKER.flush()
await GLOBAL_LOGGING_WORKER.stop()
def _flush_completed_test_logging() -> None:
queue: Final = GLOBAL_LOGGING_WORKER._queue
if queue is None or queue._unfinished_tasks == 0:
return
worker_loop: Final = GLOBAL_LOGGING_WORKER._bound_loop
if worker_loop is None or worker_loop.is_closed():
_run_coroutine_if_needed(_complete_test_logging())
return
if worker_loop.is_running():
asyncio.run_coroutine_threadsafe(_complete_test_logging(), worker_loop).result()
return
worker_loop.run_until_complete(_complete_test_logging())
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_call() -> Iterator[None]:
yield
_flush_completed_test_logging()
def _close_handler_if_needed(handler: object) -> None:
close: Final = getattr(handler, "close", None)
if not callable(close):
return
_run_coroutine_if_needed(close())
def _reset_aws_auth_caches() -> None:
modules: Final = tuple(importlib.import_module(name) for name in MODULES_WITH_AWS_AUTH_HANDLERS)
flushes: Final = (
getattr(getattr(getattr(module, attr_name), "iam_cache", None), "flush_cache", None)
for module in modules
for attr_name in dir(module)
)
for flush in filter(callable, flushes):
flush()
boto3.DEFAULT_SESSION = None
def _flush_client_caches() -> None:
litellm.in_memory_llm_clients_cache.flush_cache()
image_handling_module.in_memory_cache.flush_cache()
image_handling_module.pdf_probe_verdicts.flush_cache()
_reset_aws_auth_caches()
_MACOS_LONGEST_SOCKET_PATH: Final = 103
_LONGEST_SOCKET_SUFFIX: Final = len("/sock12345678/d/litellm-pgbouncer-12345678/.s.PGSQL.65535")
def _socket_root() -> str:
default: Final = tempfile.gettempdir()
fits: Final = len(os.fsencode(default)) + _LONGEST_SOCKET_SUFFIX <= _MACOS_LONGEST_SOCKET_PATH
return default if fits else "/tmp"
@pytest.fixture
def socket_dir(tmp_path: Path) -> Iterator[Path]:
"""``tmp_path`` behind a short link, so unix socket paths fit the sun_path limit that it exceeds on macOS."""
link_dir: Final = Path(tempfile.mkdtemp(prefix="sock", dir=_socket_root()))
link: Final = link_dir / "d"
link.symlink_to(tmp_path, target_is_directory=True)
yield link
shutil.rmtree(link_dir, ignore_errors=True)
@pytest.fixture
def socket_path(socket_dir: Path) -> Path:
return socket_dir / "spend.sock"
@pytest.fixture
def socket_tempdir(socket_dir: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
"""Point ``tempfile``'s default directory at ``socket_dir``, for code that picks its own socket directory."""
monkeypatch.setattr(tempfile, "tempdir", str(socket_dir))
return socket_dir
@pytest.fixture
def python_router_core(monkeypatch: pytest.MonkeyPatch) -> None:
"""Keep ``Router`` on ``PythonCore``, for a test that patches a member the Rust core never reads."""
monkeypatch.setenv("LITELLM_RUST", "0")
@pytest.fixture(autouse=True, scope="session")
def bundled_tiktoken_cache() -> None:
importlib.import_module("litellm.litellm_core_utils.default_encoding")
@pytest.fixture(scope="session")
def isolated_aws_config_files(tmp_path_factory: pytest.TempPathFactory) -> tuple[Path, Path]:
aws_dir: Final = tmp_path_factory.mktemp("aws-config")
credentials: Final = aws_dir / "credentials"
config: Final = aws_dir / "config"
credentials.write_text("", encoding="utf-8")
config.write_text("", encoding="utf-8")
return credentials, config
@pytest.fixture(autouse=True)
def isolate_host_environment(isolated_aws_config_files: tuple[Path, Path]) -> Iterator[None]:
credentials, config = isolated_aws_config_files
with pytest.MonkeyPatch.context() as environment:
for name in HOST_ONLY_ENVIRONMENT:
environment.delenv(name, raising=False)
environment.setenv("AWS_SHARED_CREDENTIALS_FILE", str(credentials))
environment.setenv("AWS_CONFIG_FILE", str(config))
environment.setenv("AWS_EC2_METADATA_DISABLED", "true")
for name in AMBIENT_AWS_ENV_VARS:
environment.delenv(name, raising=False)
for name in ("PROXY_BASE_URL", "SERVER_ROOT_PATH", "SERVER_ROOT_PATHS", "LITELLM_LOG"):
environment.delenv(name, raising=False)
environment.setenv("LITELLM_CLI_DISABLE_KEYRING", "1")
yield
@pytest.fixture(autouse=True)
def isolate_litellm_globals() -> Iterator[None]:
original_callbacks: Final = {name: list(getattr(litellm, name) or []) for name in CALLBACK_LISTS}
original_callback_holders: Final = frozenset(litellm_router_module._routers_holding_global_callbacks)
original_reset: Final = {name: getattr(litellm, name) for name in RESET_TO_NONE_GLOBALS}
original_restored: Final = {name: getattr(litellm, name) for name in RESTORED_GLOBALS if hasattr(litellm, name)}
original_clients: Final = {name: litellm.__dict__[name] for name in MODULE_LEVEL_CLIENTS if name in litellm.__dict__}
original_loggers: Final = {
logger: (logger.level, logger.disabled, logger.propagate, list(logger.handlers), list(logger.filters))
for logger in ALL_LOGGERS
}
original_tool_policy_registry: Final = tool_registry_writer_module._tool_policy_registry
_flush_client_caches()
for name in CALLBACK_LISTS:
setattr(litellm, name, [])
litellm_router_module._routers_holding_global_callbacks.clear()
for name in RESET_TO_NONE_GLOBALS:
setattr(litellm, name, None)
for name in MODULE_LEVEL_CLIENTS:
litellm.__dict__.pop(name, None)
tool_registry_writer_module._tool_policy_registry = None
yield
_flush_client_caches()
leaked_clients: Final = tuple(litellm.__dict__.pop(name, None) for name in MODULE_LEVEL_CLIENTS)
for name, client in zip(MODULE_LEVEL_CLIENTS, leaked_clients):
if client is not original_clients.get(name):
_close_handler_if_needed(client)
litellm.__dict__.update(original_clients)
for name, value in (original_callbacks | original_reset | original_restored).items():
setattr(litellm, name, value)
litellm_router_module._routers_holding_global_callbacks.clear()
litellm_router_module._routers_holding_global_callbacks.update(original_callback_holders)
for logger, (level, disabled, propagate, handlers, filters) in original_loggers.items():
logger.setLevel(level)
logger.disabled = disabled
logger.propagate = propagate
logger.handlers = handlers
logger.filters = filters
tool_registry_writer_module._tool_policy_registry = original_tool_policy_registry
@pytest.fixture(autouse=True)
def isolate_router_model_cost_state() -> Iterator[None]:
original_live_routers: Final = frozenset(litellm_router_module._live_routers)
original_runtime_registered_model_cost: Final = {
model_key: dict(model_value)
for model_key, model_value in litellm_utils_module._runtime_registered_model_cost.items()
}
litellm_utils_module._invalidate_model_cost_lowercase_map()
litellm_router_module._live_routers.clear()
litellm_utils_module._runtime_registered_model_cost.clear()
yield
for router in tuple(litellm_router_module._live_routers):
litellm_router_module._live_routers.discard(router)
for router in original_live_routers:
litellm_router_module._live_routers.add(router)
litellm_utils_module._runtime_registered_model_cost.clear()
litellm_utils_module._runtime_registered_model_cost.update(original_runtime_registered_model_cost)
litellm_utils_module._invalidate_model_cost_lowercase_map()
litellm.get_model_info.cache_clear()
def _provider_model_sets() -> tuple[set[str], ...]:
module_globals: Final = tuple(vars(litellm).items())
return tuple(models for name, models in module_globals if name.endswith("_models") and isinstance(models, set))
@pytest.fixture(autouse=True)
def isolate_model_catalog() -> Iterator[None]:
model_cost: Final = litellm.model_cost
model_cost_entries: Final = dict(model_cost)
models_by_provider: Final = litellm.models_by_provider
models_by_provider_entries: Final = dict(models_by_provider)
provider_model_sets: Final = tuple((models, frozenset(models)) for models in _provider_model_sets())
yield
model_cost.clear()
model_cost.update(model_cost_entries)
models_by_provider.clear()
models_by_provider.update(models_by_provider_entries)
for models, entries in provider_model_sets:
models.clear()
models.update(entries)
litellm_utils_module._invalidate_model_cost_lowercase_map()
litellm.get_model_info.cache_clear()
@pytest.fixture
def local_model_cost_map(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
litellm.get_model_info.cache_clear()
yield
litellm.get_model_info.cache_clear()
@pytest.fixture
def declare_mid_conversation_system(monkeypatch: pytest.MonkeyPatch) -> Callable[[str, bool], None]:
def declare(model_cost_key: str, supported: bool) -> None:
entry: Final = {
"litellm_provider": model_cost_key.split("/", 1)[0],
"mode": "chat",
"supports_mid_conversation_system": supported,
}
monkeypatch.setitem(litellm.model_cost, model_cost_key, entry)
litellm.get_model_info.cache_clear()
return declare
NativeStructuredOutputKey: TypeAlias = Literal["legacy", "bedrock"] | None
@pytest.fixture
def set_native_structured_output(
local_model_cost_map: None, monkeypatch: pytest.MonkeyPatch
) -> Callable[[str, NativeStructuredOutputKey], None]:
def set_flags(model_cost_key: str, native_key: NativeStructuredOutputKey) -> None:
entry: Final = litellm.model_cost[model_cost_key]
monkeypatch.setitem(entry, "supports_native_structured_output", native_key == "legacy")
monkeypatch.setitem(entry, "supports_bedrock_native_structured_output", native_key == "bedrock")
litellm.get_model_info.cache_clear()
return set_flags
@pytest.fixture
def local_beta_headers_config(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
monkeypatch.setenv("LITELLM_LOCAL_ANTHROPIC_BETA_HEADERS", "True")
reload_beta_headers_config()
yield
monkeypatch.delenv("LITELLM_LOCAL_ANTHROPIC_BETA_HEADERS", raising=False)
reload_beta_headers_config()
@dataclass(slots=True)
class AsyncOnlyImageFetch:
fetched: list[str] = field(default_factory=list) # mutable-ok: tests assert on the URLs fetched, in order
base64_png: str = base64.b64encode(ONE_PIXEL_PNG).decode()
data_url: str = "data:image/png;base64," + base64.b64encode(ONE_PIXEL_PNG).decode()
@pytest.fixture
def async_only_image_fetch(monkeypatch: pytest.MonkeyPatch) -> AsyncOnlyImageFetch:
fetch: Final = AsyncOnlyImageFetch()
def forbid_sync_fetch(client: object, url: str, **kwargs: object) -> httpx.Response:
raise litellm.ImageFetchError(f"sync image fetch ran on the event loop: {url}")
async def serve_png(client: object, url: str, **kwargs: object) -> httpx.Response:
fetch.fetched.append(url)
return httpx.Response(
200, content=ONE_PIXEL_PNG, headers={"content-type": "image/png"}, request=httpx.Request("GET", url)
)
def forbid_sync_convert(url: str, *args: object, **kwargs: object) -> str:
if url.startswith(("http://", "https://")):
raise litellm.ImageFetchError(f"sync convert_url_to_base64 ran on the request path: {url}")
return url
monkeypatch.setattr(image_handling_module, "safe_get", forbid_sync_fetch)
monkeypatch.setattr(image_handling_module, "async_safe_get", serve_png)
for module in (image_handling_module, prompt_factory_module, gemini_chat_transformation_module):
monkeypatch.setattr(module, "convert_url_to_base64", forbid_sync_convert)
return fetch
@pytest.fixture
def no_ambient_azure_credentials(monkeypatch: pytest.MonkeyPatch) -> None:
for name in AMBIENT_AZURE_CREDENTIAL_ENV_VARS:
monkeypatch.delenv(name, raising=False)
FAKE_PROVIDER_CREDENTIALS: Final = MappingProxyType(
{
"OPENAI_API_KEY": "sk-unit-test",
"ANTHROPIC_API_KEY": "sk-ant-unit-test",
"GEMINI_API_KEY": "unit-test",
"AZURE_API_KEY": "unit-test",
"AZURE_API_BASE": "https://unit-test.openai.azure.com",
"AZURE_API_VERSION": "2024-02-01",
"AWS_ACCESS_KEY_ID": "unit-test",
"AWS_SECRET_ACCESS_KEY": "unit-test",
"AWS_REGION_NAME": "us-east-1",
"COHERE_API_KEY": "unit-test",
"DD_API_KEY": "unit-test",
"DD_SITE": "us5.datadoghq.com",
}
)
@pytest.fixture
def fake_provider_credentials(monkeypatch: pytest.MonkeyPatch) -> None:
for name, value in FAKE_PROVIDER_CREDENTIALS.items():
monkeypatch.setenv(name, value)
@pytest.fixture
async def drained_logging_worker() -> None:
await asyncio.wait_for(GLOBAL_LOGGING_WORKER.clear_queue(), timeout=10)
def pytest_sessionfinish() -> None:
for name in MODULE_LEVEL_CLIENTS:
_close_handler_if_needed(litellm.__dict__.pop(name, None))
for name in SESSION_CLIENTS:
_close_handler_if_needed(getattr(litellm, name, None))
_run_coroutine_if_needed(close_litellm_async_clients())
enable_socket()