100 lines
3.1 KiB
Python
100 lines
3.1 KiB
Python
"""Tests for MCP server."""
|
|
|
|
import pytest
|
|
|
|
from mcp_server_cli.models import MCPMethod, MCPRequest
|
|
from mcp_server_cli.server import MCPServer
|
|
from mcp_server_cli.tools import FileTools, GitTools, ShellTools
|
|
|
|
|
|
@pytest.fixture
|
|
def mcp_server():
|
|
"""Create an MCP server with registered tools."""
|
|
server = MCPServer()
|
|
server.register_tool(FileTools())
|
|
server.register_tool(GitTools())
|
|
server.register_tool(ShellTools())
|
|
return server
|
|
|
|
|
|
class TestMCPServer:
|
|
"""Tests for MCP server class."""
|
|
|
|
def test_server_creation(self):
|
|
"""Test server creation."""
|
|
server = MCPServer()
|
|
assert server.connection_state.value == "disconnected"
|
|
assert len(server.tool_registry) == 0
|
|
|
|
def test_register_tool(self, mcp_server):
|
|
"""Test tool registration."""
|
|
assert "file_tools" in mcp_server.tool_registry
|
|
assert len(mcp_server.list_tools()) == 3
|
|
|
|
def test_get_tool(self, mcp_server):
|
|
"""Test getting a registered tool."""
|
|
retrieved = mcp_server.get_tool("file_tools")
|
|
assert retrieved is not None
|
|
assert retrieved.name == "file_tools"
|
|
|
|
def test_list_tools(self, mcp_server):
|
|
"""Test listing all tools."""
|
|
tools = mcp_server.list_tools()
|
|
assert len(tools) >= 2
|
|
names = [t.name for t in tools]
|
|
assert "file_tools" in names
|
|
assert "git_tools" in names
|
|
|
|
|
|
class TestMCPProtocol:
|
|
"""Tests for MCP protocol implementation."""
|
|
|
|
def test_mcp_initialize(self, mcp_server):
|
|
"""Test MCP initialize request."""
|
|
request = MCPRequest(
|
|
id=1,
|
|
method=MCPMethod.INITIALIZE,
|
|
params={"protocol_version": "2024-11-05"},
|
|
)
|
|
import asyncio
|
|
response = asyncio.run(mcp_server.handle_request(request))
|
|
assert response.id == 1
|
|
assert response.result is not None
|
|
|
|
def test_mcp_tools_list(self, mcp_server):
|
|
"""Test MCP tools/list request."""
|
|
request = MCPRequest(
|
|
id=2,
|
|
method=MCPMethod.TOOLS_LIST,
|
|
)
|
|
import asyncio
|
|
response = asyncio.run(mcp_server.handle_request(request))
|
|
assert response.id == 2
|
|
assert response.result is not None
|
|
|
|
def test_mcp_invalid_method(self, mcp_server):
|
|
"""Test MCP request with invalid tool."""
|
|
request = MCPRequest(
|
|
id=3,
|
|
method=MCPMethod.TOOLS_CALL,
|
|
params={"name": "nonexistent"},
|
|
)
|
|
import asyncio
|
|
response = asyncio.run(mcp_server.handle_request(request))
|
|
assert response.error is not None or response.result.get("is_error") is True
|
|
|
|
|
|
class TestToolCall:
|
|
"""Tests for tool calling."""
|
|
|
|
def test_call_read_file_nonexistent(self, mcp_server):
|
|
"""Test calling read on nonexistent file."""
|
|
from mcp_server_cli.models import ToolCallParams
|
|
params = ToolCallParams(
|
|
name="read_file",
|
|
arguments={"path": "/nonexistent/file.txt"},
|
|
)
|
|
import asyncio
|
|
result = asyncio.run(mcp_server._handle_tool_call(params))
|
|
assert result.is_error is True
|