ai-workspace-services/litellm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test(proxy/utils): pin bottom-of-file helper behavior (#29509)

Browse Source 
* test(proxy/utils): pin bottom-of-file helper behavior

Pin current behavior of the bottom-of-file pure-function helpers in
litellm/proxy/utils.py (projection, team config, time helpers,
guardrail merge, error helpers, URL/path helpers, premium gate, model
access, and misc DB/API-key helpers).

Adds tests/test_litellm/proxy/utils/helpers/ with one happy + one error
test per pinned symbol; folds the prior single-test
tests/test_litellm/proxy/test_utils.py into test_url_helpers.py and
deletes the old file. _pin_check.py and _coverage_check.py serve as
local stopping gates.

Adds tests/test_litellm/proxy/utils to the existing test-path block in
.github/workflows/test-unit-proxy-endpoints.yml.

Plan:     https://www.notion.so/37343b8acdab81f68f39f66915f62bcf
Pin list: https://www.notion.so/37343b8acdab8150acdbf40e5756869f

* test(proxy/utils): apply greptile fixes to behavior-pinning gates

Address findings from the sibling PR1/PR2 greptile reviews that also
apply to this PR:

- Commit pin_list.txt alongside the gate script (was previously a
  gitignored .pin_list.txt fetched from Notion). The gate is now
  reproducible without out-of-band setup.
- Resolve the coverage region by locating the first pinned symbol's
  def line in litellm/proxy/utils.py at runtime, instead of hardcoded
  line numbers that drift when lines above shift.
- Word-boundary the pin reference check so pins like update_spend do
  not falsely match update_spend_logs_job.
- Drop the dead _harness_smoke_test.py exclusion; the test_*.py glob
  already filters underscore-prefixed files.

* test(proxy/utils): drop local-only stopping-signal scripts

Remove _pin_check.py, _coverage_check.py, and pin_list.txt. These were
dev-time tooling for knowing when test authoring was done; they are
not wired into CI and the test files themselves are the merge artifact.

---------

Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
yuneng-jiang 2026-06-02 17:45:19 -07:00 committed by GitHub
parent f047b1571e
commit 1aed5e1bbd
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 1742 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.github/workflows/test-unit-proxy-endpoints.yml vendored
View File

@ -42,6 +42,7 @@ jobs:
tests/test_litellm/proxy/rag_endpoints
tests/test_litellm/proxy/realtime_endpoints
tests/test_litellm/proxy/ui_crud_endpoints
tests/test_litellm/proxy/utils
workers: 2
reruns: 2
artifact-name: proxy-endpoints

22
tests/test_litellm/proxy/test_utils.py
View File

@ -1,22 +0,0 @@
import pytest
from litellm.proxy.utils import _get_openapi_url
@pytest.mark.parametrize(
"env_vars, expected_url",
[
({}, "/openapi.json"), # default case
({"NO_OPENAPI": "True"}, None), # OpenAPI disabled
],
)
def test_get_openapi_url(monkeypatch, env_vars, expected_url):
# Clear relevant environment variables
monkeypatch.delenv("NO_OPENAPI", raising=False)
# Set test environment variables
for key, value in env_vars.items():
monkeypatch.setenv(key, value)
result = _get_openapi_url()
assert result == expected_url

0
tests/test_litellm/proxy/utils/__init__.py Normal file
View File

0
tests/test_litellm/proxy/utils/helpers/__init__.py Normal file
View File

173
tests/test_litellm/proxy/utils/helpers/test_error_helpers.py Normal file
View File

@ -0,0 +1,173 @@
import json
import pytest
from fastapi import HTTPException
from litellm.proxy._types import ProxyErrorTypes, ProxyException
from litellm.proxy.utils import get_error_message_str, handle_exception_on_proxy
def normalize(value):
return value
def test_get_error_message_str_happy_path_http_exception_with_string_detail():
exc = HTTPException(status_code=400, detail="something went wrong")
summary = {
"result": get_error_message_str(exc),
"status_code": exc.status_code,
"is_str": True,
}
assert summary == {
"result": "something went wrong",
"status_code": 400,
"is_str": True,
}
def test_get_error_message_str_happy_path_http_exception_with_dict_detail():
detail = {"error": "bad input", "code": "invalid_request"}
exc = HTTPException(status_code=422, detail=detail)
summary = {
"result": get_error_message_str(exc),
"result_parsed": json.loads(get_error_message_str(exc)),
"status_code": exc.status_code,
}
assert summary == {
"result": json.dumps(detail),
"result_parsed": detail,
"status_code": 422,
}
def test_get_error_message_str_happy_path_generic_exception():
exc = ValueError("boom")
summary = {
"result": get_error_message_str(exc),
"type": type(exc).__name__,
"args": list(exc.args),
}
assert summary == {
"result": "boom",
"type": "ValueError",
"args": ["boom"],
}
def test_get_error_message_str_with_runtime_error():
exc = RuntimeError("runtime explosion")
summary = {
"result": get_error_message_str(exc),
"type": type(exc).__name__,
"matches_str": str(exc) == get_error_message_str(exc),
}
assert summary == {
"result": "runtime explosion",
"type": "RuntimeError",
"matches_str": True,
}
def test_get_error_message_str_error_path_none_input_returns_string_none():
summary = {
"result": get_error_message_str(None),
"is_str": isinstance(get_error_message_str(None), str),
"input": None,
}
assert summary == {
"result": "None",
"is_str": True,
"input": None,
}
def test_handle_exception_on_proxy_happy_path_http_exception():
exc = HTTPException(status_code=403, detail="forbidden")
result = handle_exception_on_proxy(exc)
snapshot = {
"is_proxy_exception": isinstance(result, ProxyException),
"message": result.message,
"type": result.type,
"code": result.code,
}
assert snapshot == {
"is_proxy_exception": True,
"message": "forbidden",
"type": ProxyErrorTypes.internal_server_error.value,
"code": "403",
}
def test_handle_exception_on_proxy_happy_path_already_proxy_exception():
original = ProxyException(
message="already wrapped",
type=ProxyErrorTypes.budget_exceeded.value,
param="key",
code=402,
)
result = handle_exception_on_proxy(original)
snapshot = {
"is_same_object": result is original,
"message": result.message,
"type": result.type,
"code": result.code,
}
assert snapshot == {
"is_same_object": True,
"message": "already wrapped",
"type": ProxyErrorTypes.budget_exceeded.value,
"code": "402",
}
def test_handle_exception_on_proxy_happy_path_generic_exception_defaults_to_500():
exc = ValueError("kaboom")
result = handle_exception_on_proxy(exc)
snapshot = {
"is_proxy_exception": isinstance(result, ProxyException),
"message": result.message,
"type": result.type,
"code": result.code,
"param": result.param,
}
assert snapshot == {
"is_proxy_exception": True,
"message": "kaboom",
"type": ProxyErrorTypes.internal_server_error.value,
"code": "500",
"param": "None",
}
def test_handle_exception_on_proxy_uses_attached_status_code_when_present():
class _CustomErr(Exception):
status_code = 418
exc = _CustomErr("teapot")
result = handle_exception_on_proxy(exc)
snapshot = {
"code": result.code,
"message": result.message,
"type": result.type,
}
assert snapshot == {
"code": "418",
"message": "teapot",
"type": ProxyErrorTypes.internal_server_error.value,
}
def test_handle_exception_on_proxy_error_path_none_input_wraps_as_500():
result = handle_exception_on_proxy(None)
snapshot = {
"is_proxy_exception": isinstance(result, ProxyException),
"message": result.message,
"code": result.code,
"type": result.type,
}
assert snapshot == {
"is_proxy_exception": True,
"message": "None",
"code": "500",
"type": ProxyErrorTypes.internal_server_error.value,
}

201
tests/test_litellm/proxy/utils/helpers/test_guardrail_merge.py Normal file
View File

@ -0,0 +1,201 @@
from types import SimpleNamespace
from unittest.mock import MagicMock
import pytest
from litellm.proxy.utils import (
_check_and_merge_model_level_guardrails,
_merge_guardrails_with_existing,
)
def normalize(value):
return value
def _router_with_deployment(guardrails):
deployment = SimpleNamespace(litellm_params={"guardrails": guardrails})
router = MagicMock()
router.get_deployment.return_value = deployment
return router
def _router_without_deployment():
router = MagicMock()
router.get_deployment.return_value = None
return router
def test_check_and_merge_model_level_guardrails_happy_path_merges_lists():
router = _router_with_deployment(["pii-redact", "toxic-filter"])
data = {
"model": "gpt-4o",
"metadata": {
"model_info": {"id": "deployment-123"},
"guardrails": ["user-policy"],
},
}
result = _check_and_merge_model_level_guardrails(data, router)
snapshot = {
"model": result["model"],
"model_info_id": result["metadata"]["model_info"]["id"],
"guardrails_sorted": sorted(result["metadata"]["guardrails"]),
}
assert snapshot == {
"model": "gpt-4o",
"model_info_id": "deployment-123",
"guardrails_sorted": ["pii-redact", "toxic-filter", "user-policy"],
}
def test_check_and_merge_model_level_guardrails_returns_data_when_router_none():
data = {"metadata": {"model_info": {"id": "x"}}, "model": "m", "other": 1}
result = _check_and_merge_model_level_guardrails(data, None)
assert result is data
assert normalize(result) == {
"metadata": {"model_info": {"id": "x"}},
"model": "m",
"other": 1,
}
def test_check_and_merge_model_level_guardrails_returns_data_when_model_id_missing():
router = _router_with_deployment(["pii"])
data = {"metadata": {"model_info": {}}, "model": "m", "extra": "v"}
result = _check_and_merge_model_level_guardrails(data, router)
snapshot = {
"is_same_object": result is data,
"metadata": result["metadata"],
"model": result["model"],
"extra": result["extra"],
}
assert snapshot == {
"is_same_object": True,
"metadata": {"model_info": {}},
"model": "m",
"extra": "v",
}
router.get_deployment.assert_not_called()
def test_check_and_merge_model_level_guardrails_returns_data_when_deployment_none():
router = _router_without_deployment()
data = {"metadata": {"model_info": {"id": "x"}}, "model": "m"}
result = _check_and_merge_model_level_guardrails(data, router)
assert result is data
def test_check_and_merge_model_level_guardrails_returns_data_when_guardrails_none():
router = _router_with_deployment(None)
data = {"metadata": {"model_info": {"id": "x"}}, "model": "m"}
result = _check_and_merge_model_level_guardrails(data, router)
assert result is data
def test_check_and_merge_model_level_guardrails_handles_missing_metadata():
router = _router_with_deployment(["pii"])
data = {"model": "m"}
result = _check_and_merge_model_level_guardrails(data, router)
snapshot = {
"is_same_object": result is data,
"model": result["model"],
"metadata_present": "metadata" in result,
}
assert snapshot == {
"is_same_object": True,
"model": "m",
"metadata_present": False,
}
def test_check_and_merge_model_level_guardrails_raises_when_metadata_is_not_dict():
router = _router_with_deployment(["pii"])
data = {"metadata": "not-a-dict", "model": "m"}
with pytest.raises(AttributeError):
_check_and_merge_model_level_guardrails(data, router)
def test_merge_guardrails_with_existing_happy_path_combines_lists():
data = {
"metadata": {"guardrails": ["a", "b"], "user": "u"},
"model": "m",
}
result = _merge_guardrails_with_existing(data, ["c", "a"])
snapshot = {
"guardrails_sorted": sorted(result["metadata"]["guardrails"]),
"user": result["metadata"]["user"],
"model": result["model"],
"is_copy": result is not data,
}
assert snapshot == {
"guardrails_sorted": ["a", "b", "c"],
"user": "u",
"model": "m",
"is_copy": True,
}
def test_merge_guardrails_with_existing_wraps_scalar_existing_guardrail():
data = {"metadata": {"guardrails": "single-policy"}}
result = _merge_guardrails_with_existing(data, ["model-policy"])
snapshot = {
"guardrails_sorted": sorted(result["metadata"]["guardrails"]),
"is_list": isinstance(result["metadata"]["guardrails"], list),
"count": len(result["metadata"]["guardrails"]),
}
assert snapshot == {
"guardrails_sorted": ["model-policy", "single-policy"],
"is_list": True,
"count": 2,
}
def test_merge_guardrails_with_existing_wraps_scalar_model_guardrail():
data = {"metadata": {}}
result = _merge_guardrails_with_existing(data, "model-policy")
snapshot = {
"guardrails": result["metadata"]["guardrails"],
"is_list": isinstance(result["metadata"]["guardrails"], list),
"count": len(result["metadata"]["guardrails"]),
}
assert snapshot == {
"guardrails": ["model-policy"],
"is_list": True,
"count": 1,
}
def test_merge_guardrails_with_existing_empty_existing_empty_model_yields_empty():
data = {"metadata": {"guardrails": None}}
result = _merge_guardrails_with_existing(data, None)
snapshot = {
"guardrails": result["metadata"]["guardrails"],
"is_list": isinstance(result["metadata"]["guardrails"], list),
"count": len(result["metadata"]["guardrails"]),
}
assert snapshot == {
"guardrails": [],
"is_list": True,
"count": 0,
}
def test_merge_guardrails_with_existing_creates_metadata_when_missing():
data = {"model": "m"}
result = _merge_guardrails_with_existing(data, ["g1"])
snapshot = {
"guardrails": result["metadata"]["guardrails"],
"model_preserved": result["model"],
"original_data_unchanged": "metadata" not in data,
}
assert snapshot == {
"guardrails": ["g1"],
"model_preserved": "m",
"original_data_unchanged": True,
}
def test_merge_guardrails_with_existing_raises_on_unhashable_guardrail():
data = {"metadata": {"guardrails": [{"unhashable": True}]}}
with pytest.raises(TypeError):
_merge_guardrails_with_existing(data, ["g1"])

201
tests/test_litellm/proxy/utils/helpers/test_misc_helpers.py Normal file
View File

@ -0,0 +1,201 @@
import pytest
from fastapi import HTTPException
from litellm.proxy.utils import (
construct_database_url_from_env_vars,
get_prisma_client_or_throw,
is_valid_api_key,
)
def normalize(value):
return value
def test_get_prisma_client_or_throw_happy_path_returns_client(monkeypatch):
sentinel = object()
import litellm.proxy.proxy_server as ps
monkeypatch.setattr(ps, "prisma_client", sentinel, raising=False)
result = get_prisma_client_or_throw("some message")
summary = {
"is_sentinel": result is sentinel,
"message_arg": "some message",
"raised": False,
}
assert summary == {
"is_sentinel": True,
"message_arg": "some message",
"raised": False,
}
def test_get_prisma_client_or_throw_raises_when_client_none(monkeypatch):
import litellm.proxy.proxy_server as ps
monkeypatch.setattr(ps, "prisma_client", None, raising=False)
with pytest.raises(HTTPException) as exc_info:
get_prisma_client_or_throw("db not connected")
snapshot = {
"status_code": exc_info.value.status_code,
"is_dict_detail": isinstance(exc_info.value.detail, dict),
"error_message": exc_info.value.detail["error"],
}
assert snapshot == {
"status_code": 500,
"is_dict_detail": True,
"error_message": "db not connected",
}
def test_is_valid_api_key_happy_path_sk_prefix():
summary = {
"result": is_valid_api_key("sk-abc123_XYZ-456"),
"key": "sk-abc123_XYZ-456",
"len": len("sk-abc123_XYZ-456"),
}
assert summary == {
"result": True,
"key": "sk-abc123_XYZ-456",
"len": 17,
}
def test_is_valid_api_key_happy_path_hashed_64_hex():
key = "a" * 64
summary = {
"result": is_valid_api_key(key),
"key_len": len(key),
"is_hex": True,
}
assert summary == {
"result": True,
"key_len": 64,
"is_hex": True,
}
def test_is_valid_api_key_happy_path_mixed_case_hex():
key = "AbCdEf0123456789" * 4
summary = {
"result": is_valid_api_key(key),
"key_len": len(key),
"first": key[0],
}
assert summary == {
"result": True,
"key_len": 64,
"first": "A",
}
def test_is_valid_api_key_error_path_too_long():
assert is_valid_api_key("sk-" + "a" * 200) is False
def test_is_valid_api_key_error_path_non_string():
assert is_valid_api_key(12345) is False # type: ignore[arg-type]
def test_is_valid_api_key_error_path_invalid_format():
assert is_valid_api_key("not-a-valid-key-format!!!!") is False
def test_is_valid_api_key_error_path_too_short():
assert is_valid_api_key("sk") is False
def test_construct_database_url_from_env_vars_happy_path_full(monkeypatch):
monkeypatch.setenv("DATABASE_HOST", "db.example.com")
monkeypatch.setenv("DATABASE_USERNAME", "user")
monkeypatch.setenv("DATABASE_PASSWORD", "pass")
monkeypatch.setenv("DATABASE_NAME", "litellm")
monkeypatch.delenv("DATABASE_SCHEMA", raising=False)
result = construct_database_url_from_env_vars()
summary = {
"result": result,
"host": "db.example.com",
"scheme": result.split("://", 1)[0] if result else None,
"has_password": "pass" in (result or ""),
}
assert summary == {
"result": "postgresql://user:pass@db.example.com/litellm",
"host": "db.example.com",
"scheme": "postgresql",
"has_password": True,
}
def test_construct_database_url_from_env_vars_happy_path_no_password(monkeypatch):
monkeypatch.setenv("DATABASE_HOST", "db.example.com")
monkeypatch.setenv("DATABASE_USERNAME", "user")
monkeypatch.delenv("DATABASE_PASSWORD", raising=False)
monkeypatch.setenv("DATABASE_NAME", "litellm")
monkeypatch.delenv("DATABASE_SCHEMA", raising=False)
result = construct_database_url_from_env_vars()
summary = {
"result": result,
"no_colon_password": ":pass@" not in (result or ""),
"host": "db.example.com",
"user": "user",
}
assert summary == {
"result": "postgresql://user@db.example.com/litellm",
"no_colon_password": True,
"host": "db.example.com",
"user": "user",
}
def test_construct_database_url_from_env_vars_special_chars_encoded(monkeypatch):
monkeypatch.setenv("DATABASE_HOST", "db.example.com")
monkeypatch.setenv("DATABASE_USERNAME", "us er@x")
monkeypatch.setenv("DATABASE_PASSWORD", "p@ss/word")
monkeypatch.setenv("DATABASE_NAME", "lite/llm")
monkeypatch.delenv("DATABASE_SCHEMA", raising=False)
result = construct_database_url_from_env_vars()
summary = {
"result": result,
"username_encoded": "us+er%40x" in result,
"password_encoded": "p%40ss%2Fword" in result,
"name_encoded": "lite%2Fllm" in result,
}
assert summary == {
"result": "postgresql://us+er%40x:p%40ss%2Fword@db.example.com/lite%2Fllm",
"username_encoded": True,
"password_encoded": True,
"name_encoded": True,
}
def test_construct_database_url_from_env_vars_with_schema(monkeypatch):
monkeypatch.setenv("DATABASE_HOST", "db.example.com")
monkeypatch.setenv("DATABASE_USERNAME", "user")
monkeypatch.setenv("DATABASE_PASSWORD", "pass")
monkeypatch.setenv("DATABASE_NAME", "litellm")
monkeypatch.setenv("DATABASE_SCHEMA", "public")
result = construct_database_url_from_env_vars()
summary = {
"result": result,
"schema_appended": result.endswith("?schema=public"),
"host": "db.example.com",
}
assert summary == {
"result": "postgresql://user:pass@db.example.com/litellm?schema=public",
"schema_appended": True,
"host": "db.example.com",
}
def test_construct_database_url_from_env_vars_error_path_missing_host(monkeypatch):
monkeypatch.delenv("DATABASE_HOST", raising=False)
monkeypatch.setenv("DATABASE_USERNAME", "user")
monkeypatch.setenv("DATABASE_NAME", "litellm")
assert construct_database_url_from_env_vars() is None
def test_construct_database_url_from_env_vars_error_path_missing_username(monkeypatch):
monkeypatch.setenv("DATABASE_HOST", "db.example.com")
monkeypatch.delenv("DATABASE_USERNAME", raising=False)
monkeypatch.setenv("DATABASE_NAME", "litellm")
assert construct_database_url_from_env_vars() is None

406
tests/test_litellm/proxy/utils/helpers/test_model_access.py Normal file
View File

@ -0,0 +1,406 @@
from unittest.mock import MagicMock
import pytest
from fastapi import HTTPException
import litellm
from litellm import ModelResponse
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.utils import (
create_model_info_response,
get_available_models_for_user,
is_known_model,
is_known_vector_store_index,
model_dump_with_preserved_fields,
validate_model_access,
)
def normalize(value):
return value
def _router_with_models(model_names):
router = MagicMock()
router.get_model_names.return_value = model_names
router.get_model_access_groups.return_value = {}
return router
def test_is_known_model_happy_path_returns_true_when_in_router():
router = _router_with_models(["gpt-4o", "claude-haiku"])
summary = {
"result": is_known_model("gpt-4o", router),
"model": "gpt-4o",
"router_models": ["gpt-4o", "claude-haiku"],
}
assert summary == {
"result": True,
"model": "gpt-4o",
"router_models": ["gpt-4o", "claude-haiku"],
}
def test_is_known_model_returns_false_when_not_in_router():
router = _router_with_models(["gpt-4o"])
summary = {
"result": is_known_model("claude-haiku", router),
"model": "claude-haiku",
"router_models": ["gpt-4o"],
}
assert summary == {
"result": False,
"model": "claude-haiku",
"router_models": ["gpt-4o"],
}
def test_is_known_model_error_path_none_model():
router = _router_with_models(["gpt-4o"])
assert is_known_model(None, router) is False
def test_is_known_model_error_path_none_router():
assert is_known_model("gpt-4o", None) is False
def test_is_known_vector_store_index_happy_path(monkeypatch):
registry = MagicMock()
registry.get_vector_store_indexes.return_value = ["index-a", "index-b"]
monkeypatch.setattr(litellm, "vector_store_index_registry", registry)
summary = {
"result": is_known_vector_store_index("index-a"),
"indexes": ["index-a", "index-b"],
"input": "index-a",
}
assert summary == {
"result": True,
"indexes": ["index-a", "index-b"],
"input": "index-a",
}
def test_is_known_vector_store_index_returns_false_when_missing(monkeypatch):
registry = MagicMock()
registry.get_vector_store_indexes.return_value = ["index-a"]
monkeypatch.setattr(litellm, "vector_store_index_registry", registry)
summary = {
"result": is_known_vector_store_index("missing"),
"indexes": ["index-a"],
"input": "missing",
}
assert summary == {
"result": False,
"indexes": ["index-a"],
"input": "missing",
}
def test_is_known_vector_store_index_error_path_no_registry(monkeypatch):
monkeypatch.setattr(litellm, "vector_store_index_registry", None)
assert is_known_vector_store_index("anything") is False
def test_create_model_info_response_happy_path_no_metadata():
result = create_model_info_response(model_id="gpt-4o", provider="openai")
assert result == {
"id": "gpt-4o",
"object": "model",
"created": result["created"],
"owned_by": "openai",
}
snapshot = {
"id": result["id"],
"object": result["object"],
"owned_by": result["owned_by"],
"created_is_int": isinstance(result["created"], int),
"metadata_absent": "metadata" not in result,
}
assert snapshot == {
"id": "gpt-4o",
"object": "model",
"owned_by": "openai",
"created_is_int": True,
"metadata_absent": True,
}
def test_create_model_info_response_with_metadata_default_general(monkeypatch):
monkeypatch.setattr(
"litellm.proxy.auth.model_checks.get_all_fallbacks",
lambda **_kwargs: [{"model": "fallback-1"}],
)
result = create_model_info_response(
model_id="gpt-4o",
provider="openai",
include_metadata=True,
)
snapshot = {
"id": result["id"],
"owned_by": result["owned_by"],
"object": result["object"],
"fallbacks": result["metadata"]["fallbacks"],
}
assert snapshot == {
"id": "gpt-4o",
"owned_by": "openai",
"object": "model",
"fallbacks": [{"model": "fallback-1"}],
}
def test_create_model_info_response_with_explicit_fallback_type(monkeypatch):
captured = {}
def _capture(model, llm_router, fallback_type):
captured["fallback_type"] = fallback_type
return ["x"]
monkeypatch.setattr("litellm.proxy.auth.model_checks.get_all_fallbacks", _capture)
result = create_model_info_response(
model_id="gpt-4o",
provider="openai",
include_metadata=True,
fallback_type="context_window",
)
snapshot = {
"id": result["id"],
"fallbacks": result["metadata"]["fallbacks"],
"captured_fallback_type": captured["fallback_type"],
"owned_by": result["owned_by"],
}
assert snapshot == {
"id": "gpt-4o",
"fallbacks": ["x"],
"captured_fallback_type": "context_window",
"owned_by": "openai",
}
def test_create_model_info_response_invalid_fallback_type_raises():
with pytest.raises(HTTPException) as exc_info:
create_model_info_response(
model_id="gpt-4o",
provider="openai",
include_metadata=True,
fallback_type="bogus",
)
assert exc_info.value.status_code == 400
assert "Invalid fallback_type" in str(exc_info.value.detail)
def test_validate_model_access_happy_path_single_model_in_list():
summary = {
"result": validate_model_access("gpt-4o", ["gpt-4o", "claude-haiku"]),
"model": "gpt-4o",
"available": ["gpt-4o", "claude-haiku"],
}
assert summary == {
"result": None,
"model": "gpt-4o",
"available": ["gpt-4o", "claude-haiku"],
}
def test_validate_model_access_happy_path_batch_all_accessible():
summary = {
"result": validate_model_access(
"gpt-4o,claude-haiku", ["gpt-4o", "claude-haiku", "gemini"]
),
"input": "gpt-4o,claude-haiku",
"available": ["gpt-4o", "claude-haiku", "gemini"],
}
assert summary == {
"result": None,
"input": "gpt-4o,claude-haiku",
"available": ["gpt-4o", "claude-haiku", "gemini"],
}
def test_validate_model_access_single_model_not_accessible_raises():
with pytest.raises(HTTPException) as exc_info:
validate_model_access("missing-model", ["gpt-4o"])
assert exc_info.value.status_code == 404
assert "missing-model" in str(exc_info.value.detail)
def test_validate_model_access_batch_partial_inaccessible_raises():
with pytest.raises(HTTPException) as exc_info:
validate_model_access("gpt-4o,unknown-x", ["gpt-4o"])
assert exc_info.value.status_code == 404
assert "unknown-x" in str(exc_info.value.detail)
assert "gpt-4o" not in str(exc_info.value.detail).split("not accessible:")[1]
def _make_model_response():
return ModelResponse(
id="resp-123",
choices=[
{
"message": {
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {"name": "do_thing", "arguments": "{}"},
}
],
},
"index": 0,
"finish_reason": "tool_calls",
}
],
model="gpt-4o",
)
def test_model_dump_with_preserved_fields_restores_none_content():
resp = _make_model_response()
result = model_dump_with_preserved_fields(resp)
message = result["choices"][0]["message"]
snapshot = {
"content_is_none": message["content"] is None,
"role": message["role"],
"has_tool_calls": "tool_calls" in message,
"model": result["model"],
}
assert snapshot == {
"content_is_none": True,
"role": "assistant",
"has_tool_calls": True,
"model": "gpt-4o",
}
def test_model_dump_with_preserved_fields_no_choices_returns_plain_dump():
class _Bare:
def model_dump(self, **_kwargs):
return {"id": "x", "object": "y", "extra": "z"}
bare = _Bare()
result = model_dump_with_preserved_fields(bare)
assert result == {"id": "x", "object": "y", "extra": "z"}
def test_model_dump_with_preserved_fields_error_path_invalid_obj_raises():
with pytest.raises(AttributeError):
model_dump_with_preserved_fields(None)
@pytest.mark.asyncio
async def test_get_available_models_for_user_happy_path_returns_complete_list(
monkeypatch,
):
monkeypatch.setattr(
"litellm.proxy.auth.model_checks.get_key_models",
lambda **_k: ["gpt-4o"],
)
monkeypatch.setattr(
"litellm.proxy.auth.model_checks.get_team_models",
lambda **_k: ["claude-haiku"],
)
monkeypatch.setattr(
"litellm.proxy.auth.model_checks.get_complete_model_list",
lambda **_k: ["gpt-4o", "claude-haiku", "gemini"],
)
router = _router_with_models(["gpt-4o", "claude-haiku", "gemini"])
user_api_key_dict = UserAPIKeyAuth(
api_key="sk-test-key",
user_id="user-1",
team_id=None,
team_models=[],
)
result = await get_available_models_for_user(
user_api_key_dict=user_api_key_dict,
llm_router=router,
general_settings={},
user_model=None,
)
summary = {
"result_sorted": sorted(result),
"count": len(result),
"user_id": user_api_key_dict.user_id,
"router_set": True,
}
assert summary == {
"result_sorted": ["claude-haiku", "gemini", "gpt-4o"],
"count": 3,
"user_id": "user-1",
"router_set": True,
}
@pytest.mark.asyncio
async def test_get_available_models_for_user_with_none_router(monkeypatch):
monkeypatch.setattr(
"litellm.proxy.auth.model_checks.get_key_models",
lambda **_k: [],
)
monkeypatch.setattr(
"litellm.proxy.auth.model_checks.get_team_models",
lambda **_k: [],
)
monkeypatch.setattr(
"litellm.proxy.auth.model_checks.get_complete_model_list",
lambda **_k: ["user-model"],
)
user_api_key_dict = UserAPIKeyAuth(
api_key="sk-test-key",
user_id="user-1",
team_id=None,
team_models=[],
)
result = await get_available_models_for_user(
user_api_key_dict=user_api_key_dict,
llm_router=None,
general_settings={},
user_model="user-model",
)
summary = {
"result": result,
"router_is_none": True,
"user_model": "user-model",
"count": len(result),
}
assert summary == {
"result": ["user-model"],
"router_is_none": True,
"user_model": "user-model",
"count": 1,
}
@pytest.mark.asyncio
async def test_get_available_models_for_user_error_path_complete_list_raises(
monkeypatch,
):
monkeypatch.setattr(
"litellm.proxy.auth.model_checks.get_key_models",
lambda **_k: [],
)
monkeypatch.setattr(
"litellm.proxy.auth.model_checks.get_team_models",
lambda **_k: [],
)
def _boom(**_kwargs):
raise RuntimeError("downstream failure")
monkeypatch.setattr(
"litellm.proxy.auth.model_checks.get_complete_model_list", _boom
)
user_api_key_dict = UserAPIKeyAuth(
api_key="sk-test-key",
user_id="user-1",
team_id=None,
team_models=[],
)
with pytest.raises(RuntimeError):
await get_available_models_for_user(
user_api_key_dict=user_api_key_dict,
llm_router=None,
general_settings={},
user_model=None,
)

