mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-30 06:43:27 +00:00
96 lines
2.9 KiB
Python
96 lines
2.9 KiB
Python
import ast
|
|
import importlib
|
|
import inspect
|
|
import json
|
|
import sys
|
|
from unittest import mock
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
from fastapi.testclient import TestClient
|
|
|
|
import litellm
|
|
from litellm import constants
|
|
|
|
|
|
def _build_constant_env_var_map() -> dict[str, str]:
|
|
"""
|
|
Build a mapping of CONSTANT_NAME -> ENV_VAR_NAME by parsing constants.py.
|
|
|
|
This keeps the test resilient when a constant name and env var name differ
|
|
(e.g., aliases like LITELLM_* env vars).
|
|
"""
|
|
env_var_map: dict[str, str] = {}
|
|
constants_source = inspect.getsource(constants)
|
|
parsed = ast.parse(constants_source)
|
|
|
|
for node in parsed.body:
|
|
if not isinstance(node, ast.Assign):
|
|
continue
|
|
|
|
if len(node.targets) != 1 or not isinstance(node.targets[0], ast.Name):
|
|
continue
|
|
|
|
constant_name = node.targets[0].id
|
|
env_var_name = None
|
|
|
|
for child in ast.walk(node.value):
|
|
if not isinstance(child, ast.Call):
|
|
continue
|
|
|
|
# os.getenv("ENV_NAME", default)
|
|
if (
|
|
isinstance(child.func, ast.Attribute)
|
|
and isinstance(child.func.value, ast.Name)
|
|
and child.func.value.id == "os"
|
|
and child.func.attr == "getenv"
|
|
and len(child.args) >= 1
|
|
and isinstance(child.args[0], ast.Constant)
|
|
and isinstance(child.args[0].value, str)
|
|
):
|
|
env_var_name = child.args[0].value
|
|
break
|
|
|
|
# get_env_int("ENV_NAME", default)
|
|
if (
|
|
isinstance(child.func, ast.Name)
|
|
and child.func.id == "get_env_int"
|
|
and len(child.args) >= 1
|
|
and isinstance(child.args[0], ast.Constant)
|
|
and isinstance(child.args[0].value, str)
|
|
):
|
|
env_var_name = child.args[0].value
|
|
break
|
|
|
|
if env_var_name:
|
|
env_var_map[constant_name] = env_var_name
|
|
|
|
return env_var_map
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("raw", "expected"),
|
|
(
|
|
(None, 64 * 1024 * 1024),
|
|
("", 64 * 1024 * 1024),
|
|
("invalid", 64 * 1024 * 1024),
|
|
("0", 64 * 1024 * 1024),
|
|
("-1", 64 * 1024 * 1024),
|
|
("1", 1),
|
|
("200000", 200000),
|
|
(str(sys.maxsize), sys.maxsize),
|
|
(str(sys.maxsize + 1), 64 * 1024 * 1024),
|
|
),
|
|
)
|
|
def test_mcp_sse_event_limit_uses_positive_override_or_default(raw: str | None, expected: int) -> None:
|
|
try:
|
|
with pytest.MonkeyPatch.context() as environment:
|
|
if raw is None:
|
|
environment.delenv("LITELLM_MCP_MAX_SSE_EVENT_SIZE_BYTES", raising=False)
|
|
else:
|
|
environment.setenv("LITELLM_MCP_MAX_SSE_EVENT_SIZE_BYTES", raw)
|
|
importlib.reload(constants)
|
|
assert constants.MCP_MAX_SSE_EVENT_SIZE_BYTES == expected
|
|
finally:
|
|
importlib.reload(constants)
|