"""Tests for MCP server setup and configuration.
Requirements covered:
- IN-001: Implement the Model Context Protocol (MCP) specification
- NF-001: Stateless server (no server-side session storage)
- NF-002: Delegate data operations to Jana backend
- 4.1: Tool inventory
"""
from unittest.mock import MagicMock, patch
import pytest
from jana_mcp.config import Settings
class TestMCPServerCreation:
"""Test MCP server creation and initialization."""
def test_create_mcp_server_returns_tuple(self):
"""Test create_mcp_server returns server and client tuple."""
from jana_mcp.server import create_mcp_server
with patch("jana_mcp.server._register_all_tools"):
server, client = create_mcp_server()
assert server is not None
assert client is not None
def test_create_mcp_server_with_settings(self):
"""Test create_mcp_server accepts custom settings."""
from jana_mcp.server import create_mcp_server
settings = Settings(
jana_backend_url="http://custom:8000",
jana_username="test",
)
with patch("jana_mcp.server._register_all_tools"):
server, client = create_mcp_server(settings)
assert client.settings.jana_backend_url == "http://custom:8000"
def test_create_mcp_server_name(self):
"""Test MCP server has correct name."""
from jana_mcp.server import create_mcp_server
with patch("jana_mcp.server._register_all_tools"):
server, client = create_mcp_server()
assert server.name == "jana-mcp-server"
def test_create_mcp_server_with_client(self):
"""Test create_mcp_server_with_client creates server with provided client."""
from jana_mcp.client import JanaClient
from jana_mcp.server import create_mcp_server_with_client
# Create a mock client
mock_client = MagicMock(spec=JanaClient)
with patch("jana_mcp.server._register_all_tools") as mock_register:
server = create_mcp_server_with_client(mock_client)
assert server is not None
assert server.name == "jana-mcp-server"
# Verify the client was passed to tool registration
mock_register.assert_called_once()
call_args = mock_register.call_args[0]
assert call_args[1] is mock_client
def test_create_mcp_server_with_client_for_per_user_auth(self):
"""Test creating server with user-specific client (per-user auth)."""
from pydantic import SecretStr
from jana_mcp.client import JanaClient
from jana_mcp.config import Settings
from jana_mcp.server import create_mcp_server_with_client
# Simulate per-user client with user's token
user_settings = Settings(
jana_backend_url="http://backend:8000",
jana_token=SecretStr("user_specific_token_abc123"),
)
user_client = JanaClient(user_settings)
server = create_mcp_server_with_client(user_client)
assert server is not None
assert server.name == "jana-mcp-server"
class TestToolRegistration:
"""Test tool registration with MCP server."""
def test_all_p0_tools_registered(self):
"""Test all P0 priority tools are registered (4.1)."""
from jana_mcp.server import create_mcp_server
server, client = create_mcp_server()
# Get registered tools through the handlers
# The server should have handlers for list_tools
assert hasattr(server, "_request_handlers") or hasattr(server, "list_tools")
def test_air_quality_tool_exists(self):
"""Test get_air_quality tool is available."""
from jana_mcp.tools.air_quality import AIR_QUALITY_TOOL
assert AIR_QUALITY_TOOL.name == "get_air_quality"
def test_emissions_tool_exists(self):
"""Test get_emissions tool is available."""
from jana_mcp.tools.emissions import EMISSIONS_TOOL
assert EMISSIONS_TOOL.name == "get_emissions"
def test_nearby_tool_exists(self):
"""Test find_nearby tool is available."""
from jana_mcp.tools.nearby import NEARBY_TOOL
assert NEARBY_TOOL.name == "find_nearby"
def test_trends_tool_exists(self):
"""Test get_trends tool is available."""
from jana_mcp.tools.trends import TRENDS_TOOL
assert TRENDS_TOOL.name == "get_trends"
def test_data_summary_tool_exists(self):
"""Test get_data_summary tool is available."""
from jana_mcp.tools.system import DATA_SUMMARY_TOOL
assert DATA_SUMMARY_TOOL.name == "get_data_summary"
def test_system_health_tool_exists(self):
"""Test get_system_health tool is available."""
from jana_mcp.tools.system import SYSTEM_HEALTH_TOOL
assert SYSTEM_HEALTH_TOOL.name == "get_system_health"
class TestToolInventory:
"""Test complete tool inventory matches requirements (4.1)."""
def test_p0_tools_count(self):
"""Test P0 tools are available (get_air_quality, get_emissions, find_nearby)."""
from jana_mcp.tools.air_quality import AIR_QUALITY_TOOL
from jana_mcp.tools.emissions import EMISSIONS_TOOL
from jana_mcp.tools.nearby import NEARBY_TOOL
p0_tools = [AIR_QUALITY_TOOL, EMISSIONS_TOOL, NEARBY_TOOL]
assert len(p0_tools) == 3
def test_p1_tools_count(self):
"""Test P1 tools are available (get_trends, get_data_summary)."""
from jana_mcp.tools.system import DATA_SUMMARY_TOOL
from jana_mcp.tools.trends import TRENDS_TOOL
p1_tools = [TRENDS_TOOL, DATA_SUMMARY_TOOL]
assert len(p1_tools) == 2
def test_p2_tools_count(self):
"""Test P2 tools are available (get_system_health)."""
from jana_mcp.tools.system import SYSTEM_HEALTH_TOOL
p2_tools = [SYSTEM_HEALTH_TOOL]
assert len(p2_tools) == 1
def test_total_tools_count(self):
"""Test total tool count matches specification."""
from jana_mcp.tools.air_quality import AIR_QUALITY_TOOL
from jana_mcp.tools.emissions import EMISSIONS_TOOL
from jana_mcp.tools.nearby import NEARBY_TOOL
from jana_mcp.tools.system import DATA_SUMMARY_TOOL, SYSTEM_HEALTH_TOOL
from jana_mcp.tools.trends import TRENDS_TOOL
all_tools = [
AIR_QUALITY_TOOL,
EMISSIONS_TOOL,
NEARBY_TOOL,
TRENDS_TOOL,
DATA_SUMMARY_TOOL,
SYSTEM_HEALTH_TOOL,
]
# 4.1 specifies 7 tools but we have 6 (get_correlations is P1 but deferred)
assert len(all_tools) >= 6
class TestStatelessArchitecture:
"""Test stateless architecture (NF-001)."""
def test_server_no_session_storage(self):
"""Test server doesn't maintain session state."""
from jana_mcp.server import create_mcp_server
server1, client1 = create_mcp_server()
server2, client2 = create_mcp_server()
# Each call should create independent instances
assert server1 is not server2
assert client1 is not client2
def test_client_no_query_history(self):
"""Test client doesn't maintain query history."""
from jana_mcp.client import JanaClient
from jana_mcp.config import Settings
settings = Settings(jana_backend_url="http://test:8000")
client = JanaClient(settings)
# Client should not have attributes for storing history
assert not hasattr(client, "query_history")
assert not hasattr(client, "session_data")
assert not hasattr(client, "conversation_history")
class TestBackendDelegation:
"""Test backend delegation (NF-002)."""
def test_client_delegates_to_backend(self):
"""Test all operations delegate to backend."""
from jana_mcp.client import JanaClient
from jana_mcp.config import Settings
settings = Settings(jana_backend_url="http://test:8000")
client = JanaClient(settings)
# Client should have methods that call backend API
assert hasattr(client, "get_air_quality")
assert hasattr(client, "get_emissions")
assert hasattr(client, "find_nearby")
assert hasattr(client, "get_summary")
assert hasattr(client, "check_health")
def test_no_local_data_storage(self):
"""Test client doesn't store data locally."""
from jana_mcp.client import JanaClient
from jana_mcp.config import Settings
settings = Settings(jana_backend_url="http://test:8000")
client = JanaClient(settings)
# Client should not have local data cache
assert not hasattr(client, "data_cache")
assert not hasattr(client, "local_storage")
class TestToolCallRouting:
"""Test tool call routing in server."""
@pytest.mark.asyncio
async def test_unknown_tool_returns_error(self):
"""Test unknown tool name returns error message."""
from unittest.mock import AsyncMock
from jana_mcp.server import TOOL_HANDLERS, create_mcp_server
server, client = create_mcp_server()
# Mock the client methods
client.get_air_quality = AsyncMock(return_value={"results": []})
# Test that unknown tool is not in handlers
assert "unknown_tool" not in TOOL_HANDLERS
# Test that calling with unknown tool via handler returns error
# Simulate the dispatch logic
unknown_handler = TOOL_HANDLERS.get("unknown_tool")
assert unknown_handler is None
# Test that all expected tools are in handlers
assert "get_air_quality" in TOOL_HANDLERS
assert "get_emissions" in TOOL_HANDLERS
assert "find_nearby" in TOOL_HANDLERS
assert "get_trends" in TOOL_HANDLERS
assert "get_data_summary" in TOOL_HANDLERS
assert "get_system_health" in TOOL_HANDLERS
class TestToolSchema:
"""Test tool schemas are valid MCP format."""
def test_all_tools_have_input_schema(self):
"""Test all tools have inputSchema property."""
from jana_mcp.tools.air_quality import AIR_QUALITY_TOOL
from jana_mcp.tools.emissions import EMISSIONS_TOOL
from jana_mcp.tools.nearby import NEARBY_TOOL
from jana_mcp.tools.system import DATA_SUMMARY_TOOL, SYSTEM_HEALTH_TOOL
from jana_mcp.tools.trends import TRENDS_TOOL
all_tools = [
AIR_QUALITY_TOOL,
EMISSIONS_TOOL,
NEARBY_TOOL,
TRENDS_TOOL,
DATA_SUMMARY_TOOL,
SYSTEM_HEALTH_TOOL,
]
for tool in all_tools:
assert hasattr(tool, "inputSchema")
assert "type" in tool.inputSchema
assert tool.inputSchema["type"] == "object"
def test_all_tools_have_name(self):
"""Test all tools have name property."""
from jana_mcp.tools.air_quality import AIR_QUALITY_TOOL
from jana_mcp.tools.emissions import EMISSIONS_TOOL
from jana_mcp.tools.nearby import NEARBY_TOOL
from jana_mcp.tools.system import DATA_SUMMARY_TOOL, SYSTEM_HEALTH_TOOL
from jana_mcp.tools.trends import TRENDS_TOOL
all_tools = [
AIR_QUALITY_TOOL,
EMISSIONS_TOOL,
NEARBY_TOOL,
TRENDS_TOOL,
DATA_SUMMARY_TOOL,
SYSTEM_HEALTH_TOOL,
]
for tool in all_tools:
assert hasattr(tool, "name")
assert isinstance(tool.name, str)
assert len(tool.name) > 0
def test_all_tools_have_description(self):
"""Test all tools have description property."""
from jana_mcp.tools.air_quality import AIR_QUALITY_TOOL
from jana_mcp.tools.emissions import EMISSIONS_TOOL
from jana_mcp.tools.nearby import NEARBY_TOOL
from jana_mcp.tools.system import DATA_SUMMARY_TOOL, SYSTEM_HEALTH_TOOL
from jana_mcp.tools.trends import TRENDS_TOOL
all_tools = [
AIR_QUALITY_TOOL,
EMISSIONS_TOOL,
NEARBY_TOOL,
TRENDS_TOOL,
DATA_SUMMARY_TOOL,
SYSTEM_HEALTH_TOOL,
]
for tool in all_tools:
assert hasattr(tool, "description")
assert isinstance(tool.description, str)
assert len(tool.description) > 0
