mcp-forge/tests/config/test_mcp_tool_config.py

"""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"