mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-03 02:22:24 +00:00
24584d3d3d
* test(proxy): move proxy_server, _experimental and db tests into tests/unit/proxy Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(proxy): keep tuple identity in proxy state restore and fix misc target paths Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: yuneng <yuneng@berri.ai> Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
1656 lines
67 KiB
Python
1656 lines
67 KiB
Python
"""Pin tests for proxy_server.py control-plane config routes (PR3).
|
|
|
|
Routes covered:
|
|
- POST /config/update
|
|
- POST /config/field/update
|
|
- GET /config/field/info
|
|
- GET /config/list
|
|
- POST /config/field/delete
|
|
- POST /config/callback/delete
|
|
- GET /get/config/callbacks
|
|
- GET /config/yaml
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from .conftest import VOLATILE_KEYS, normalize
|
|
|
|
|
|
def _seed_settings_store(monkeypatch, db_row: dict, yaml_values: dict | None = None) -> None:
|
|
"""Point proxy_config.settings at a store holding the same row the mocked table returns,
|
|
the way a booted proxy does, so the read routes resolve against it."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy.config_resolvers import SettingsStore
|
|
|
|
store = SettingsStore("general_settings")
|
|
store.load_yaml(yaml_values or {})
|
|
store.apply_db_row("general_settings", db_row)
|
|
monkeypatch.setattr(ps.proxy_config, "settings", store)
|
|
|
|
|
|
def _install_litellm_config(mock_prisma: MagicMock) -> MagicMock:
|
|
"""Ensure mock_prisma.db.litellm_config exists with async methods (the
|
|
conftest only stubs ``litellm_configtable`` — this is a different table)."""
|
|
table = MagicMock()
|
|
table.find_unique = AsyncMock(return_value=None)
|
|
table.find_first = AsyncMock(return_value=None)
|
|
table.find_many = AsyncMock(return_value=[])
|
|
table.create = AsyncMock()
|
|
table.update = AsyncMock()
|
|
table.upsert = AsyncMock(return_value=None)
|
|
table.delete = AsyncMock()
|
|
mock_prisma.db.litellm_config = table
|
|
return table
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /config/update
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_config_update_happy_admin(client, auth_as, mock_prisma, monkeypatch):
|
|
"""POST /config/update with admin role merges + upserts general_settings
|
|
and returns the canonical success message."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
fake_proxy_config = MagicMock()
|
|
fake_proxy_config.add_deployment = AsyncMock()
|
|
monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.post(
|
|
"/config/update",
|
|
json={"general_settings": {"alerting": ["slack"]}},
|
|
)
|
|
assert response.status_code == 200
|
|
assert normalize(response.json()) == {"message": "Config updated successfully"}
|
|
|
|
|
|
def test_config_update_persists_optional_pre_call_checks(client, auth_as, mock_prisma, monkeypatch):
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
fake_proxy_config = MagicMock()
|
|
fake_proxy_config.add_deployment = AsyncMock()
|
|
monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.post(
|
|
"/config/update",
|
|
json={"router_settings": {"optional_pre_call_checks": ["prompt_caching"]}},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
persisted = json.loads(table.upsert.call_args.kwargs["data"]["create"]["param_value"])
|
|
assert persisted["optional_pre_call_checks"] == ["prompt_caching"]
|
|
|
|
|
|
def test_config_update_persists_model_group_affinity_config(client, auth_as, mock_prisma, monkeypatch):
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
fake_proxy_config = MagicMock()
|
|
fake_proxy_config.add_deployment = AsyncMock()
|
|
monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
|
|
|
|
model_group_affinity_config = {"gpt-4": ["session_affinity"]}
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.post(
|
|
"/config/update",
|
|
json={"router_settings": {"model_group_affinity_config": model_group_affinity_config}},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
persisted = json.loads(table.upsert.call_args.kwargs["data"]["create"]["param_value"])
|
|
assert persisted["model_group_affinity_config"] == model_group_affinity_config
|
|
|
|
|
|
def test_config_update_persists_disable_cooldowns(client, auth_as, mock_prisma, monkeypatch):
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
fake_proxy_config = MagicMock()
|
|
fake_proxy_config.add_deployment = AsyncMock()
|
|
monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.post(
|
|
"/config/update",
|
|
json={"router_settings": {"disable_cooldowns": True}},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
persisted = json.loads(table.upsert.call_args.kwargs["data"]["create"]["param_value"])
|
|
assert persisted["disable_cooldowns"] is True
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("section", "store_attr", "yaml_values", "changed_values"),
|
|
[
|
|
("general_settings", "settings", {"alerting": ["slack"]}, {"alerting": ["email"]}),
|
|
("litellm_settings", "litellm_settings", {"success_callback": ["langfuse"]}, {"success_callback": ["otel"]}),
|
|
("router_settings", "router_settings", {"num_retries": 0}, {"num_retries": 2}),
|
|
],
|
|
)
|
|
def test_config_update_rejects_config_owned_keys_and_accepts_the_same_value(
|
|
client, auth_as, mock_prisma, monkeypatch, section, store_attr, yaml_values, changed_values
|
|
):
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
monkeypatch.setattr(ps.proxy_config, "add_deployment", AsyncMock())
|
|
store = getattr(ps.proxy_config, store_attr)
|
|
store.load_yaml(yaml_values)
|
|
try:
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
rejected = client.post("/config/update", json={section: changed_values})
|
|
rejected_message = rejected.json()["error"]["message"]
|
|
table.upsert.assert_not_called()
|
|
accepted = client.post("/config/update", json={section: yaml_values})
|
|
finally:
|
|
store.load_yaml({})
|
|
|
|
assert rejected.status_code == 400
|
|
assert f"{section} key '{next(iter(yaml_values))}' is set in the config file and cannot be changed here" in (
|
|
rejected_message
|
|
)
|
|
assert accepted.status_code == 200
|
|
persisted = json.loads(table.upsert.call_args.kwargs["data"]["create"]["param_value"])
|
|
assert persisted[next(iter(yaml_values))] == yaml_values[next(iter(yaml_values))]
|
|
|
|
|
|
def test_config_update_persists_only_the_general_settings_keys_the_request_set(
|
|
client, auth_as, mock_prisma, monkeypatch
|
|
):
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
monkeypatch.setattr(ps.proxy_config, "add_deployment", AsyncMock())
|
|
ps.proxy_config.settings.load_yaml({"health_check_interval": 60})
|
|
try:
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.post("/config/update", json={"general_settings": {"alerting_threshold": 600}})
|
|
finally:
|
|
ps.proxy_config.settings.load_yaml({})
|
|
|
|
assert response.status_code == 200
|
|
persisted = json.loads(table.upsert.call_args.kwargs["data"]["create"]["param_value"])
|
|
assert persisted == {"alerting_threshold": 600}
|
|
|
|
|
|
def test_config_update_persists_only_the_router_settings_keys_the_request_set(
|
|
client, auth_as, mock_prisma, monkeypatch
|
|
):
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
monkeypatch.setattr(ps.proxy_config, "add_deployment", AsyncMock())
|
|
ps.proxy_config.router_settings.load_yaml({"model_group_alias": {"opus": "claude-opus-5"}})
|
|
try:
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.post(
|
|
"/config/update", json={"router_settings": {"retry_policy": {"TimeoutErrorRetries": 3}}}
|
|
)
|
|
finally:
|
|
ps.proxy_config.router_settings.load_yaml({})
|
|
|
|
assert response.status_code == 200, response.text
|
|
persisted = json.loads(table.upsert.call_args.kwargs["data"]["create"]["param_value"])
|
|
assert persisted == {"retry_policy": {"TimeoutErrorRetries": 3}}
|
|
|
|
|
|
def test_config_update_accepts_a_config_owned_success_callback_the_file_spells_in_mixed_case(
|
|
client, auth_as, mock_prisma, monkeypatch
|
|
):
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
monkeypatch.setattr(ps.proxy_config, "add_deployment", AsyncMock())
|
|
ps.proxy_config.litellm_settings.load_yaml({"success_callback": ["Langfuse"]})
|
|
try:
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.post("/config/update", json={"litellm_settings": {"success_callback": ["Langfuse"]}})
|
|
finally:
|
|
ps.proxy_config.litellm_settings.load_yaml({})
|
|
|
|
assert response.status_code == 200
|
|
persisted = json.loads(table.upsert.call_args.kwargs["data"]["create"]["param_value"])
|
|
assert persisted["success_callback"] == ["langfuse"]
|
|
|
|
|
|
def test_config_update_rejects_assistants_config(client, auth_as, mock_prisma, monkeypatch):
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.post(
|
|
"/config/update",
|
|
json={"router_settings": {"assistants_config": {"enabled": True}}},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "assistants_config" in response.json()["error"]["message"]
|
|
table.upsert.assert_not_called()
|
|
|
|
|
|
def test_config_update_rejects_router_general_settings(client, auth_as, mock_prisma, monkeypatch):
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.post(
|
|
"/config/update",
|
|
json={"router_settings": {"router_general_settings": {"async_only_mode": True}}},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "router_general_settings" in response.json()["error"]["message"]
|
|
table.upsert.assert_not_called()
|
|
|
|
|
|
def test_config_update_rejects_unknown_router_setting(client, auth_as, mock_prisma, monkeypatch):
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.post(
|
|
"/config/update",
|
|
json={"router_settings": {"optional_precall_checks": ["prompt_caching"]}},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "optional_precall_checks" in response.json()["error"]["message"]
|
|
table.upsert.assert_not_called()
|
|
|
|
|
|
def test_config_update_unknown_router_setting_non_admin_forbidden(client, auth_as, mock_prisma, monkeypatch):
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
|
|
with auth_as(LitellmUserRoles.INTERNAL_USER):
|
|
response = client.post(
|
|
"/config/update",
|
|
json={"router_settings": {"optional_precall_checks": ["prompt_caching"]}},
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
assert "admin" in response.json()["error"]["message"].lower()
|
|
|
|
|
|
def test_config_update_non_admin_forbidden(client, auth_as, mock_prisma, monkeypatch):
|
|
"""POST /config/update by a non-admin caller is rejected; the error
|
|
surfaces as a ProxyException with the admin-only message."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
|
|
with auth_as(LitellmUserRoles.INTERNAL_USER):
|
|
response = client.post(
|
|
"/config/update",
|
|
json={"general_settings": {"alerting": ["slack"]}},
|
|
)
|
|
assert response.status_code != 200
|
|
body = response.json()
|
|
# ProxyException wraps the 403 detail string in its `message` field.
|
|
assert "admin" in str(body).lower() or "auth" in str(body).lower()
|
|
|
|
|
|
def test_config_update_no_db_error(client, auth_as, monkeypatch):
|
|
"""POST /config/update with prisma_client=None returns a 'No DB Connected'
|
|
style error (the route raises Exception which the handler maps to 400)."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
monkeypatch.setattr(ps, "prisma_client", None)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.post(
|
|
"/config/update",
|
|
json={"general_settings": {"alerting": ["slack"]}},
|
|
)
|
|
assert response.status_code != 200
|
|
assert "db" in str(response.json()).lower() or "connect" in str(response.json()).lower()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /config/field/update
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_config_field_update_happy_admin(client, auth_as, mock_prisma, monkeypatch):
|
|
"""POST /config/field/update for a known field upserts the DB row and
|
|
returns the upsert response (we pin it to a specific shape)."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
table.find_first = AsyncMock(return_value=None)
|
|
upsert_row = {
|
|
"param_name": "general_settings",
|
|
"param_value": {"max_parallel_requests": 5},
|
|
"id": "row-1",
|
|
}
|
|
table.upsert = AsyncMock(return_value=upsert_row)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.post(
|
|
"/config/field/update",
|
|
json={
|
|
"field_name": "max_parallel_requests",
|
|
"field_value": 5,
|
|
"config_type": "general_settings",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert normalize(response.json()) == {
|
|
"param_name": "general_settings",
|
|
"param_value": {"max_parallel_requests": 5},
|
|
"id": "<VOLATILE>",
|
|
}
|
|
|
|
|
|
def test_config_field_update_non_admin_rejected(client, auth_as, mock_prisma, monkeypatch):
|
|
"""Non-admin cannot update config fields — returns 400 with not-allowed
|
|
detail (handler uses 400 for the auth gate, not 403)."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
|
|
with auth_as(LitellmUserRoles.INTERNAL_USER):
|
|
response = client.post(
|
|
"/config/field/update",
|
|
json={
|
|
"field_name": "max_parallel_requests",
|
|
"field_value": 5,
|
|
"config_type": "general_settings",
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
assert "error" in response.json().get("detail", {})
|
|
|
|
|
|
def test_config_field_update_invalid_field(client, auth_as, mock_prisma, monkeypatch):
|
|
"""Unknown field_name is rejected with 400 + 'Invalid field=' detail."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.post(
|
|
"/config/field/update",
|
|
json={
|
|
"field_name": "not_a_real_field_xyz",
|
|
"field_value": 1,
|
|
"config_type": "general_settings",
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
assert "Invalid field" in response.json().get("detail", {}).get("error", "")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /config/field/info
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_config_field_info_happy_admin(client, auth_as, mock_prisma, monkeypatch):
|
|
"""Admin gets back ConfigFieldInfo with the value the proxy resolved, tagged with where it came from."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
row = MagicMock()
|
|
row.param_value = {"max_parallel_requests": 7}
|
|
table.find_first = AsyncMock(return_value=row)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
_seed_settings_store(monkeypatch, row.param_value)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.get("/config/field/info", params={"field_name": "max_parallel_requests"})
|
|
assert response.status_code == 200
|
|
assert normalize(response.json()) == {
|
|
"field_name": "max_parallel_requests",
|
|
"field_value": 7,
|
|
"source": "db",
|
|
"editable": True,
|
|
}
|
|
|
|
|
|
def test_config_field_info_non_admin_rejected(client, auth_as, mock_prisma, monkeypatch):
|
|
"""Non-admin (INTERNAL_USER) is denied — admin-view gate fires."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
|
|
with auth_as(LitellmUserRoles.INTERNAL_USER):
|
|
response = client.get("/config/field/info", params={"field_name": "max_parallel_requests"})
|
|
assert response.status_code == 400
|
|
assert "error" in response.json().get("detail", {})
|
|
|
|
|
|
def test_config_field_info_field_not_in_db(client, auth_as, mock_prisma, monkeypatch):
|
|
"""When nothing sets the field, neither the config file nor the DB row, it 400s."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
row = MagicMock()
|
|
row.param_value = {"some_other_field": "value"}
|
|
table.find_first = AsyncMock(return_value=row)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
_seed_settings_store(monkeypatch, row.param_value)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.get("/config/field/info", params={"field_name": "max_parallel_requests"})
|
|
assert response.status_code == 400
|
|
assert "is not set" in response.json().get("detail", {}).get("error", "")
|
|
|
|
|
|
def test_config_field_info_redacts_nested_secret_for_view_only_admin(client, auth_as, mock_prisma, monkeypatch):
|
|
"""A view-only admin reading a structured field must not receive nested
|
|
credentials. database_args carries aws_web_identity_token (a DynamoDB
|
|
role-assumption credential); it must come back redacted while non-secret
|
|
siblings like region_name stay visible."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
row = MagicMock()
|
|
row.param_value = {
|
|
"database_args": {
|
|
"region_name": "us-east-1",
|
|
"user_table_name": "LiteLLM_UserTable",
|
|
"aws_web_identity_token": "sk-super-secret-token",
|
|
}
|
|
}
|
|
table.find_first = AsyncMock(return_value=row)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
_seed_settings_store(monkeypatch, row.param_value)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY):
|
|
response = client.get("/config/field/info", params={"field_name": "database_args"})
|
|
assert response.status_code == 200
|
|
value = response.json()["field_value"]
|
|
assert value["aws_web_identity_token"] == "REDACTED"
|
|
assert value["region_name"] == "us-east-1"
|
|
assert value["user_table_name"] == "LiteLLM_UserTable"
|
|
|
|
|
|
def test_config_field_info_full_admin_sees_nested_secret(client, auth_as, mock_prisma, monkeypatch):
|
|
"""The redaction must not over-redact for a full PROXY_ADMIN, who needs
|
|
the real nested value to populate the edit form."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
row = MagicMock()
|
|
row.param_value = {
|
|
"database_args": {
|
|
"region_name": "us-east-1",
|
|
"aws_web_identity_token": "sk-super-secret-token",
|
|
}
|
|
}
|
|
table.find_first = AsyncMock(return_value=row)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
_seed_settings_store(monkeypatch, row.param_value)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.get("/config/field/info", params={"field_name": "database_args"})
|
|
assert response.status_code == 200
|
|
value = response.json()["field_value"]
|
|
assert value["aws_web_identity_token"] == "sk-super-secret-token"
|
|
assert value["region_name"] == "us-east-1"
|
|
|
|
|
|
def test_config_field_info_redacts_top_level_scalar_for_view_only(client, auth_as, mock_prisma, monkeypatch):
|
|
"""The top-level scalar branch must also redact for a view-only admin.
|
|
database_url carries DB credentials and is not caught by the name masker,
|
|
so it is in the explicit secret set."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
row = MagicMock()
|
|
row.param_value = {"database_url": "postgresql://admin:p4ss@db:5432/litellm"}
|
|
table.find_first = AsyncMock(return_value=row)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
_seed_settings_store(monkeypatch, row.param_value)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY):
|
|
response = client.get("/config/field/info", params={"field_name": "database_url"})
|
|
assert response.status_code == 200
|
|
assert response.json()["field_value"] == "REDACTED"
|
|
|
|
|
|
def test_redact_general_setting_value_recurses_list_of_dicts():
|
|
"""The list branch of the recursor redacts secret leaves inside each dict
|
|
while non-secret keys survive, and a full admin gets the value untouched."""
|
|
from litellm.proxy import proxy_server as ps
|
|
|
|
value = [
|
|
{"path": "/foo", "headers": {"Authorization": "Bearer sk-x"}},
|
|
{"path": "/bar", "client_secret": "sk-y"},
|
|
]
|
|
redacted = ps._redact_general_setting_value("some_list_field", value, is_full_admin=False)
|
|
assert redacted[0]["headers"]["Authorization"] == "REDACTED"
|
|
assert redacted[0]["path"] == "/foo"
|
|
assert redacted[1]["client_secret"] == "REDACTED"
|
|
assert redacted[1]["path"] == "/bar"
|
|
assert ps._redact_general_setting_value("some_list_field", value, is_full_admin=True) == value
|
|
|
|
|
|
def test_redact_secret_values_in_obj_fails_closed_at_max_depth():
|
|
"""Past _REDACT_SECRET_MAX_DEPTH the whole subtree is replaced with
|
|
"REDACTED" rather than returned verbatim, so a secret buried below the cap
|
|
can never leak via depth-overrun. A future refactor that flips the cap
|
|
branch to fail-open would surface here."""
|
|
from litellm.proxy import proxy_server as ps
|
|
|
|
# leaf and wrap keys are both NON-secret so neither the key-name
|
|
# short-circuit nor the explicit-secret set catches the leak. The cap is
|
|
# the only thing standing between the secret and the response — flip the
|
|
# cap to fail-open and the secret comes back verbatim.
|
|
nested: object = {"notes": "sk-leak-bottom"}
|
|
for _ in range(ps._REDACT_SECRET_MAX_DEPTH + 2):
|
|
nested = {"wrap": nested}
|
|
|
|
out = ps._redact_general_setting_value("some_struct_field", nested, is_full_admin=False)
|
|
# the secret must not survive anywhere in the returned tree
|
|
assert "sk-leak-bottom" not in repr(out)
|
|
|
|
# full admin is unaffected by the cap — the value comes back untouched
|
|
admin_out = ps._redact_general_setting_value("some_struct_field", nested, is_full_admin=True)
|
|
assert admin_out is nested
|
|
|
|
|
|
def test_config_list_redacts_pass_through_secret_for_view_only(client, auth_as, mock_prisma, monkeypatch):
|
|
"""/config/list must not leak pass_through_endpoints upstream credentials
|
|
to a view-only admin. pass_through_endpoints is a known secret-bearing
|
|
field, so a non-admin gets it redacted; a full admin still sees it."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
row = MagicMock()
|
|
row.param_value = {"max_parallel_requests": 3}
|
|
table.find_first = AsyncMock(return_value=row)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
monkeypatch.setattr(
|
|
ps,
|
|
"general_settings",
|
|
{
|
|
"pass_through_endpoints": [
|
|
{
|
|
"path": "/foo",
|
|
"target": "https://upstream.example.com",
|
|
"headers": {"Authorization": "Bearer sk-UPSTREAM-SECRET"},
|
|
}
|
|
]
|
|
},
|
|
)
|
|
|
|
def _pass_through_value(body):
|
|
return next(entry["field_value"] for entry in body if entry["field_name"] == "pass_through_endpoints")
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY):
|
|
view_resp = client.get("/config/list", params={"config_type": "general_settings"})
|
|
assert view_resp.status_code == 200
|
|
assert "sk-UPSTREAM-SECRET" not in view_resp.text
|
|
assert _pass_through_value(view_resp.json()) == "REDACTED"
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
admin_resp = client.get("/config/list", params={"config_type": "general_settings"})
|
|
assert admin_resp.status_code == 200
|
|
admin_value = _pass_through_value(admin_resp.json())
|
|
assert admin_value[0]["headers"]["Authorization"] == "Bearer sk-UPSTREAM-SECRET"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /config/list
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_config_list_happy_admin(client, auth_as, mock_prisma, monkeypatch):
|
|
"""Admin gets a non-empty list of ConfigList rows for general_settings
|
|
(one entry per known allowed_arg). Each row has the documented schema."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
row = MagicMock()
|
|
row.param_value = {"max_parallel_requests": 3}
|
|
table.find_first = AsyncMock(return_value=row)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.get("/config/list", params={"config_type": "general_settings"})
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert isinstance(body, list)
|
|
assert len(body) > 0
|
|
sample = body[0]
|
|
shape = {
|
|
"has_field_name": "field_name" in sample,
|
|
"has_field_type": "field_type" in sample,
|
|
"has_field_value": "field_value" in sample,
|
|
"has_stored_in_db": "stored_in_db" in sample,
|
|
}
|
|
assert shape == {
|
|
"has_field_name": True,
|
|
"has_field_type": True,
|
|
"has_field_value": True,
|
|
"has_stored_in_db": True,
|
|
}
|
|
|
|
|
|
def test_config_list_exposes_config_reload_interval(client, auth_as, mock_prisma, monkeypatch):
|
|
"""proxy_config_reload_interval_seconds must surface in the admin UI general-settings
|
|
list as an Integer field defaulting to 30, so operators can tune multi-pod convergence
|
|
from the dashboard."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
row = MagicMock()
|
|
row.param_value = {}
|
|
table.find_first = AsyncMock(return_value=row)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.get("/config/list", params={"config_type": "general_settings"})
|
|
assert response.status_code == 200
|
|
by_name = {entry["field_name"]: entry for entry in response.json()}
|
|
assert "proxy_config_reload_interval_seconds" in by_name
|
|
entry = by_name["proxy_config_reload_interval_seconds"]
|
|
assert entry["field_type"] == "Integer"
|
|
assert entry["field_default_value"] == 30
|
|
|
|
|
|
def test_config_field_update_accepts_config_reload_interval(client, auth_as, mock_prisma, monkeypatch):
|
|
"""POST /config/field/update accepts proxy_config_reload_interval_seconds and persists
|
|
it to the DB general_settings row for all pods to pick up."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
table.find_first = AsyncMock(return_value=None)
|
|
upsert_row = {
|
|
"param_name": "general_settings",
|
|
"param_value": {"proxy_config_reload_interval_seconds": 45},
|
|
"id": "row-1",
|
|
}
|
|
table.upsert = AsyncMock(return_value=upsert_row)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.post(
|
|
"/config/field/update",
|
|
json={
|
|
"field_name": "proxy_config_reload_interval_seconds",
|
|
"field_value": 45,
|
|
"config_type": "general_settings",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
upserted = table.upsert.call_args.kwargs["data"]["create"]["param_value"]
|
|
assert json.loads(upserted)["proxy_config_reload_interval_seconds"] == 45
|
|
|
|
|
|
def test_config_field_update_rejects_non_positive_config_reload_interval(client, auth_as, mock_prisma, monkeypatch):
|
|
"""A non-positive proxy_config_reload_interval_seconds from the UI is rejected with a 400
|
|
and never persisted, since APScheduler requires a positive interval."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
table.find_first = AsyncMock(return_value=None)
|
|
table.upsert = AsyncMock()
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.post(
|
|
"/config/field/update",
|
|
json={
|
|
"field_name": "proxy_config_reload_interval_seconds",
|
|
"field_value": 0,
|
|
"config_type": "general_settings",
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
table.upsert.assert_not_called()
|
|
|
|
|
|
def test_config_list_non_admin_rejected(client, auth_as, mock_prisma, monkeypatch):
|
|
"""Non-admin gets a 400 with the role embedded in the error message."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
|
|
with auth_as(LitellmUserRoles.INTERNAL_USER):
|
|
response = client.get("/config/list", params={"config_type": "general_settings"})
|
|
assert response.status_code == 400
|
|
assert "role" in response.json().get("detail", {}).get("error", "").lower()
|
|
|
|
|
|
def test_config_list_no_db_error(client, auth_as, monkeypatch):
|
|
"""No DB → 400 with db_not_connected error."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
monkeypatch.setattr(ps, "prisma_client", None)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.get("/config/list", params={"config_type": "general_settings"})
|
|
assert response.status_code == 400
|
|
assert "error" in response.json().get("detail", {})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /config/field/delete
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_config_field_delete_happy_admin(client, auth_as, mock_prisma, monkeypatch):
|
|
"""Admin can delete a stored general_settings field — returns the upsert row."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
existing = MagicMock()
|
|
existing.param_value = {"max_parallel_requests": 5, "other": "value"}
|
|
table.find_first = AsyncMock(return_value=existing)
|
|
table.upsert = AsyncMock(
|
|
return_value={
|
|
"param_name": "general_settings",
|
|
"param_value": {"other": "value"},
|
|
"id": "row-1",
|
|
}
|
|
)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.post(
|
|
"/config/field/delete",
|
|
json={
|
|
"config_type": "general_settings",
|
|
"field_name": "max_parallel_requests",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert normalize(response.json()) == {
|
|
"param_name": "general_settings",
|
|
"param_value": {"other": "value"},
|
|
"id": "<VOLATILE>",
|
|
}
|
|
|
|
|
|
def test_config_field_delete_non_admin_rejected(client, auth_as, mock_prisma, monkeypatch):
|
|
"""Non-admin caller hits the 400 not-allowed branch with role in detail."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
|
|
with auth_as(LitellmUserRoles.INTERNAL_USER):
|
|
response = client.post(
|
|
"/config/field/delete",
|
|
json={
|
|
"config_type": "general_settings",
|
|
"field_name": "max_parallel_requests",
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
assert "role" in response.json().get("detail", {}).get("error", "").lower()
|
|
|
|
|
|
def test_config_field_delete_field_not_in_config(client, auth_as, mock_prisma, monkeypatch):
|
|
"""If there is no general_settings row at all, returns 400 'not in config'."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
table = _install_litellm_config(mock_prisma)
|
|
table.find_first = AsyncMock(return_value=None)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.post(
|
|
"/config/field/delete",
|
|
json={
|
|
"config_type": "general_settings",
|
|
"field_name": "max_parallel_requests",
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
assert "not in config" in response.json().get("detail", {}).get("error", "")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /config/callback/delete
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_config_callback_delete_happy_admin(client, auth_as, mock_prisma, monkeypatch):
|
|
"""Admin deletes a configured success callback — handler returns the
|
|
success message + remaining callbacks + a timestamp."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
monkeypatch.setattr(ps, "store_model_in_db", True)
|
|
|
|
fake_proxy_config = MagicMock()
|
|
fake_proxy_config.get_config = AsyncMock(
|
|
return_value={"litellm_settings": {"success_callback": ["langfuse", "slack"]}}
|
|
)
|
|
fake_proxy_config.save_config = AsyncMock()
|
|
fake_proxy_config.add_deployment = AsyncMock()
|
|
monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.post("/config/callback/delete", json={"callback_name": "langfuse"})
|
|
assert response.status_code == 200
|
|
# `deleted_at` is an ISO timestamp generated at request time — extend
|
|
# the volatile set just for this assertion so dict-equality still works.
|
|
volatile = VOLATILE_KEYS | {"deleted_at"}
|
|
assert normalize(response.json(), volatile) == {
|
|
"message": "Successfully deleted callback: langfuse",
|
|
"removed_callback": "langfuse",
|
|
"remaining_callbacks": ["slack"],
|
|
"deleted_at": "<VOLATILE>",
|
|
}
|
|
|
|
|
|
def test_config_callback_delete_non_admin_rejected(client, auth_as, mock_prisma, monkeypatch):
|
|
"""Non-admin caller is rejected with 400 not-allowed."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
monkeypatch.setattr(ps, "store_model_in_db", True)
|
|
|
|
with auth_as(LitellmUserRoles.INTERNAL_USER):
|
|
response = client.post("/config/callback/delete", json={"callback_name": "langfuse"})
|
|
assert response.status_code == 400
|
|
assert "role" in response.json().get("detail", {}).get("error", "").lower()
|
|
|
|
|
|
def test_config_callback_delete_not_found(client, auth_as, mock_prisma, monkeypatch):
|
|
"""Callback missing from current config returns 404."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
monkeypatch.setattr(ps, "store_model_in_db", True)
|
|
|
|
fake_proxy_config = MagicMock()
|
|
fake_proxy_config.get_config = AsyncMock(return_value={"litellm_settings": {"success_callback": ["slack"]}})
|
|
monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.post("/config/callback/delete", json={"callback_name": "langfuse"})
|
|
# The handler re-raises HTTPException(404) verbatim (only generic
|
|
# `Exception` becomes a 500 ProxyException), so pin 404 strictly.
|
|
assert response.status_code == 404
|
|
assert "langfuse" in str(response.json()).lower() or "not found" in str(response.json()).lower()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /get/config/callbacks
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_get_config_callbacks_happy(client, auth_as, mock_prisma, monkeypatch):
|
|
"""GET /get/config/callbacks returns the 5 pinned top-level keys:
|
|
status, callbacks, alerts, router_settings, available_callbacks."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
monkeypatch.setattr(ps, "llm_router", None)
|
|
|
|
fake_proxy_config = MagicMock()
|
|
fake_proxy_config.get_config = AsyncMock(
|
|
return_value={
|
|
"litellm_settings": {"success_callback": []},
|
|
"general_settings": {},
|
|
"environment_variables": {},
|
|
}
|
|
)
|
|
monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.get("/get/config/callbacks")
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
shape = {
|
|
"status": body.get("status"),
|
|
"has_callbacks": "callbacks" in body,
|
|
"has_alerts": "alerts" in body,
|
|
"has_router_settings": "router_settings" in body,
|
|
"has_available_callbacks": "available_callbacks" in body,
|
|
}
|
|
assert shape == {
|
|
"status": "success",
|
|
"has_callbacks": True,
|
|
"has_alerts": True,
|
|
"has_router_settings": True,
|
|
"has_available_callbacks": True,
|
|
}
|
|
|
|
|
|
def test_get_config_callbacks_internal_error(client, auth_as, mock_prisma, monkeypatch):
|
|
"""If proxy_config.get_config() raises, the handler wraps the failure in
|
|
a ProxyException → non-2xx response with an error body."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
|
|
fake_proxy_config = MagicMock()
|
|
fake_proxy_config.get_config = AsyncMock(side_effect=RuntimeError("boom"))
|
|
monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.get("/get/config/callbacks")
|
|
assert response.status_code >= 400
|
|
assert "boom" in str(response.json()).lower() or "error" in str(response.json()).lower()
|
|
|
|
|
|
_CALLBACK_ENV_FIXTURE = {
|
|
"LANGFUSE_PUBLIC_KEY": "pk-public-1234567890",
|
|
"LANGFUSE_SECRET_KEY": "sk-langfuse-super-secret",
|
|
"LANGFUSE_HOST": "https://cloud.langfuse.com",
|
|
"DD_API_KEY": "dd-super-secret-api-key",
|
|
"DD_SITE": "datadoghq.com",
|
|
"OTEL_HEADERS": "Authorization=Bearer otel-super-secret",
|
|
"OTEL_ENDPOINT": "https://otlp.example.com",
|
|
"SLACK_WEBHOOK_URL": "https://hooks.slack.com/services/T000/B000/SLACK-WEBHOOK-FIXTURE-SECRET",
|
|
}
|
|
|
|
|
|
def _install_callbacks_config(monkeypatch, mock_prisma):
|
|
from litellm.proxy import proxy_server as ps
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
monkeypatch.setattr(ps, "llm_router", None)
|
|
|
|
fake_proxy_config = MagicMock()
|
|
fake_proxy_config.get_config = AsyncMock(
|
|
return_value={
|
|
"litellm_settings": {"success_callback": ["langfuse", "datadog", "otel"]},
|
|
"general_settings": {"alerting": ["slack"]},
|
|
"environment_variables": dict(_CALLBACK_ENV_FIXTURE),
|
|
}
|
|
)
|
|
monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
|
|
|
|
|
|
def _callback_variables(body: dict, name: str) -> dict:
|
|
return next(cb["variables"] for cb in body["callbacks"] if cb["name"] == name)
|
|
|
|
|
|
def test_get_config_callbacks_redacts_secret_env_vars_for_view_only_admin(client, auth_as, mock_prisma, monkeypatch):
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_callbacks_config(monkeypatch, mock_prisma)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY):
|
|
response = client.get("/get/config/callbacks")
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
|
|
for secret in (
|
|
_CALLBACK_ENV_FIXTURE["LANGFUSE_SECRET_KEY"],
|
|
_CALLBACK_ENV_FIXTURE["DD_API_KEY"],
|
|
_CALLBACK_ENV_FIXTURE["OTEL_HEADERS"],
|
|
_CALLBACK_ENV_FIXTURE["LANGFUSE_PUBLIC_KEY"],
|
|
):
|
|
assert secret not in response.text
|
|
|
|
langfuse_vars = _callback_variables(body, "langfuse")
|
|
assert langfuse_vars["LANGFUSE_PUBLIC_KEY"] == "REDACTED"
|
|
assert langfuse_vars["LANGFUSE_SECRET_KEY"] == "REDACTED"
|
|
assert langfuse_vars["LANGFUSE_HOST"] == _CALLBACK_ENV_FIXTURE["LANGFUSE_HOST"]
|
|
|
|
datadog_vars = _callback_variables(body, "datadog")
|
|
assert datadog_vars["DD_API_KEY"] == "REDACTED"
|
|
assert datadog_vars["DD_SITE"] == _CALLBACK_ENV_FIXTURE["DD_SITE"]
|
|
|
|
otel_vars = _callback_variables(body, "otel")
|
|
assert otel_vars["OTEL_HEADERS"] == "REDACTED"
|
|
assert otel_vars["OTEL_ENDPOINT"] == _CALLBACK_ENV_FIXTURE["OTEL_ENDPOINT"]
|
|
|
|
|
|
def test_get_config_callbacks_full_admin_still_sees_secret_env_vars(client, auth_as, mock_prisma, monkeypatch):
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_callbacks_config(monkeypatch, mock_prisma)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.get("/get/config/callbacks")
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
|
|
langfuse_vars = _callback_variables(body, "langfuse")
|
|
assert langfuse_vars["LANGFUSE_SECRET_KEY"] == _CALLBACK_ENV_FIXTURE["LANGFUSE_SECRET_KEY"]
|
|
assert langfuse_vars["LANGFUSE_PUBLIC_KEY"] == _CALLBACK_ENV_FIXTURE["LANGFUSE_PUBLIC_KEY"]
|
|
|
|
datadog_vars = _callback_variables(body, "datadog")
|
|
assert datadog_vars["DD_API_KEY"] == _CALLBACK_ENV_FIXTURE["DD_API_KEY"]
|
|
|
|
otel_vars = _callback_variables(body, "otel")
|
|
assert otel_vars["OTEL_HEADERS"] == _CALLBACK_ENV_FIXTURE["OTEL_HEADERS"]
|
|
|
|
|
|
def test_get_config_callbacks_redacts_slack_webhook_urls_for_view_only_admin(client, auth_as, mock_prisma, monkeypatch):
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_callbacks_config(monkeypatch, mock_prisma)
|
|
|
|
webhooks = {
|
|
"spend_reports": "https://hooks.slack.com/services/T000/B000/SPEND-WEBHOOK-SECRET",
|
|
"budget_alerts": "https://hooks.slack.com/services/T000/B111/BUDGET-WEBHOOK-SECRET",
|
|
}
|
|
monkeypatch.setattr(
|
|
ps.proxy_logging_obj.slack_alerting_instance,
|
|
"alert_to_webhook_url",
|
|
webhooks,
|
|
raising=False,
|
|
)
|
|
|
|
def _slack_block(body):
|
|
return next(a for a in body["alerts"] if a["name"] == "slack")
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY):
|
|
view_resp = client.get("/get/config/callbacks")
|
|
assert view_resp.status_code == 200
|
|
for url in webhooks.values():
|
|
assert url not in view_resp.text
|
|
assert _CALLBACK_ENV_FIXTURE["SLACK_WEBHOOK_URL"] not in view_resp.text
|
|
view_slack = _slack_block(view_resp.json())
|
|
assert view_slack["alerts_to_webhook"] == {
|
|
"spend_reports": "REDACTED",
|
|
"budget_alerts": "REDACTED",
|
|
}
|
|
assert view_slack["variables"]["SLACK_WEBHOOK_URL"] == "REDACTED"
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
admin_resp = client.get("/get/config/callbacks")
|
|
assert admin_resp.status_code == 200
|
|
admin_slack = _slack_block(admin_resp.json())
|
|
assert admin_slack["alerts_to_webhook"] == webhooks
|
|
assert admin_slack["variables"]["SLACK_WEBHOOK_URL"] != "REDACTED"
|
|
|
|
|
|
def test_redact_callback_env_vars_helper_handles_none_and_non_secret_keys():
|
|
from litellm.proxy import proxy_server as ps
|
|
|
|
out = ps._redact_callback_env_vars(
|
|
{
|
|
"LANGFUSE_SECRET_KEY": "sk-leak",
|
|
"LANGFUSE_HOST": "https://cloud.langfuse.com",
|
|
"DD_API_KEY": None,
|
|
"GALILEO_USERNAME": "galileo-user-1234",
|
|
"GENERIC_LOGGER_HEADERS": "Authorization=Bearer x",
|
|
"GCS_PATH_SERVICE_ACCOUNT": "/etc/secrets/gcs.json",
|
|
"SLACK_WEBHOOK_URL": "https://hooks.slack.com/services/T/B/token",
|
|
"SMTP_USERNAME": "smtp-user-1234",
|
|
}
|
|
)
|
|
assert out == {
|
|
"LANGFUSE_SECRET_KEY": "REDACTED",
|
|
"LANGFUSE_HOST": "https://cloud.langfuse.com",
|
|
"DD_API_KEY": None,
|
|
"GALILEO_USERNAME": "REDACTED",
|
|
"GENERIC_LOGGER_HEADERS": "REDACTED",
|
|
"GCS_PATH_SERVICE_ACCOUNT": "REDACTED",
|
|
"SLACK_WEBHOOK_URL": "REDACTED",
|
|
"SMTP_USERNAME": "REDACTED",
|
|
}
|
|
|
|
|
|
def test_get_config_callbacks_redacts_email_alerting_vars_for_view_only_admin(
|
|
client, auth_as, mock_prisma, monkeypatch
|
|
):
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
monkeypatch.setattr(ps, "llm_router", None)
|
|
|
|
fake_proxy_config = MagicMock()
|
|
fake_proxy_config.get_config = AsyncMock(
|
|
return_value={
|
|
"litellm_settings": {"success_callback": []},
|
|
"general_settings": {"alerting": ["email"]},
|
|
"environment_variables": {
|
|
"SMTP_HOST": "smtp.resend.com",
|
|
"SMTP_PORT": "587",
|
|
"SMTP_USERNAME": "smtp-user-fixture-1234",
|
|
"SMTP_PASSWORD": "smtp-password-fixture-1234",
|
|
"SMTP_SENDER_EMAIL": "alerts@example.com",
|
|
"TEST_EMAIL_ADDRESS": "admin@example.com",
|
|
"EMAIL_LOGO_URL": "https://example.com/logo.png",
|
|
"EMAIL_SUPPORT_CONTACT": "support@example.com",
|
|
},
|
|
}
|
|
)
|
|
monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
|
|
|
|
def _email_block(body):
|
|
return next(a for a in body["alerts"] if a["name"] == "email")
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY):
|
|
view_resp = client.get("/get/config/callbacks")
|
|
assert view_resp.status_code == 200
|
|
for secret in ("smtp-user-fixture-1234", "smtp-password-fixture-1234"):
|
|
assert secret not in view_resp.text
|
|
view_email = _email_block(view_resp.json())["variables"]
|
|
assert view_email["SMTP_PASSWORD"] == "REDACTED"
|
|
assert view_email["SMTP_USERNAME"] == "REDACTED"
|
|
assert view_email["SMTP_HOST"] == "smtp.resend.com"
|
|
assert view_email["SMTP_PORT"] == "587"
|
|
assert view_email["SMTP_SENDER_EMAIL"] == "alerts@example.com"
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
admin_resp = client.get("/get/config/callbacks")
|
|
assert admin_resp.status_code == 200
|
|
admin_email = _email_block(admin_resp.json())["variables"]
|
|
assert admin_email["SMTP_USERNAME"] == "smtp-user-fixture-1234"
|
|
assert admin_email["SMTP_PASSWORD"] != "REDACTED"
|
|
assert admin_email["SMTP_HOST"] == "smtp.resend.com"
|
|
|
|
|
|
def test_get_config_callbacks_appends_runtime_only_callbacks(client, auth_as, mock_prisma, monkeypatch):
|
|
"""LIT-5281: a YAML callback that the DB callback list replaced in the merged config still runs, so it must
|
|
show up as a read_only row next to the editable DB-configured one."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
monkeypatch.setattr(ps, "llm_router", None)
|
|
|
|
fake_proxy_config = MagicMock()
|
|
fake_proxy_config.get_config = AsyncMock(
|
|
return_value={
|
|
"litellm_settings": {"success_callback": ["langfuse"]},
|
|
"general_settings": {},
|
|
"environment_variables": dict(_CALLBACK_ENV_FIXTURE),
|
|
}
|
|
)
|
|
monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
|
|
|
|
import litellm
|
|
from litellm.integrations.langsmith import LangsmithLogger
|
|
from litellm.integrations.opentelemetry import OpenTelemetry
|
|
|
|
monkeypatch.setattr(litellm, "success_callback", ["langfuse", LangsmithLogger()])
|
|
monkeypatch.setattr(litellm, "_async_success_callback", [])
|
|
monkeypatch.setattr(litellm, "failure_callback", [])
|
|
monkeypatch.setattr(litellm, "_async_failure_callback", [])
|
|
monkeypatch.setattr(litellm, "callbacks", [OpenTelemetry()])
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.get("/get/config/callbacks")
|
|
assert response.status_code == 200
|
|
|
|
assert [(cb["name"], cb["type"], cb.get("read_only", False)) for cb in response.json()["callbacks"]] == [
|
|
("langfuse", "success", False),
|
|
("langsmith", "success", True),
|
|
("otel", "success_and_failure", True),
|
|
]
|
|
|
|
|
|
def test_get_config_callbacks_accepts_scalar_and_null_yaml_callbacks(client, auth_as, mock_prisma, monkeypatch):
|
|
"""`success_callback: langfuse` (a YAML scalar) is one configured callback, not eight single-letter ones, and a
|
|
`callbacks: null` key contributes nothing."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
monkeypatch.setattr(ps, "llm_router", None)
|
|
|
|
fake_proxy_config = MagicMock()
|
|
fake_proxy_config.get_config = AsyncMock(
|
|
return_value={
|
|
"litellm_settings": {"success_callback": "langfuse", "failure_callback": None, "callbacks": None},
|
|
"general_settings": {},
|
|
"environment_variables": dict(_CALLBACK_ENV_FIXTURE),
|
|
}
|
|
)
|
|
monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
|
|
|
|
import litellm
|
|
from litellm.integrations.langsmith import LangsmithLogger
|
|
|
|
monkeypatch.setattr(litellm, "success_callback", ["langfuse", LangsmithLogger()])
|
|
monkeypatch.setattr(litellm, "_async_success_callback", [])
|
|
monkeypatch.setattr(litellm, "failure_callback", [])
|
|
monkeypatch.setattr(litellm, "_async_failure_callback", [])
|
|
monkeypatch.setattr(litellm, "callbacks", [])
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.get("/get/config/callbacks")
|
|
assert response.status_code == 200
|
|
|
|
assert [(cb["name"], cb["type"], cb.get("read_only", False)) for cb in response.json()["callbacks"]] == [
|
|
("langfuse", "success", False),
|
|
("langsmith", "success", True),
|
|
]
|
|
|
|
|
|
def test_get_config_callbacks_deduplicates_configured_and_runtime(client, auth_as, mock_prisma, monkeypatch):
|
|
"""A configured callback shows once as editable, whether the runtime holds its string or an initialized instance
|
|
(arize initializes an ArizeLogger, logfire a bare OpenTelemetry that only its class identifies)."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
monkeypatch.setattr(ps, "llm_router", None)
|
|
|
|
fake_proxy_config = MagicMock()
|
|
fake_proxy_config.get_config = AsyncMock(
|
|
return_value={
|
|
"litellm_settings": {"success_callback": ["langfuse", "arize", "logfire"]},
|
|
"general_settings": {},
|
|
"environment_variables": dict(_CALLBACK_ENV_FIXTURE),
|
|
}
|
|
)
|
|
monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
|
|
|
|
import litellm
|
|
from litellm.integrations.arize.arize import ArizeLogger
|
|
from litellm.integrations.opentelemetry import OpenTelemetry, OpenTelemetryConfig
|
|
|
|
arize_logger = ArizeLogger(config=OpenTelemetryConfig(exporter="console"), callback_name="arize")
|
|
monkeypatch.setattr(litellm, "success_callback", ["langfuse", arize_logger, OpenTelemetry()])
|
|
monkeypatch.setattr(litellm, "callbacks", [])
|
|
monkeypatch.setattr(litellm, "failure_callback", [])
|
|
monkeypatch.setattr(litellm, "_async_success_callback", [])
|
|
monkeypatch.setattr(litellm, "_async_failure_callback", [])
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.get("/get/config/callbacks")
|
|
assert response.status_code == 200
|
|
|
|
assert [(cb["name"], cb["type"], cb.get("read_only", False)) for cb in response.json()["callbacks"]] == [
|
|
("langfuse", "success", False),
|
|
("arize", "success", False),
|
|
("logfire", "success", False),
|
|
]
|
|
|
|
|
|
def test_get_config_callbacks_keeps_yaml_otel_family_callbacks_next_to_configured_one(
|
|
client, auth_as, mock_prisma, monkeypatch
|
|
):
|
|
"""LIT-5281: arize, weave_otel and langfuse_otel all initialize OpenTelemetry subclasses. Saving one of them
|
|
from the dashboard replaces the YAML `callbacks` list, so the YAML siblings keep running and must stay listed
|
|
under their own names instead of being hidden as duplicates of the configured OTel callback."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
monkeypatch.setattr(ps, "llm_router", None)
|
|
|
|
fake_proxy_config = MagicMock()
|
|
fake_proxy_config.get_config = AsyncMock(
|
|
return_value={
|
|
"litellm_settings": {"callbacks": ["langfuse_otel"]},
|
|
"general_settings": {},
|
|
"environment_variables": dict(_CALLBACK_ENV_FIXTURE),
|
|
}
|
|
)
|
|
monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
|
|
|
|
import litellm
|
|
from litellm.integrations.arize.arize import ArizeLogger
|
|
from litellm.integrations.langfuse.langfuse_otel import LangfuseOtelLogger
|
|
from litellm.integrations.langsmith import LangsmithLogger
|
|
from litellm.integrations.opentelemetry import OpenTelemetryConfig
|
|
from litellm.integrations.weave.weave_otel import WeaveOtelLogger
|
|
|
|
console_config = OpenTelemetryConfig(exporter="console")
|
|
monkeypatch.setattr(litellm, "success_callback", [LangsmithLogger()])
|
|
monkeypatch.setattr(litellm, "_async_success_callback", [])
|
|
monkeypatch.setattr(litellm, "failure_callback", [])
|
|
monkeypatch.setattr(litellm, "_async_failure_callback", [])
|
|
monkeypatch.setattr(
|
|
litellm,
|
|
"callbacks",
|
|
[
|
|
ArizeLogger(config=console_config, callback_name="arize"),
|
|
WeaveOtelLogger(config=console_config),
|
|
LangfuseOtelLogger(config=console_config, callback_name="langfuse_otel"),
|
|
],
|
|
)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.get("/get/config/callbacks")
|
|
assert response.status_code == 200
|
|
|
|
assert [(cb["name"], cb["type"], cb.get("read_only", False)) for cb in response.json()["callbacks"]] == [
|
|
("langfuse_otel", "success_and_failure", False),
|
|
("arize", "success_and_failure", True),
|
|
("langsmith", "success", True),
|
|
("weave_otel", "success_and_failure", True),
|
|
]
|
|
|
|
|
|
def _dotted_path_test_function(*args, **kwargs):
|
|
pass
|
|
|
|
|
|
@pytest.mark.parametrize("handler_kind", ["instance", "function"])
|
|
@pytest.mark.parametrize(
|
|
"config_key,expected_type",
|
|
[
|
|
("success_callback", "success"),
|
|
("failure_callback", "failure"),
|
|
("callbacks", "success_and_failure"),
|
|
],
|
|
)
|
|
def test_get_config_callbacks_deduplicates_dotted_path_callback(
|
|
client, auth_as, mock_prisma, monkeypatch, config_key, expected_type, handler_kind
|
|
):
|
|
"""A dotted-path callback stays a single editable row instead of duplicating under its class or function name."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
monkeypatch.setattr(ps, "llm_router", None)
|
|
|
|
import litellm
|
|
from litellm.integrations.custom_logger import CustomLogger
|
|
|
|
class _DottedPathTestHandler(CustomLogger):
|
|
pass
|
|
|
|
dotted_handler = _DottedPathTestHandler() if handler_kind == "instance" else _dotted_path_test_function
|
|
dotted_path = f"{__name__}.dotted_handler"
|
|
|
|
fake_proxy_config = MagicMock()
|
|
fake_proxy_config.get_config = AsyncMock(
|
|
return_value={
|
|
"litellm_settings": {config_key: [dotted_path]},
|
|
"general_settings": {},
|
|
"environment_variables": dict(_CALLBACK_ENV_FIXTURE),
|
|
}
|
|
)
|
|
monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
|
|
|
|
monkeypatch.setattr(litellm, "callbacks", [dotted_handler])
|
|
monkeypatch.setattr(litellm, "success_callback", [])
|
|
monkeypatch.setattr(litellm, "failure_callback", [])
|
|
monkeypatch.setattr(litellm, "_async_success_callback", [])
|
|
monkeypatch.setattr(litellm, "_async_failure_callback", [])
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.get("/get/config/callbacks")
|
|
|
|
assert response.status_code == 200
|
|
callbacks = response.json()["callbacks"]
|
|
assert [(callback["name"], callback["type"], callback.get("read_only", False)) for callback in callbacks] == [
|
|
(dotted_path, expected_type, False)
|
|
]
|
|
|
|
|
|
def test_get_config_callbacks_lists_dict_shaped_config_callbacks(client, auth_as, mock_prisma, monkeypatch):
|
|
"""Dict-shaped success_callback config values list their keys as editable rows."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
monkeypatch.setattr(ps, "llm_router", None)
|
|
|
|
fake_proxy_config = MagicMock()
|
|
fake_proxy_config.get_config = AsyncMock(
|
|
return_value={
|
|
"litellm_settings": {"success_callback": {"langsmith": {"batch_size": 1}}},
|
|
"general_settings": {},
|
|
"environment_variables": dict(_CALLBACK_ENV_FIXTURE),
|
|
}
|
|
)
|
|
monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
|
|
|
|
import litellm
|
|
|
|
monkeypatch.setattr(
|
|
litellm.logging_callback_manager,
|
|
"get_callbacks_by_type",
|
|
MagicMock(return_value={"success": ["langsmith"], "failure": [], "success_and_failure": []}),
|
|
)
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.get("/get/config/callbacks")
|
|
|
|
assert response.status_code == 200
|
|
callbacks = response.json()["callbacks"]
|
|
assert [(callback["name"], callback.get("read_only", False)) for callback in callbacks] == [("langsmith", False)]
|
|
|
|
|
|
def test_get_config_callbacks_excludes_internal_runtime_callbacks(client, auth_as, mock_prisma, monkeypatch):
|
|
"""Proxy infrastructure callbacks are excluded from callback inventory."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
monkeypatch.setattr(ps, "llm_router", None)
|
|
|
|
fake_proxy_config = MagicMock()
|
|
fake_proxy_config.get_config = AsyncMock(
|
|
return_value={
|
|
"litellm_settings": {"success_callback": []},
|
|
"general_settings": {},
|
|
"environment_variables": dict(_CALLBACK_ENV_FIXTURE),
|
|
}
|
|
)
|
|
monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
|
|
|
|
from litellm_enterprise.proxy.hooks.managed_files import _PROXY_LiteLLMManagedFiles
|
|
|
|
import litellm
|
|
from litellm._service_logger import ServiceLogging
|
|
from litellm.integrations.custom_guardrail import CustomGuardrail
|
|
from litellm.integrations.custom_logger import CustomLogger
|
|
from litellm.integrations.langsmith import LangsmithLogger
|
|
from litellm.integrations.s3_v2 import S3Logger
|
|
from litellm.integrations.sqs import SQSLogger
|
|
from litellm.integrations.vector_store_integrations.vector_store_pre_call_hook import VectorStorePreCallHook
|
|
from litellm.proxy.hooks.cache_control_check import _PROXY_CacheControlCheck
|
|
from litellm.router import Router
|
|
|
|
class _InventoryTestGuardrail(CustomGuardrail):
|
|
pass
|
|
|
|
class _UserCodeLogger(CustomLogger):
|
|
pass
|
|
|
|
def user_code_function(*args, **kwargs):
|
|
pass
|
|
|
|
async def build_aws_loggers() -> tuple[S3Logger, SQSLogger]:
|
|
return S3Logger(s3_bucket_name="inventory-bucket"), SQSLogger(sqs_queue_url="https://sqs.example/inventory")
|
|
|
|
s3_logger, sqs_logger = asyncio.run(build_aws_loggers())
|
|
router = Router(model_list=[])
|
|
monkeypatch.setattr(litellm, "input_callback", [])
|
|
monkeypatch.setattr(
|
|
litellm, "success_callback", [LangsmithLogger(), s3_logger, router.sync_deployment_callback_on_success]
|
|
)
|
|
monkeypatch.setattr(litellm, "_async_success_callback", [sqs_logger, router.deployment_callback_on_success])
|
|
monkeypatch.setattr(litellm, "failure_callback", [user_code_function])
|
|
monkeypatch.setattr(litellm, "_async_failure_callback", [router.async_deployment_callback_on_failure])
|
|
monkeypatch.setattr(
|
|
litellm,
|
|
"callbacks",
|
|
[
|
|
_PROXY_CacheControlCheck(),
|
|
_PROXY_LiteLLMManagedFiles(internal_usage_cache=MagicMock(), prisma_client=MagicMock()),
|
|
ServiceLogging(),
|
|
VectorStorePreCallHook(),
|
|
_InventoryTestGuardrail(guardrail_name="inventory-test-guardrail"),
|
|
_UserCodeLogger(),
|
|
],
|
|
)
|
|
monkeypatch.setattr(litellm, "cache", litellm.Cache(type="local"))
|
|
assert "cache" in litellm.success_callback and "cache" in litellm._async_success_callback
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
response = client.get("/get/config/callbacks")
|
|
|
|
assert response.status_code == 200
|
|
assert [
|
|
(callback["name"], callback["type"], callback["read_only"]) for callback in response.json()["callbacks"]
|
|
] == [
|
|
("_UserCodeLogger", "success_and_failure", True),
|
|
("langsmith", "success", True),
|
|
("s3", "success", True),
|
|
("sqs", "success", True),
|
|
("user_code_function", "failure", True),
|
|
]
|
|
|
|
|
|
def test_get_config_callbacks_redacts_runtime_only_row_secrets_for_view_only_admin(
|
|
client, auth_as, mock_prisma, monkeypatch
|
|
):
|
|
"""Runtime-only callback rows are subject to the same redaction gate as configured."""
|
|
from litellm.proxy import proxy_server as ps
|
|
from litellm.proxy._types import LitellmUserRoles
|
|
|
|
_install_litellm_config(mock_prisma)
|
|
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
|
|
monkeypatch.setattr(ps, "llm_router", None)
|
|
|
|
fake_proxy_config = MagicMock()
|
|
fake_proxy_config.get_config = AsyncMock(
|
|
return_value={
|
|
"litellm_settings": {"success_callback": []},
|
|
"general_settings": {},
|
|
"environment_variables": dict(_CALLBACK_ENV_FIXTURE),
|
|
}
|
|
)
|
|
monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
|
|
|
|
import litellm
|
|
|
|
monkeypatch.setattr(litellm, "success_callback", [])
|
|
monkeypatch.setattr(litellm, "_async_success_callback", [])
|
|
monkeypatch.setattr(litellm, "failure_callback", [])
|
|
monkeypatch.setattr(litellm, "_async_failure_callback", [])
|
|
monkeypatch.setattr(litellm, "callbacks", ["otel"])
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY):
|
|
response = client.get("/get/config/callbacks")
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
|
|
callbacks = body["callbacks"]
|
|
otel_cb = next((cb for cb in callbacks if cb["name"] == "otel"), None)
|
|
assert otel_cb is not None
|
|
assert otel_cb["type"] == "success_and_failure"
|
|
assert otel_cb["read_only"] is True
|
|
assert otel_cb["variables"]["OTEL_HEADERS"] == "REDACTED"
|
|
assert otel_cb["variables"]["OTEL_ENDPOINT"] == _CALLBACK_ENV_FIXTURE["OTEL_ENDPOINT"]
|
|
|
|
with auth_as(LitellmUserRoles.PROXY_ADMIN):
|
|
admin_response = client.get("/get/config/callbacks")
|
|
assert admin_response.status_code == 200
|
|
admin_body = admin_response.json()
|
|
admin_otel = next((cb for cb in admin_body["callbacks"] if cb["name"] == "otel"), None)
|
|
assert admin_otel is not None
|
|
assert admin_otel["variables"]["OTEL_HEADERS"] == _CALLBACK_ENV_FIXTURE["OTEL_HEADERS"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /config/yaml
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_config_yaml_returns_demo_payload(client, auth_as):
|
|
"""GET /config/yaml is documented as a mock endpoint. It declares
|
|
ConfigYAML as the body parameter, so a GET with an empty JSON body is
|
|
accepted and returns the canonical demo dict."""
|
|
with auth_as():
|
|
response = client.request("GET", "/config/yaml", json={})
|
|
shape = {
|
|
"status": response.status_code,
|
|
"media_type_yaml": response.headers.get("content-type", "").startswith("application/json"),
|
|
"has_body": len(response.content) > 0,
|
|
}
|
|
assert shape == {
|
|
"status": 200,
|
|
"media_type_yaml": True,
|
|
"has_body": True,
|
|
}
|
|
assert response.json() == {"hello": "world"}
|
|
|
|
|
|
def test_config_yaml_invalid_method(client):
|
|
"""POST against the GET-only /config/yaml is rejected (error path)."""
|
|
response = client.post("/config/yaml", json={})
|
|
assert response.status_code == 405
|
|
# Method-not-allowed responses still return a JSON-ish body via the
|
|
# FastAPI default handler — assert the body is not the success payload.
|
|
assert response.content != b'{"hello":"world"}'
|