"""Tests for system and administrative MCP tools.
Requirements covered:
- AD-001: Provide platform summary information (data coverage, record counts)
- AD-002: Provide system health status
- AN-003: Provide data quality insights (freshness, coverage)
- 4.2: Tool parameters specification for get_data_summary and get_system_health
"""
import json
from unittest.mock import AsyncMock, MagicMock
import httpx
import pytest
from jana_mcp.client import APIError, AuthenticationError
from jana_mcp.tools.system import DATA_SUMMARY_TOOL, SYSTEM_HEALTH_TOOL
class TestDataSummaryToolDefinition:
"""Test data summary tool definition and schema."""
def test_tool_name(self):
"""Test tool has correct name."""
assert DATA_SUMMARY_TOOL.name == "get_data_summary"
def test_tool_has_description(self):
"""Test tool has description."""
assert DATA_SUMMARY_TOOL.description is not None
assert len(DATA_SUMMARY_TOOL.description) > 0
assert "summary" in DATA_SUMMARY_TOOL.description.lower()
def test_tool_mentions_coverage(self):
"""Test tool description mentions data coverage (AD-001)."""
assert "coverage" in DATA_SUMMARY_TOOL.description.lower()
def test_input_schema_no_required_params(self):
"""Test tool has no required parameters."""
props = DATA_SUMMARY_TOOL.inputSchema.get("properties", {})
required = DATA_SUMMARY_TOOL.inputSchema.get("required", [])
assert required == []
class TestSystemHealthToolDefinition:
"""Test system health tool definition and schema."""
def test_tool_name(self):
"""Test tool has correct name."""
assert SYSTEM_HEALTH_TOOL.name == "get_system_health"
def test_tool_has_description(self):
"""Test tool has description."""
assert SYSTEM_HEALTH_TOOL.description is not None
assert len(SYSTEM_HEALTH_TOOL.description) > 0
assert "health" in SYSTEM_HEALTH_TOOL.description.lower()
def test_tool_mentions_availability(self):
"""Test tool description mentions availability (AD-002)."""
description = SYSTEM_HEALTH_TOOL.description.lower()
assert "availability" in description or "status" in description
def test_input_schema_no_required_params(self):
"""Test tool has no required parameters."""
props = SYSTEM_HEALTH_TOOL.inputSchema.get("properties", {})
required = SYSTEM_HEALTH_TOOL.inputSchema.get("required", [])
assert required == []
class TestDataSummaryExecution:
"""Test data summary tool execution."""
@pytest.mark.asyncio
async def test_execute_returns_summary(self, mock_client):
"""Test execution returns platform summary (AD-001)."""
from jana_mcp.tools.system import execute_data_summary
mock_client.get_summary.return_value = {
"total_records": 1000000,
"sources": {
"openaq": {"records": 500000, "last_updated": "2024-01-01"},
"climatetrace": {"records": 300000, "last_updated": "2024-01-01"},
"edgar": {"records": 200000, "last_updated": "2024-01-01"},
},
"geographic_coverage": {
"countries": 195,
"regions": ["North America", "Europe", "Asia"],
},
}
result = await execute_data_summary(mock_client)
mock_client.get_summary.assert_called_once()
assert len(result) == 1
assert result[0].type == "text"
data = json.loads(result[0].text)
assert "total_records" in data
@pytest.mark.asyncio
async def test_execute_handles_api_error(self, mock_client):
"""Test execution provides fallback on API error (AN-003)."""
from jana_mcp.tools.system import execute_data_summary
import httpx
mock_client.get_summary.side_effect = httpx.RequestError("API Error")
result = await execute_data_summary(mock_client)
assert len(result) == 1
data = json.loads(result[0].text)
# Should provide fallback info about available sources
assert "available_sources" in data or "status" in data
@pytest.mark.asyncio
async def test_result_format_json(self, mock_client):
"""Test result is properly formatted JSON."""
from jana_mcp.tools.system import execute_data_summary
mock_client.get_summary.return_value = {"status": "ok"}
result = await execute_data_summary(mock_client)
# Should parse as valid JSON
data = json.loads(result[0].text)
assert data is not None
class TestSystemHealthExecution:
"""Test system health tool execution."""
@pytest.mark.asyncio
async def test_execute_returns_health_status(self, mock_client):
"""Test execution returns health status (AD-002)."""
from jana_mcp.tools.system import execute_system_health
mock_client.check_health.return_value = {
"status": "healthy",
"backend": {"status": "ok", "response_time_ms": 50},
}
result = await execute_system_health(mock_client)
mock_client.check_health.assert_called_once()
assert len(result) == 1
data = json.loads(result[0].text)
assert "status" in data or "mcp_server" in data
@pytest.mark.asyncio
async def test_execute_includes_mcp_server_status(self, mock_client):
"""Test execution includes MCP server status."""
from jana_mcp.tools.system import execute_system_health
mock_client.check_health.return_value = {"status": "healthy"}
result = await execute_system_health(mock_client)
data = json.loads(result[0].text)
assert "mcp_server" in data
assert data["mcp_server"]["status"] == "healthy"
@pytest.mark.asyncio
async def test_execute_includes_version(self, mock_client):
"""Test execution includes server version."""
from jana_mcp.tools.system import execute_system_health
mock_client.check_health.return_value = {"status": "healthy"}
result = await execute_system_health(mock_client)
data = json.loads(result[0].text)
assert "version" in data["mcp_server"]
@pytest.mark.asyncio
async def test_execute_handles_backend_unreachable(self, mock_client):
"""Test execution handles unreachable backend (EH-002)."""
from jana_mcp.tools.system import execute_system_health
import httpx
mock_client.check_health.side_effect = httpx.RequestError("Connection refused")
result = await execute_system_health(mock_client)
data = json.loads(result[0].text)
# MCP server should still report healthy
assert data["mcp_server"]["status"] == "healthy"
# Backend should report unreachable
assert data["backend"]["status"] == "unreachable"
assert "error" in data["backend"]
@pytest.mark.asyncio
async def test_result_format_json(self, mock_client):
"""Test result is properly formatted JSON."""
from jana_mcp.tools.system import execute_system_health
mock_client.check_health.return_value = {"status": "ok"}
result = await execute_system_health(mock_client)
# Should parse as valid JSON
data = json.loads(result[0].text)
assert data is not None
class TestSystemToolsIntegration:
"""Integration tests for system tools."""
@pytest.mark.asyncio
async def test_summary_and_health_independent(self, mock_client):
"""Test summary and health tools work independently."""
from jana_mcp.tools.system import execute_data_summary, execute_system_health
summary_result = await execute_data_summary(mock_client)
health_result = await execute_system_health(mock_client)
# Both should return valid results
summary_data = json.loads(summary_result[0].text)
health_data = json.loads(health_result[0].text)
assert summary_data is not None
assert health_data is not None
@pytest.mark.asyncio
async def test_health_provides_backend_connectivity(self, mock_client):
"""Test health tool provides backend connectivity info (AN-003)."""
from jana_mcp.tools.system import execute_system_health
mock_client.check_health.return_value = {
"status": "healthy",
"backend": {"status": "ok"},
}
result = await execute_system_health(mock_client)
data = json.loads(result[0].text)
# Should indicate backend connectivity status
assert "backend" in data or "status" in data
class TestSystemToolsErrorPaths:
"""Test error handling paths for system tools."""
@pytest.mark.asyncio
async def test_data_summary_handles_api_error(self, mock_client):
"""Test data_summary handles APIError gracefully."""
from jana_mcp.tools.system import execute_data_summary
mock_client.get_summary.side_effect = APIError("API Error", status_code=500)
result = await execute_data_summary(mock_client)
# Returns fallback response with partial status
data = json.loads(result[0].text)
assert data.get("status") == "partial"
assert data.get("error_code") == "API_ERROR"
@pytest.mark.asyncio
async def test_data_summary_handles_authentication_error(self, mock_client):
"""Test data_summary handles AuthenticationError gracefully."""
from jana_mcp.tools.system import execute_data_summary
mock_client.get_summary.side_effect = AuthenticationError("Auth failed")
result = await execute_data_summary(mock_client)
# Returns fallback response with partial status
data = json.loads(result[0].text)
assert data.get("status") == "partial"
assert data.get("error_code") == "API_ERROR"
@pytest.mark.asyncio
async def test_data_summary_handles_network_error(self, mock_client):
"""Test data_summary handles network errors gracefully."""
from jana_mcp.tools.system import execute_data_summary
mock_client.get_summary.side_effect = httpx.RequestError("Network error")
result = await execute_data_summary(mock_client)
# Returns fallback response with partial status
data = json.loads(result[0].text)
assert data.get("status") == "partial"
assert data.get("error_code") == "NETWORK_ERROR"
@pytest.mark.asyncio
async def test_data_summary_handles_data_error(self, mock_client):
"""Test data_summary handles data parsing errors."""
from jana_mcp.tools.system import execute_data_summary
mock_client.get_summary.side_effect = KeyError("missing_key")
result = await execute_data_summary(mock_client)
# Returns error response
data = json.loads(result[0].text)
assert data.get("success") is False
assert data.get("error_code") == "DATA_ERROR"
@pytest.mark.asyncio
async def test_system_health_handles_api_error(self, mock_client):
"""Test system_health handles APIError gracefully."""
from jana_mcp.tools.system import execute_system_health
mock_client.check_health.side_effect = APIError("API Error", status_code=500)
result = await execute_system_health(mock_client)
data = json.loads(result[0].text)
# MCP should still be healthy, backend should show error
assert data["mcp_server"]["status"] == "healthy"
assert data["backend"]["status"] == "error"
@pytest.mark.asyncio
async def test_system_health_handles_network_error(self, mock_client):
"""Test system_health handles network errors gracefully."""
from jana_mcp.tools.system import execute_system_health
mock_client.check_health.side_effect = httpx.RequestError("Network error")
result = await execute_system_health(mock_client)
data = json.loads(result[0].text)
# MCP should still be healthy, backend should show unreachable
assert data["mcp_server"]["status"] == "healthy"
assert data["backend"]["status"] == "unreachable"
