mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-14 04:09:09 +00:00
* 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>
127 lines
4 KiB
Python
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"
|
|
}
|