scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-09 03:18:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 108

test: restore global MCP server manager after access-group test

Browse source
This commit is contained in:
Yuta Saito 2026-01-19 12:29:37 +09:00
parent 30c4a38179
commit 1fbbe0a983
1 changed files with 28 additions and 23 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

51
tests/mcp_tests/test_mcp_server.py
View file

@ -1001,34 +1001,39 @@ async def test_mcp_server_manager_access_groups_from_config():
MCPRequestHandler,
)
# Patch global_mcp_server_manager for this test
# Patch global_mcp_server_manager for this test and restore afterwards to
# avoid leaking state into other tests (e.g. the proxy MCP e2e suite).
import litellm.proxy._experimental.mcp_server.mcp_server_manager as mcp_server_manager_mod
original_manager = mcp_server_manager_mod.global_mcp_server_manager
mcp_server_manager_mod.global_mcp_server_manager = test_manager
# Should find config_server for group-a, both for group-b, other_server for group-c
import asyncio
try:
# Should find config_server for group-a, both for group-b, other_server for group-c
import asyncio
server_ids_a = await MCPRequestHandler._get_mcp_servers_from_access_groups([
"group-a"
])
server_ids_b = await MCPRequestHandler._get_mcp_servers_from_access_groups([
"group-b"
])
server_ids_c = await MCPRequestHandler._get_mcp_servers_from_access_groups([
"group-c"
])
assert any(config_server.server_id == sid for sid in server_ids_a)
assert set(server_ids_b) == set(
[
s.server_id
server_ids_a = await MCPRequestHandler._get_mcp_servers_from_access_groups([
"group-a"
])
server_ids_b = await MCPRequestHandler._get_mcp_servers_from_access_groups([
"group-b"
])
server_ids_c = await MCPRequestHandler._get_mcp_servers_from_access_groups([
"group-c"
])
assert any(config_server.server_id == sid for sid in server_ids_a)
assert set(server_ids_b) == set(
[
s.server_id
for s in test_manager.config_mcp_servers.values()
if "group-b" in s.access_groups
]
)
assert any(
s.name == "other_server" and s.server_id in server_ids_c
for s in test_manager.config_mcp_servers.values()
if "group-b" in s.access_groups
]
)
assert any(
s.name == "other_server" and s.server_id in server_ids_c
for s in test_manager.config_mcp_servers.values()
)
)
finally:
mcp_server_manager_mod.global_mcp_server_manager = original_manager
async def test_mcp_server_manager_config_integration_with_database():