232
tests/test_litellm/proxy/utils/helpers/test_month_end_projection.py Normal file
View File

@ -0,0 +1,232 @@
from datetime import date, timedelta
import pytest
from litellm.proxy.utils import (
_get_month_end_date,
_get_projected_spend_over_limit,
_is_projected_spend_over_limit,
)
def normalize(value):
return value
def _freeze_today(monkeypatch, frozen):
class _FrozenDate(date):
@classmethod
def today(cls):
return frozen
monkeypatch.setattr("litellm.proxy.utils.date", _FrozenDate)
@pytest.mark.parametrize(
"today, expected",
[
(date(2024, 1, 15), date(2024, 1, 31)),
(date(2024, 2, 1), date(2024, 2, 29)),
(date(2023, 2, 1), date(2023, 2, 28)),
(date(2024, 4, 10), date(2024, 4, 30)),
(date(2024, 12, 1), date(2024, 12, 31)),
],
)
def test_get_month_end_date_happy_path(today, expected):
result = _get_month_end_date(today)
assert normalize(
{
"year": result.year,
"month": result.month,
"day": result.day,
"expected": expected.isoformat(),
"input": today.isoformat(),
}
) == {
"year": expected.year,
"month": expected.month,
"day": expected.day,
"expected": expected.isoformat(),
"input": today.isoformat(),
}
def test_get_month_end_date_raises_on_non_date_input():
with pytest.raises(AttributeError):
_get_month_end_date("2024-01-15")
def test_is_projected_spend_over_limit_happy_path_under_budget(monkeypatch):
_freeze_today(monkeypatch, date(2024, 1, 11))
summary = {
"result": _is_projected_spend_over_limit(
current_spend=10.0, soft_budget_limit=1_000_000.0
),
"current_spend": 10.0,
"soft_budget_limit": 1_000_000.0,
}
assert summary == {
"result": False,
"current_spend": 10.0,
"soft_budget_limit": 1_000_000.0,
}
def test_is_projected_spend_over_limit_happy_path_over_budget(monkeypatch):
_freeze_today(monkeypatch, date(2024, 1, 11))
summary = {
"result": _is_projected_spend_over_limit(
current_spend=100.0, soft_budget_limit=50.0
),
"current_spend": 100.0,
"soft_budget_limit": 50.0,
}
assert summary == {
"result": True,
"current_spend": 100.0,
"soft_budget_limit": 50.0,
}
def test_is_projected_spend_over_limit_first_of_month_no_division_by_zero(monkeypatch):
_freeze_today(monkeypatch, date(2024, 1, 1))
summary = {
"result": _is_projected_spend_over_limit(
current_spend=5.0, soft_budget_limit=10.0
),
"current_spend": 5.0,
"soft_budget_limit": 10.0,
}
assert summary == {
"result": True,
"current_spend": 5.0,
"soft_budget_limit": 10.0,
}
def test_is_projected_spend_over_limit_none_limit_returns_false():
assert (
_is_projected_spend_over_limit(current_spend=10_000.0, soft_budget_limit=None)
is False
)
def test_is_projected_spend_over_limit_raises_when_today_missing(monkeypatch):
class _Broken:
@classmethod
def today(cls):
raise RuntimeError("clock unavailable")
monkeypatch.setattr("litellm.proxy.utils.date", _Broken)
with pytest.raises(RuntimeError):
_is_projected_spend_over_limit(current_spend=1.0, soft_budget_limit=1.0)
def test_get_projected_spend_over_limit_happy_path_over_budget(monkeypatch):
_freeze_today(monkeypatch, date(2024, 1, 11))
result = _get_projected_spend_over_limit(
current_spend=100.0, soft_budget_limit=50.0
)
assert result is not None
projected, exceed_date = result
summary = {
"projected_spend": projected,
"exceed_date": exceed_date.isoformat(),
"current_spend": 100.0,
"soft_budget_limit": 50.0,
}
assert summary == {
"projected_spend": 300.0,
"exceed_date": "2024-01-11",
"current_spend": 100.0,
"soft_budget_limit": 50.0,
}
def test_get_projected_spend_over_limit_first_of_month_uses_current_as_daily(
monkeypatch,
):
_freeze_today(monkeypatch, date(2024, 1, 1))
result = _get_projected_spend_over_limit(current_spend=5.0, soft_budget_limit=10.0)
assert result is not None
projected, exceed_date = result
expected_exceed = date(2024, 1, 1) + timedelta(days=1.0)
summary = {
"projected_spend": projected,
"exceed_date": exceed_date.isoformat(),
"expected_exceed_date": expected_exceed.isoformat(),
"soft_budget_limit": 10.0,
}
assert summary == {
"projected_spend": 155.0,
"exceed_date": expected_exceed.isoformat(),
"expected_exceed_date": expected_exceed.isoformat(),
"soft_budget_limit": 10.0,
}
def test_get_projected_spend_over_limit_zero_daily_spend_exceed_today(monkeypatch):
_freeze_today(monkeypatch, date(2024, 1, 11))
result = _get_projected_spend_over_limit(current_spend=0.0, soft_budget_limit=-1.0)
assert result is not None
projected, exceed_date = result
summary = {
"projected_spend": projected,
"exceed_date": exceed_date.isoformat(),
"soft_budget_limit": -1.0,
}
assert summary == {
"projected_spend": 0.0,
"exceed_date": "2024-01-11",
"soft_budget_limit": -1.0,
}
def test_get_projected_spend_over_limit_under_budget_returns_none(monkeypatch):
_freeze_today(monkeypatch, date(2024, 1, 11))
assert (
_get_projected_spend_over_limit(
current_spend=1.0, soft_budget_limit=1_000_000.0
)
is None
)
def test_get_projected_spend_over_limit_exceed_date_uses_remaining_budget(monkeypatch):
_freeze_today(monkeypatch, date(2024, 1, 11))
result = _get_projected_spend_over_limit(current_spend=20.0, soft_budget_limit=30.0)
assert result is not None
projected, exceed_date = result
daily = 20.0 / 10
remaining_budget = 30.0 - 20.0
expected_exceed = date(2024, 1, 11) + timedelta(days=remaining_budget / daily)
summary = {
"projected_spend": projected,
"exceed_date": exceed_date.isoformat(),
"expected_exceed_date": expected_exceed.isoformat(),
"soft_budget_limit": 30.0,
}
assert summary == {
"projected_spend": 60.0,
"exceed_date": expected_exceed.isoformat(),
"expected_exceed_date": expected_exceed.isoformat(),
"soft_budget_limit": 30.0,
}
def test_get_projected_spend_over_limit_none_limit_returns_none():
assert (
_get_projected_spend_over_limit(current_spend=1.0, soft_budget_limit=None)
is None
)
def test_get_projected_spend_over_limit_raises_when_today_missing(monkeypatch):
class _Broken:
@classmethod
def today(cls):
raise RuntimeError("clock unavailable")
monkeypatch.setattr("litellm.proxy.utils.date", _Broken)
with pytest.raises(RuntimeError):
_get_projected_spend_over_limit(current_spend=1.0, soft_budget_limit=1.0)

