141 lines
3.8 KiB
Python
141 lines
3.8 KiB
Python
"""Tests for MCPToolConfig schema with HTTP/SSE transport."""
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
from mcp_forge.config.schema import MCPToolConfig
|
|
|
|
|
|
def test_stdio_transport_config_valid():
|
|
"""Test valid stdio transport configuration."""
|
|
config = MCPToolConfig(
|
|
transport="stdio",
|
|
command="python",
|
|
args=["-m", "server"],
|
|
env={"KEY": "value"}
|
|
)
|
|
|
|
assert config.transport == "stdio"
|
|
assert config.command == "python"
|
|
assert config.args == ["-m", "server"]
|
|
assert config.env == {"KEY": "value"}
|
|
|
|
|
|
def test_http_transport_config_valid():
|
|
"""Test valid HTTP transport configuration."""
|
|
config = MCPToolConfig(
|
|
transport="http",
|
|
url="http://localhost:8006/mcp",
|
|
headers={"Authorization": "Bearer token"}
|
|
)
|
|
|
|
assert config.transport == "http"
|
|
assert config.url == "http://localhost:8006/mcp"
|
|
assert config.headers == {"Authorization": "Bearer token"}
|
|
|
|
|
|
def test_sse_transport_config_valid():
|
|
"""Test valid SSE transport configuration."""
|
|
config = MCPToolConfig(
|
|
transport="sse",
|
|
url="http://localhost:9000/events",
|
|
headers={"X-Custom": "value"}
|
|
)
|
|
|
|
assert config.transport == "sse"
|
|
assert config.url == "http://localhost:9000/events"
|
|
assert config.headers == {"X-Custom": "value"}
|
|
|
|
|
|
def test_stdio_without_command_invalid():
|
|
"""Test that stdio transport requires command."""
|
|
with pytest.raises(ValidationError, match="command is required for stdio transport"):
|
|
MCPToolConfig(
|
|
transport="stdio",
|
|
args=["-m", "server"]
|
|
)
|
|
|
|
|
|
def test_http_without_url_invalid():
|
|
"""Test that HTTP transport requires URL."""
|
|
with pytest.raises(ValidationError, match="url is required for http transport"):
|
|
MCPToolConfig(
|
|
transport="http",
|
|
headers={"Authorization": "Bearer token"}
|
|
)
|
|
|
|
|
|
def test_sse_without_url_invalid():
|
|
"""Test that SSE transport requires URL."""
|
|
with pytest.raises(ValidationError, match="url is required for sse transport"):
|
|
MCPToolConfig(
|
|
transport="sse",
|
|
headers={"X-Custom": "value"}
|
|
)
|
|
|
|
|
|
def test_default_transport_is_stdio():
|
|
"""Test that default transport is stdio."""
|
|
config = MCPToolConfig(
|
|
command="python",
|
|
args=["-m", "server"]
|
|
)
|
|
|
|
assert config.transport == "stdio"
|
|
|
|
|
|
def test_http_config_with_empty_headers():
|
|
"""Test HTTP config with no headers."""
|
|
config = MCPToolConfig(
|
|
transport="http",
|
|
url="http://localhost:8006/mcp"
|
|
)
|
|
|
|
assert config.headers == {}
|
|
|
|
|
|
def test_stdio_config_with_empty_env():
|
|
"""Test stdio config with no env vars."""
|
|
config = MCPToolConfig(
|
|
transport="stdio",
|
|
command="python"
|
|
)
|
|
|
|
assert config.env == {}
|
|
assert config.args == []
|
|
|
|
|
|
def test_invalid_transport_type():
|
|
"""Test that invalid transport type is rejected."""
|
|
with pytest.raises(ValidationError):
|
|
MCPToolConfig(
|
|
transport="invalid",
|
|
command="python"
|
|
)
|
|
|
|
|
|
def test_stdio_config_minimal():
|
|
"""Test minimal stdio config with just command."""
|
|
config = MCPToolConfig(
|
|
command="python"
|
|
)
|
|
|
|
assert config.transport == "stdio"
|
|
assert config.command == "python"
|
|
assert config.args == []
|
|
assert config.env == {}
|
|
|
|
|
|
def test_http_config_with_multiple_headers():
|
|
"""Test HTTP config with multiple headers."""
|
|
config = MCPToolConfig(
|
|
transport="http",
|
|
url="http://localhost:8006/mcp",
|
|
headers={
|
|
"Authorization": "Bearer token123",
|
|
"X-Custom-Header": "value",
|
|
"Content-Type": "application/json"
|
|
}
|
|
)
|
|
|
|
assert len(config.headers) == 3
|
|
assert config.headers["Authorization"] == "Bearer token123"
|