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

test(mcp): assert the granted plain server's exchange through what it reports, not through the mock

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng 2026-10-02 18:13:32 +00:00
parent ee7c8a20d0
commit d2e0d4e9e1
1 changed files with 13 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
tests/unit/proxy/_experimental/mcp_server/test_mcp_server_tool_calls_and_headers.py
View file

@ -8986,7 +8986,7 @@ class TestConnectPreflightRoutesLikeTheScopedRouter:
assert guardrail.asked_about == ["d-id"]
@staticmethod
def _register_obo_alias_collision() -> tuple[MCPServer, MCPServer]:
def _register_obo_alias_collision() -> None:
obo: Final = MCPServer(
server_id="o-id",
name="obo_server",
@ -9008,7 +9008,6 @@ class TestConnectPreflightRoutesLikeTheScopedRouter:
auth_type=MCPAuth.none,
)
mcp_operations.global_mcp_server_manager.registry.update({"o-id": obo, "p-id": plain})
return obo, plain
@staticmethod
async def _connect_to(route_name: str) -> None:
@ -9027,20 +9026,26 @@ class TestConnectPreflightRoutesLikeTheScopedRouter:
@pytest.mark.asyncio
@pytest.mark.parametrize("route_name", ["obo", "OBO"], ids=["exact_alias", "case_variant"])
async def test_granted_plain_server_connects_past_an_ungranted_obo_alias_holder(self, route_name):
_, plain = self._register_obo_alias_collision()
preflight: Final = AsyncMock()
self._register_obo_alias_collision()
async def report_exchange(server, **kwargs):
raise HTTPException(
status_code=401, detail=f"exchange ran for {server.server_id} as {kwargs['connected_as']}"
)
with (
patch.object( # test-quality-ok: the key's grant list lives in the DB; the real router runs on it
mcp_operations.global_mcp_server_manager, "get_allowed_mcp_servers", AsyncMock(return_value=["p-id"])
),
patch.object( # test-quality-ok: a real exchanger would call an IdP; this one records which server ran
mcp_operations.global_mcp_server_manager, "preflight_token_exchange", preflight
patch.object( # test-quality-ok: a real exchanger would call an IdP; this one reports which server it ran for
mcp_operations.global_mcp_server_manager, "preflight_token_exchange", report_exchange
),
pytest.raises(HTTPException) as exc,
):
await self._connect_to(route_name)
assert preflight.await_args is not None, "the granted plain server must reach the preflight, not a 401"
assert preflight.await_args.kwargs["server"] is plain
assert exc.value.status_code == 401
assert exc.value.detail == f"exchange ran for p-id as {route_name}"
@pytest.mark.asyncio
@pytest.mark.parametrize("route_name", ["obo", "obo_server"], ids=["alias", "server_name"])