77
tests/test_litellm/proxy/utils/helpers/test_premium_user_check.py Normal file
View File

@ -0,0 +1,77 @@
import pytest
from fastapi import HTTPException
from litellm.proxy.utils import _premium_user_check
def normalize(value):
return value
def test_premium_user_check_happy_path_no_raise_when_premium(monkeypatch):
import litellm.proxy.proxy_server as ps
monkeypatch.setattr(ps, "premium_user", True, raising=False)
summary = {
"result": _premium_user_check(),
"premium_user": True,
"raised": False,
}
assert summary == {
"result": None,
"premium_user": True,
"raised": False,
}
def test_premium_user_check_happy_path_with_feature_no_raise(monkeypatch):
import litellm.proxy.proxy_server as ps
monkeypatch.setattr(ps, "premium_user", True, raising=False)
summary = {
"result": _premium_user_check(feature="model-routing"),
"premium_user": True,
"feature": "model-routing",
}
assert summary == {
"result": None,
"premium_user": True,
"feature": "model-routing",
}
def test_premium_user_check_raises_when_not_premium(monkeypatch):
import litellm.proxy.proxy_server as ps
monkeypatch.setattr(ps, "premium_user", False, raising=False)
with pytest.raises(HTTPException) as exc_info:
_premium_user_check()
snapshot = {
"status_code": exc_info.value.status_code,
"is_dict_detail": isinstance(exc_info.value.detail, dict),
"has_error_key": "error" in exc_info.value.detail,
}
assert snapshot == {
"status_code": 403,
"is_dict_detail": True,
"has_error_key": True,
}
def test_premium_user_check_raises_with_feature_message(monkeypatch):
import litellm.proxy.proxy_server as ps
monkeypatch.setattr(ps, "premium_user", False, raising=False)
with pytest.raises(HTTPException) as exc_info:
_premium_user_check(feature="custom-callbacks")
error_msg = exc_info.value.detail["error"]
snapshot = {
"status_code": exc_info.value.status_code,
"feature_in_message": "custom-callbacks" in error_msg,
"enterprise_in_message": "LiteLLM Enterprise" in error_msg,
}
assert snapshot == {
"status_code": 403,
"feature_in_message": True,
"enterprise_in_message": True,
}

