test_smoke_e2e.py•3.29 kB
"""end-to-end smoke tests with real sdks."""
import pytest
import asyncio
from unittest import mock
from src.mcp_server import GeneralizedMCPServer, SDKConfig
from src.safety import SafetyConfig
@pytest.fixture
def test_server():
"""create test server with minimal sdks."""
sdks = [
SDKConfig(
name="test_sdk",
module="os", # use stdlib for testing
allow_patterns=["*.get*", "*.list*"],
allow_mutating=False
)
]
return GeneralizedMCPServer(
sdks=sdks,
use_llm_schemas=False, # disable for testing
safety_config=SafetyConfig(allow_mutating=False, dry_run=False)
)
def test_server_loads_tools(test_server):
"""test that server loads tools from sdk."""
assert len(test_server.tools) > 0
def test_tools_list(test_server):
"""test tools/list handler."""
response = test_server.handle_tools_list({})
assert "tools" in response
assert len(response["tools"]) > 0
# check structure of first tool
tool = response["tools"][0]
assert "name" in tool
assert "description" in tool
assert "inputSchema" in tool
@pytest.mark.asyncio
async def test_tools_call_success(test_server):
"""test successful tools/call."""
# find a read-only tool
tool_name = list(test_server.tools.keys())[0]
response = await test_server.handle_tools_call({
"name": tool_name,
"arguments": {}
})
assert "content" in response
@pytest.mark.asyncio
async def test_tools_call_not_found(test_server):
"""test tools/call with unknown tool."""
with pytest.raises(ValueError, match="tool not found"):
await test_server.handle_tools_call({
"name": "nonexistent.tool",
"arguments": {}
})
@pytest.mark.asyncio
async def test_safety_blocks_mutating_operations():
"""test that safety layer blocks destructive operations."""
sdks = [
SDKConfig(
name="test_sdk",
module="os",
allow_patterns=["*"],
allow_mutating=False # block mutating
)
]
server = GeneralizedMCPServer(
sdks=sdks,
use_llm_schemas=False,
safety_config=SafetyConfig(allow_mutating=False)
)
# all mutating operations should be filtered out
mutating_tools = [t for t in server.tools.values() if t.is_mutating]
assert len(mutating_tools) == 0
@pytest.mark.asyncio
async def test_dry_run_mode():
"""test that dry-run mode prevents execution."""
sdks = [
SDKConfig(
name="test_sdk",
module="os",
allow_patterns=["*"],
allow_mutating=True # allow for this test
)
]
server = GeneralizedMCPServer(
sdks=sdks,
use_llm_schemas=False,
safety_config=SafetyConfig(allow_mutating=True, dry_run=True)
)
# find a mutating tool
mutating_tool = next((t for t in server.tools.values() if t.is_mutating), None)
if mutating_tool:
response = await server.handle_tools_call({
"name": mutating_tool.fq_name,
"arguments": {}
})
# response should indicate dry-run
assert "dry_run" in str(response)