mirror of
https://github.com/tiennm99/litellm.git
synced 2026-08-05 14:23:44 +00:00
test: remove unnecessary test
This commit is contained in:
@@ -745,84 +745,3 @@ async def test_call_mcp_tool_user_unauthorized_access():
|
||||
# Verify the exception details
|
||||
assert exc_info.value.status_code == 403
|
||||
assert "User not allowed to call this tool" in exc_info.value.detail
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_call_mcp_tool_user_authorized_access():
|
||||
"""Test that a user can call a tool from a server they have access to"""
|
||||
from mcp.types import CallToolResult, TextContent
|
||||
|
||||
from litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp import (
|
||||
MCPRequestHandler,
|
||||
)
|
||||
from litellm.proxy._experimental.mcp_server.server import call_mcp_tool
|
||||
from litellm.proxy._types import UserAPIKeyAuth
|
||||
|
||||
# Create a mock user with access to the server
|
||||
mock_user_auth = UserAPIKeyAuth(
|
||||
api_key="test-key",
|
||||
user_id="test-user",
|
||||
team_id="team-admin",
|
||||
object_permission_id="key-permission-456",
|
||||
)
|
||||
|
||||
# Mock successful tool call result
|
||||
mock_result = CallToolResult(
|
||||
content=[TextContent(type="text", text="Email sent successfully")],
|
||||
isError=False,
|
||||
)
|
||||
|
||||
# Mock the database calls that determine access permissions
|
||||
# Mock get_object_permission to return access to the allowed server
|
||||
with patch(
|
||||
"litellm.proxy.auth.auth_checks.get_object_permission"
|
||||
) as mock_get_object_permission:
|
||||
# Mock get_team_object to return team with MCP access
|
||||
with patch(
|
||||
"litellm.proxy.auth.auth_checks.get_team_object"
|
||||
) as mock_get_team_object:
|
||||
# Mock object permission - key has access to allowed_server
|
||||
mock_key_permission = MagicMock()
|
||||
mock_key_permission.mcp_servers = ["allowed_server"] # Direct server access
|
||||
mock_key_permission.mcp_access_groups = ["admin_group"] # Access groups
|
||||
mock_get_object_permission.return_value = mock_key_permission
|
||||
|
||||
# Mock team object - team has MCP access
|
||||
mock_team = MagicMock()
|
||||
mock_team_permission = MagicMock()
|
||||
mock_team_permission.mcp_servers = ["allowed_server", "team_server"]
|
||||
mock_team_permission.mcp_access_groups = ["admin_group", "team_group"]
|
||||
mock_team.object_permission = mock_team_permission
|
||||
mock_get_team_object.return_value = mock_team
|
||||
|
||||
# Mock _get_mcp_servers_from_access_groups to return servers from access groups
|
||||
with patch.object(
|
||||
MCPRequestHandler, "_get_mcp_servers_from_access_groups"
|
||||
) as mock_get_servers_from_groups:
|
||||
mock_get_servers_from_groups.return_value = ["allowed_server"]
|
||||
|
||||
# Mock global_mcp_server_manager.call_tool to return successful result
|
||||
with patch(
|
||||
"litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager"
|
||||
) as mock_manager:
|
||||
mock_manager.call_tool = AsyncMock(return_value=mock_result)
|
||||
|
||||
# Call the tool - should succeed
|
||||
result = await call_mcp_tool(
|
||||
name="allowed_server-send_email",
|
||||
arguments={
|
||||
"to": "test@example.com",
|
||||
"subject": "Test",
|
||||
"body": "Test",
|
||||
},
|
||||
user_api_key_auth=mock_user_auth,
|
||||
mcp_auth_header="Bearer test_token",
|
||||
)
|
||||
|
||||
# Verify the result
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], TextContent)
|
||||
assert result[0].text == "Email sent successfully"
|
||||
|
||||
# Verify that the manager's call_tool was called (meaning authorization passed)
|
||||
mock_manager.call_tool.assert_called_once()
|
||||
|
||||
Reference in New Issue
Block a user