76
tests/test_litellm/proxy/utils/helpers/test_team_configs.py Normal file
View File

@ -0,0 +1,76 @@
import pytest
from litellm.proxy.utils import _is_valid_team_configs
def normalize(value):
return value
def test_is_valid_team_configs_happy_path_allowed_model_mutates_config():
team_config = {"models": ["gpt-4o", "gpt-4o-mini"], "max_budget": 100.0}
request_data = {"model": "gpt-4o"}
snapshot = {
"result": _is_valid_team_configs(
team_id="team-1",
team_config=team_config,
request_data=request_data,
),
"models_popped": "models" not in team_config,
"remaining_keys": sorted(team_config.keys()),
}
assert snapshot == {
"result": None,
"models_popped": True,
"remaining_keys": ["max_budget"],
}
def test_is_valid_team_configs_no_models_key_is_noop():
team_config = {"max_budget": 100.0, "tpm_limit": 1000}
request_data = {"model": "anything"}
snapshot = {
"result": _is_valid_team_configs(
team_id="team-1",
team_config=team_config,
request_data=request_data,
),
"team_config": team_config,
"request_data": request_data,
}
assert snapshot == {
"result": None,
"team_config": {"max_budget": 100.0, "tpm_limit": 1000},
"request_data": {"model": "anything"},
}
def test_is_valid_team_configs_short_circuits_when_team_id_none():
team_config = {"models": ["only-this"]}
snapshot = {
"result": _is_valid_team_configs(
team_id=None,
team_config=team_config,
request_data={"model": "anything-else"},
),
"team_config_unchanged": team_config,
"models_key_preserved": "models" in team_config,
}
assert snapshot == {
"result": None,
"team_config_unchanged": {"models": ["only-this"]},
"models_key_preserved": True,
}
def test_is_valid_team_configs_raises_on_model_not_in_team_models():
team_config = {"models": ["gpt-4o"]}
request_data = {"model": "claude-haiku"}
with pytest.raises(Exception) as exc_info:
_is_valid_team_configs(
team_id="team-1",
team_config=team_config,
request_data=request_data,
)
assert "Invalid model for team team-1" in str(exc_info.value)
assert "claude-haiku" in str(exc_info.value)

