litellm/tests/unit/proxy/management_endpoints/test_mcp_servers.py
devin-ai-integration[bot] 1dbde0845e
test: migrate SDK and proxy-booting legacy CircleCI tests to their tiers (#45758)
* test: move guardrails, image, audio, batches, search legacy tests to unit

* test: restore the custom guardrail and remaining chunk 1 unit rows

* test: move the ocr matrix to e2e

* test: drop comments from moved tests and fix vertex ocr credentials

* test: move litellm utils and unified google legacy tests to their tiers

* test: move logging callback and pass-through unit legacy tests to their tiers

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test: move store_model_in_db legacy tests and start the admin ui migration

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test: finish the proxy job rows and drop their CircleCI jobs

Moves the remaining multi-instance, openai endpoint, otel, pass-through and
top-level proxy tests into tests/integration with scripted upstreams, rewrites
the weak replacements to assert the outbound wire and exact responses, and
removes the build_and_test job group this PR owns from CircleCI

* test: tighten the key generation and error log assertions, rename migrated unit files to their mapped names

* test: shard the pagerduty unit tests and record the rust harness file on the coverage allowlist

* test: drive the SSO sign-in tests through a scripted IdP and isolate the cost center stand-in

* test: type the e2e beta header, prompt cache and secret manager values

* test: assert the missing model error message on the OpenAI routes

* test: drop the deleted otel base class import and restore mcp redaction and exact langfuse usage checks

* test: make the langfuse fixture directory a package for the layout check

* test: restore collected no-match cases, real langfuse and gcs payload checks, and gate the aws secret lane

* test: keep the ocr e2e pdf fixture next to the test

* test: put meta below covers on the streaming tool search e2e

* test: run the OCR matrix against a scripted upstream and fix two runner-only e2e failures

* test: count only OCR calls in the OCR matrix recorder

* test: pin OCR parse bodies and billed cost, group new e2e cases into classes

* ci: restore the python image anchor dropped in the main sync

---------

Co-authored-by: yuneng <yuneng@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-10-11 00:17:21 -07:00

127 lines
4 KiB
Python

from datetime import datetime
from typing import Final
import pytest
from fastapi import HTTPException
from litellm.proxy._types import (
LiteLLM_MCPServerTable,
MCPTransport,
NewMCPServerRequest,
)
from litellm.proxy._experimental.mcp_server.db import McpIdentifierConflict
from litellm.proxy._experimental.mcp_server.utils import validate_mcp_server_name
from litellm.proxy.management_endpoints.mcp_management_endpoints import (
_redact_mcp_credentials,
does_mcp_server_exist,
raise_mcp_identifier_conflict,
validate_and_normalize_mcp_server_payload,
)
def test_does_mcp_server_exist() -> None:
created_at: Final = datetime(2025, 1, 1)
records: Final = (
LiteLLM_MCPServerTable(
server_id="server-1",
alias="first",
url="https://first.example.com/mcp",
transport=MCPTransport.sse,
created_at=created_at,
updated_at=created_at,
),
LiteLLM_MCPServerTable(
server_id="server-2",
alias="second",
url="https://second.example.com/mcp",
transport=MCPTransport.http,
created_at=created_at,
updated_at=created_at,
),
)
assert does_mcp_server_exist(records, "server-1")
assert does_mcp_server_exist(records, "server-2")
assert not does_mcp_server_exist(records, "missing-server")
def test_create_duplicate_mcp_server_returns_conflict_error() -> None:
conflict: Final = McpIdentifierConflict(
field="alias",
value="duplicate-server",
server_id="existing-server",
)
with pytest.raises(HTTPException) as error:
raise_mcp_identifier_conflict(conflict)
assert error.value.status_code == 400
assert error.value.detail == {
"error": "An MCP server with alias 'duplicate-server' already exists "
"(server_id=existing-server). MCP server names and aliases must be unique, case-insensitive."
}
def test_create_mcp_server_direct_normalizes_alias() -> None:
payload: Final = NewMCPServerRequest(
server_id="server-1",
alias="Test Server",
url="https://server.example.com/mcp",
transport=MCPTransport.sse,
)
validate_and_normalize_mcp_server_payload(payload)
assert payload.alias == "Test_Server"
assert payload.url == "https://server.example.com/mcp"
assert payload.transport == MCPTransport.sse
def test_edit_mcp_server_redacts_credentials() -> None:
created_at: Final = datetime(2025, 1, 1)
server: Final = LiteLLM_MCPServerTable(
server_id="server-1",
alias="server",
url="https://server.example.com/mcp",
transport=MCPTransport.http,
created_at=created_at,
updated_at=created_at,
credentials={"auth_value": "secret-value"},
)
redacted: Final = _redact_mcp_credentials(server)
assert redacted.credentials is None
assert server.credentials == {"auth_value": "secret-value"}
def test_create_mcp_server_invalid_alias() -> None:
payload: Final = NewMCPServerRequest(
alias="invalid-alias",
url="https://server.example.com/mcp",
transport=MCPTransport.sse,
)
with pytest.raises(HTTPException) as error:
validate_and_normalize_mcp_server_payload(payload)
assert error.value.status_code == 400
assert error.value.detail == {
"error": "Server name cannot contain '-'. Use an alternative character instead Found: invalid-alias"
}
def test_validate_mcp_server_name_direct() -> None:
validate_mcp_server_name("valid_name")
validate_mcp_server_name("valid name")
with pytest.raises(Exception, match="Server name cannot contain"):
validate_mcp_server_name("invalid-name")
with pytest.raises(HTTPException) as error:
validate_mcp_server_name("invalid-name", raise_http_exception=True)
assert error.value.status_code == 400
assert error.value.detail == {
"error": "Server name cannot contain '-'. Use an alternative character instead Found: invalid-name"
}