59
tests/test_litellm/proxy/utils/helpers/test_to_ns.py Normal file
View File

@ -0,0 +1,59 @@
from datetime import datetime, timezone
import pytest
from litellm.proxy.utils import _to_ns
def normalize(value):
return value
def test_to_ns_happy_path_utc_epoch():
dt = datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
expected = int(dt.timestamp() * 1e9)
summary = {
"input_iso": dt.isoformat(),
"result": _to_ns(dt),
"expected": expected,
}
assert summary == {
"input_iso": "2024-01-01T00:00:00+00:00",
"result": expected,
"expected": expected,
}
def test_to_ns_happy_path_microsecond_precision():
dt = datetime(2024, 6, 15, 12, 30, 45, 123456, tzinfo=timezone.utc)
expected = int(dt.timestamp() * 1e9)
summary = {
"input_iso": dt.isoformat(),
"result": _to_ns(dt),
"expected": expected,
}
assert summary == {
"input_iso": "2024-06-15T12:30:45.123456+00:00",
"result": expected,
"expected": expected,
}
def test_to_ns_result_is_int():
dt = datetime(2024, 1, 1, tzinfo=timezone.utc)
result = _to_ns(dt)
summary = {
"type": type(result).__name__,
"is_positive": result > 0,
"result": result,
}
assert summary == {
"type": "int",
"is_positive": True,
"result": int(dt.timestamp() * 1e9),
}
def test_to_ns_raises_on_invalid_input():
with pytest.raises(AttributeError):
_to_ns("2024-01-01T00:00:00")

316
tests/test_litellm/proxy/utils/helpers/test_url_helpers.py Normal file
View File

@ -0,0 +1,316 @@
import pytest
from litellm.proxy.utils import (
_get_docs_url,
_get_openapi_url,
_get_redoc_url,
get_custom_url,
get_proxy_base_url,
get_server_root_path,
join_paths,
normalize_route_for_root_path,
)
def normalize(value):
return value
def _clear_url_env(monkeypatch):
for var in (
"REDOC_URL",
"NO_REDOC",
"DOCS_URL",
"NO_DOCS",
"OPENAPI_URL",
"NO_OPENAPI",
"PROXY_BASE_URL",
"SERVER_ROOT_PATH",
):
monkeypatch.delenv(var, raising=False)
def test_get_redoc_url_default(monkeypatch):
_clear_url_env(monkeypatch)
summary = {
"result": _get_redoc_url(),
"redoc_url_env": None,
"no_redoc_env": None,
}
assert summary == {
"result": "/redoc",
"redoc_url_env": None,
"no_redoc_env": None,
}
def test_get_redoc_url_custom_env(monkeypatch):
_clear_url_env(monkeypatch)
monkeypatch.setenv("REDOC_URL", "/custom-redoc")
summary = {
"result": _get_redoc_url(),
"redoc_url_env": "/custom-redoc",
"default_overridden": True,
}
assert summary == {
"result": "/custom-redoc",
"redoc_url_env": "/custom-redoc",
"default_overridden": True,
}
def test_get_redoc_url_disabled_returns_none_error_path(monkeypatch):
_clear_url_env(monkeypatch)
monkeypatch.setenv("NO_REDOC", "True")
assert _get_redoc_url() is None
def test_get_docs_url_default(monkeypatch):
_clear_url_env(monkeypatch)
summary = {
"result": _get_docs_url(),
"no_docs": None,
"docs_url": None,
}
assert summary == {
"result": "/",
"no_docs": None,
"docs_url": None,
}
def test_get_docs_url_custom_env(monkeypatch):
_clear_url_env(monkeypatch)
monkeypatch.setenv("DOCS_URL", "/api-docs")
summary = {
"result": _get_docs_url(),
"env": "/api-docs",
"default_overridden": True,
}
assert summary == {
"result": "/api-docs",
"env": "/api-docs",
"default_overridden": True,
}
def test_get_docs_url_disabled_returns_none_error_path(monkeypatch):
_clear_url_env(monkeypatch)
monkeypatch.setenv("NO_DOCS", "True")
assert _get_docs_url() is None
def test_get_openapi_url_default(monkeypatch):
_clear_url_env(monkeypatch)
summary = {
"result": _get_openapi_url(),
"no_openapi": None,
"openapi_url": None,
}
assert summary == {
"result": "/openapi.json",
"no_openapi": None,
"openapi_url": None,
}
def test_get_openapi_url_custom_env(monkeypatch):
_clear_url_env(monkeypatch)
monkeypatch.setenv("OPENAPI_URL", "/api-schema")
summary = {
"result": _get_openapi_url(),
"env": "/api-schema",
"default_overridden": True,
}
assert summary == {
"result": "/api-schema",
"env": "/api-schema",
"default_overridden": True,
}
def test_get_openapi_url_disabled_returns_none_error_path(monkeypatch):
_clear_url_env(monkeypatch)
monkeypatch.setenv("NO_OPENAPI", "True")
assert _get_openapi_url() is None
@pytest.mark.parametrize(
"base, route, expected",
[
("https://proxy.example.com", "/v1/chat", "https://proxy.example.com/v1/chat"),
("https://proxy.example.com/", "/v1/chat", "https://proxy.example.com/v1/chat"),
("https://proxy.example.com", "v1/chat", "https://proxy.example.com/v1/chat"),
("https://proxy.example.com", "", "https://proxy.example.com"),
("", "/v1/chat", "/v1/chat"),
("", "", "/"),
],
)
def test_join_paths_happy_path(base, route, expected):
result = join_paths(base, route)
assert {
"input_base": base,
"input_route": route,
"result": result,
"expected": expected,
} == {
"input_base": base,
"input_route": route,
"result": expected,
"expected": expected,
}
def test_join_paths_avoids_duplicating_route_suffix():
summary = {
"result": join_paths("https://api.example.com/v1/chat", "/v1/chat"),
"base": "https://api.example.com/v1/chat",
"route": "/v1/chat",
}
assert summary == {
"result": "https://api.example.com/v1/chat",
"base": "https://api.example.com/v1/chat",
"route": "/v1/chat",
}
def test_join_paths_invalid_input_raises():
with pytest.raises(AttributeError):
join_paths(None, "/v1/chat")
def test_get_proxy_base_url_returns_env_when_set(monkeypatch):
_clear_url_env(monkeypatch)
monkeypatch.setenv("PROXY_BASE_URL", "https://litellm.test")
summary = {
"result": get_proxy_base_url(),
"env": "https://litellm.test",
"is_set": True,
}
assert summary == {
"result": "https://litellm.test",
"env": "https://litellm.test",
"is_set": True,
}
def test_get_proxy_base_url_error_path_returns_none_when_unset(monkeypatch):
_clear_url_env(monkeypatch)
assert get_proxy_base_url() is None
def test_get_server_root_path_returns_env(monkeypatch):
_clear_url_env(monkeypatch)
monkeypatch.setenv("SERVER_ROOT_PATH", "/proxy")
summary = {
"result": get_server_root_path(),
"env": "/proxy",
"is_set": True,
}
assert summary == {
"result": "/proxy",
"env": "/proxy",
"is_set": True,
}
def test_get_server_root_path_error_path_default_empty_string(monkeypatch):
_clear_url_env(monkeypatch)
assert get_server_root_path() == ""
def test_get_custom_url_with_proxy_base_and_root_and_route(monkeypatch):
_clear_url_env(monkeypatch)
monkeypatch.setenv("PROXY_BASE_URL", "https://api.example.com")
monkeypatch.setenv("SERVER_ROOT_PATH", "/proxy")
result = get_custom_url("https://request.example.com", "/v1/chat")
summary = {
"result": result,
"base_used": "PROXY_BASE_URL",
"root_path": "/proxy",
"route": "/v1/chat",
}
assert summary == {
"result": "https://api.example.com/proxy/v1/chat",
"base_used": "PROXY_BASE_URL",
"root_path": "/proxy",
"route": "/v1/chat",
}
def test_get_custom_url_falls_back_to_request_base(monkeypatch):
_clear_url_env(monkeypatch)
result = get_custom_url("https://request.example.com", "/v1/chat")
summary = {
"result": result,
"base_used": "request_base_url",
"root_path": "",
"route": "/v1/chat",
}
assert summary == {
"result": "https://request.example.com/v1/chat",
"base_used": "request_base_url",
"root_path": "",
"route": "/v1/chat",
}
def test_get_custom_url_no_route_uses_root_path(monkeypatch):
_clear_url_env(monkeypatch)
monkeypatch.setenv("SERVER_ROOT_PATH", "/proxy")
result = get_custom_url("https://request.example.com", route=None)
summary = {
"result": result,
"base_used": "request_base_url",
"root_path": "/proxy",
"route": None,
}
assert summary == {
"result": "https://request.example.com/proxy",
"base_used": "request_base_url",
"root_path": "/proxy",
"route": None,
}
def test_get_custom_url_error_path_invalid_base_raises(monkeypatch):
_clear_url_env(monkeypatch)
with pytest.raises(AttributeError):
get_custom_url(None, "/v1/chat")
def test_normalize_route_for_root_path_strips_prefix(monkeypatch):
_clear_url_env(monkeypatch)
monkeypatch.setenv("SERVER_ROOT_PATH", "/proxy")
summary = {
"result": normalize_route_for_root_path("/proxy/v1/chat"),
"root_path": "/proxy",
"input": "/proxy/v1/chat",
}
assert summary == {
"result": "/v1/chat",
"root_path": "/proxy",
"input": "/proxy/v1/chat",
}
def test_normalize_route_for_root_path_returns_route_when_no_root(monkeypatch):
_clear_url_env(monkeypatch)
summary = {
"result": normalize_route_for_root_path("/v1/chat"),
"root_path": "",
"input": "/v1/chat",
}
assert summary == {
"result": "/v1/chat",
"root_path": "",
"input": "/v1/chat",
}
def test_normalize_route_for_root_path_error_path_when_route_not_under_root(
monkeypatch,
):
_clear_url_env(monkeypatch)
monkeypatch.setenv("SERVER_ROOT_PATH", "/proxy")
assert normalize_route_for_root_path("/other/v1/chat") is None