"""Tests for trends analysis MCP tool.
Requirements covered:
- AN-001: Provide trend analysis for time-series data
- DA-012: Support temporal aggregations (hourly, daily, monthly)
- 4.2: Tool parameters specification for get_trends
"""
import json
from unittest.mock import AsyncMock, MagicMock
import httpx
import pytest
from jana_mcp.client import APIError, AuthenticationError
from jana_mcp.tools.trends import TRENDS_TOOL
class TestTrendsToolDefinition:
"""Test trends tool definition and schema."""
def test_tool_name(self):
"""Test tool has correct name."""
assert TRENDS_TOOL.name == "get_trends"
def test_tool_has_description(self):
"""Test tool has description."""
assert TRENDS_TOOL.description is not None
assert len(TRENDS_TOOL.description) > 0
assert "trend" in TRENDS_TOOL.description.lower()
def test_tool_mentions_time_series(self):
"""Test tool description mentions time-series (AN-001)."""
assert "time" in TRENDS_TOOL.description.lower()
def test_input_schema_has_source(self):
"""Test schema includes required source parameter."""
props = TRENDS_TOOL.inputSchema["properties"]
assert "source" in props
assert props["source"]["type"] == "string"
# Check source enum values
assert "enum" in props["source"]
sources = props["source"]["enum"]
assert "openaq" in sources
assert "climatetrace" in sources
assert "edgar" in sources
def test_input_schema_has_parameter(self):
"""Test schema includes required parameter."""
props = TRENDS_TOOL.inputSchema["properties"]
assert "parameter" in props
assert props["parameter"]["type"] == "string"
def test_input_schema_has_temporal_resolution(self):
"""Test schema includes temporal_resolution parameter (DA-012)."""
props = TRENDS_TOOL.inputSchema["properties"]
assert "temporal_resolution" in props
assert props["temporal_resolution"]["type"] == "string"
# Check resolution enum values
assert "enum" in props["temporal_resolution"]
resolutions = props["temporal_resolution"]["enum"]
assert "daily" in resolutions
assert "weekly" in resolutions
assert "monthly" in resolutions
def test_input_schema_has_location_bbox(self):
"""Test schema includes location_bbox parameter."""
props = TRENDS_TOOL.inputSchema["properties"]
assert "location_bbox" in props
def test_input_schema_has_country_codes(self):
"""Test schema includes country_codes parameter."""
props = TRENDS_TOOL.inputSchema["properties"]
assert "country_codes" in props
def test_input_schema_has_date_from(self):
"""Test schema includes date_from parameter."""
props = TRENDS_TOOL.inputSchema["properties"]
assert "date_from" in props
def test_input_schema_has_date_to(self):
"""Test schema includes date_to parameter."""
props = TRENDS_TOOL.inputSchema["properties"]
assert "date_to" in props
def test_required_parameters(self):
"""Test source and parameter are required."""
required = TRENDS_TOOL.inputSchema.get("required", [])
assert "source" in required
assert "parameter" in required
class TestTrendsToolExecution:
"""Test trends tool execution."""
@pytest.mark.asyncio
async def test_execute_with_openaq_source(self, mock_client):
"""Test execution with OpenAQ source calls air quality endpoint."""
from jana_mcp.tools.trends import execute_trends
mock_client.get_air_quality.return_value = {
"results": [
{"value": 25.5, "datetime": "2024-01-01T12:00:00Z"},
{"value": 28.3, "datetime": "2024-01-15T12:00:00Z"},
]
}
result = await execute_trends(
mock_client,
{
"source": "openaq",
"parameter": "pm25",
"country_codes": ["USA"],
},
)
mock_client.get_air_quality.assert_called_once()
mock_client.get_emissions.assert_not_called()
@pytest.mark.asyncio
async def test_execute_with_climatetrace_source(self, mock_client):
"""Test execution with Climate TRACE source calls emissions endpoint."""
from jana_mcp.tools.trends import execute_trends
mock_client.get_emissions.return_value = {
"results": [
{"emissions": 1000000, "datetime": "2024-01-01"},
]
}
result = await execute_trends(
mock_client,
{
"source": "climatetrace",
"parameter": "co2",
"country_codes": ["USA"],
},
)
mock_client.get_emissions.assert_called_once()
mock_client.get_air_quality.assert_not_called()
@pytest.mark.asyncio
async def test_execute_with_edgar_source(self, mock_client):
"""Test execution with EDGAR source calls emissions endpoint."""
from jana_mcp.tools.trends import execute_trends
mock_client.get_emissions.return_value = {"results": []}
result = await execute_trends(
mock_client,
{
"source": "edgar",
"parameter": "co2",
"country_codes": ["USA"],
},
)
mock_client.get_emissions.assert_called_once()
call_kwargs = mock_client.get_emissions.call_args[1]
assert call_kwargs["sources"] == ["edgar"]
@pytest.mark.asyncio
async def test_execute_requires_source(self, mock_client):
"""Test execution fails without source."""
from jana_mcp.tools.trends import execute_trends
result = await execute_trends(
mock_client,
{"parameter": "pm25", "country_codes": ["USA"]},
)
assert len(result) == 1
import json
error_data = json.loads(result[0].text)
assert error_data.get("success") is False
assert "error" in error_data
@pytest.mark.asyncio
async def test_execute_requires_parameter(self, mock_client):
"""Test execution fails without parameter."""
from jana_mcp.tools.trends import execute_trends
result = await execute_trends(
mock_client,
{"source": "openaq", "country_codes": ["USA"]},
)
assert len(result) == 1
import json
error_data = json.loads(result[0].text)
assert error_data.get("success") is False
assert "error" in error_data
@pytest.mark.asyncio
async def test_execute_requires_location_filter(self, mock_client):
"""Test execution fails without location filter."""
from jana_mcp.tools.trends import execute_trends
result = await execute_trends(
mock_client,
{"source": "openaq", "parameter": "pm25"},
)
assert len(result) == 1
import json
error_data = json.loads(result[0].text)
assert error_data.get("success") is False
assert "error" in error_data
assert "location" in result[0].text.lower() and "required" in result[0].text.lower()
@pytest.mark.asyncio
async def test_execute_with_temporal_resolution(self, mock_client):
"""Test execution with temporal resolution (DA-012)."""
from jana_mcp.tools.trends import execute_trends
mock_client.get_air_quality.return_value = {"results": []}
result = await execute_trends(
mock_client,
{
"source": "openaq",
"parameter": "pm25",
"country_codes": ["USA"],
"temporal_resolution": "daily",
},
)
# Result should include the resolution
data = json.loads(result[0].text)
assert data["temporal_resolution"] == "daily"
@pytest.mark.asyncio
async def test_execute_default_temporal_resolution(self, mock_client):
"""Test execution uses monthly as default resolution."""
from jana_mcp.tools.trends import execute_trends
mock_client.get_air_quality.return_value = {"results": []}
result = await execute_trends(
mock_client,
{
"source": "openaq",
"parameter": "pm25",
"country_codes": ["USA"],
},
)
data = json.loads(result[0].text)
assert data["temporal_resolution"] == "monthly"
@pytest.mark.asyncio
async def test_execute_with_date_range(self, mock_client):
"""Test execution with date range."""
from jana_mcp.tools.trends import execute_trends
mock_client.get_air_quality.return_value = {"results": []}
result = await execute_trends(
mock_client,
{
"source": "openaq",
"parameter": "pm25",
"country_codes": ["USA"],
"date_from": "2023-01-01",
"date_to": "2024-01-01",
},
)
call_kwargs = mock_client.get_air_quality.call_args[1]
assert call_kwargs["date_from"] == "2023-01-01"
assert call_kwargs["date_to"] == "2024-01-01"
@pytest.mark.asyncio
async def test_execute_with_bbox(self, mock_client):
"""Test execution with bounding box."""
from jana_mcp.tools.trends import execute_trends
mock_client.get_air_quality.return_value = {"results": []}
result = await execute_trends(
mock_client,
{
"source": "openaq",
"parameter": "pm25",
"location_bbox": [-122.5, 37.5, -122.0, 38.0],
},
)
call_kwargs = mock_client.get_air_quality.call_args[1]
assert call_kwargs["bbox"] == [-122.5, 37.5, -122.0, 38.0]
@pytest.mark.asyncio
async def test_execute_handles_empty_data(self, mock_client):
"""Test execution handles empty data gracefully."""
from jana_mcp.tools.trends import execute_trends
mock_client.get_air_quality.return_value = {"results": []}
result = await execute_trends(
mock_client,
{
"source": "openaq",
"parameter": "pm25",
"country_codes": ["USA"],
},
)
data = json.loads(result[0].text)
assert "error" in data or "data_points" in data
@pytest.mark.asyncio
async def test_execute_handles_api_error(self, mock_client):
"""Test execution handles API errors gracefully."""
from jana_mcp.tools.trends import execute_trends
import httpx
mock_client.get_air_quality.side_effect = httpx.RequestError("API Error")
result = await execute_trends(
mock_client,
{
"source": "openaq",
"parameter": "pm25",
"country_codes": ["USA"],
},
)
assert len(result) == 1
import json
error_data = json.loads(result[0].text)
assert error_data.get("success") is False
assert "error" in error_data
@pytest.mark.asyncio
async def test_result_includes_source_and_parameter(self, mock_client):
"""Test result includes source and parameter in response."""
from jana_mcp.tools.trends import execute_trends
mock_client.get_air_quality.return_value = {"results": []}
result = await execute_trends(
mock_client,
{
"source": "openaq",
"parameter": "pm25",
"country_codes": ["USA"],
},
)
data = json.loads(result[0].text)
assert data["source"] == "openaq"
assert data["parameter"] == "pm25"
@pytest.mark.asyncio
async def test_execute_fetches_more_data_for_trends(self, mock_client):
"""Test execution fetches more data (limit=1000) for trend analysis."""
from jana_mcp.tools.trends import execute_trends
mock_client.get_air_quality.return_value = {"results": []}
result = await execute_trends(
mock_client,
{
"source": "openaq",
"parameter": "pm25",
"country_codes": ["USA"],
},
)
call_kwargs = mock_client.get_air_quality.call_args[1]
assert call_kwargs["limit"] == 1000
class TestTrendCalculations:
"""Test trend calculation logic."""
def testaggregate_by_period_monthly(self):
"""Test aggregation by monthly period."""
from jana_mcp.tools.trends import aggregate_by_period
data = [
{"datetime": "2024-01-15T12:00:00Z", "value": 10},
{"datetime": "2024-01-20T12:00:00Z", "value": 20},
{"datetime": "2024-02-10T12:00:00Z", "value": 30},
]
result = aggregate_by_period(data, "monthly")
assert "2024-01" in result
assert "2024-02" in result
assert len(result["2024-01"]) == 2
assert len(result["2024-02"]) == 1
def testaggregate_by_period_daily(self):
"""Test aggregation by daily period."""
from jana_mcp.tools.trends import aggregate_by_period
data = [
{"datetime": "2024-01-15T10:00:00Z", "value": 10},
{"datetime": "2024-01-15T14:00:00Z", "value": 20},
{"datetime": "2024-01-16T12:00:00Z", "value": 30},
]
result = aggregate_by_period(data, "daily")
assert "2024-01-15" in result
assert "2024-01-16" in result
assert len(result["2024-01-15"]) == 2
assert len(result["2024-01-16"]) == 1
def testaggregate_by_period_weekly(self):
"""Test aggregation by weekly period."""
from jana_mcp.tools.trends import aggregate_by_period
data = [
{"datetime": "2024-01-01T12:00:00Z", "value": 10},
{"datetime": "2024-01-08T12:00:00Z", "value": 20},
]
result = aggregate_by_period(data, "weekly")
# Should have different week keys
assert len(result) >= 1
def test_aggregate_handles_missing_timestamp(self):
"""Test aggregation handles items without timestamp."""
from jana_mcp.tools.trends import aggregate_by_period
data = [
{"datetime": "2024-01-15T12:00:00Z", "value": 10},
{"value": 20}, # No timestamp
]
result = aggregate_by_period(data, "monthly")
assert "2024-01" in result
assert len(result["2024-01"]) == 1
def test_aggregate_handles_different_value_fields(self):
"""Test aggregation handles different value field names."""
from jana_mcp.tools.trends import aggregate_by_period
data = [
{"datetime": "2024-01-15T12:00:00Z", "value": 10},
{"datetime": "2024-01-16T12:00:00Z", "emissions": 20},
{"datetime": "2024-01-17T12:00:00Z", "measurement": 30},
]
result = aggregate_by_period(data, "daily")
assert len(result) == 3
def testcalculate_trend_stats_empty(self):
"""Test trend stats with empty data."""
from jana_mcp.tools.trends import calculate_trend_stats
result = calculate_trend_stats({})
assert "error" in result
def testcalculate_trend_stats_basic(self):
"""Test trend stats calculation."""
from jana_mcp.tools.trends import calculate_trend_stats
aggregated = {
"2024-01": [10, 20],
"2024-02": [15, 25],
"2024-03": [20, 30],
}
result = calculate_trend_stats(aggregated)
assert "periods" in result
assert "summary" in result
assert result["summary"]["total_periods"] == 3
assert "trend_direction" in result["summary"]
assert "change_percent" in result["summary"]
def testcalculate_trend_stats_increasing(self):
"""Test trend stats detects increasing trend."""
from jana_mcp.tools.trends import calculate_trend_stats
aggregated = {
"2024-01": [10],
"2024-02": [20],
"2024-03": [30],
"2024-04": [40],
}
result = calculate_trend_stats(aggregated)
assert result["summary"]["trend_direction"] == "increasing"
assert result["summary"]["change_percent"] > 0
def testcalculate_trend_stats_decreasing(self):
"""Test trend stats detects decreasing trend."""
from jana_mcp.tools.trends import calculate_trend_stats
aggregated = {
"2024-01": [40],
"2024-02": [30],
"2024-03": [20],
"2024-04": [10],
}
result = calculate_trend_stats(aggregated)
assert result["summary"]["trend_direction"] == "decreasing"
assert result["summary"]["change_percent"] < 0
class TestTrendsToolErrorPaths:
"""Test error handling paths for trends tool."""
@pytest.mark.asyncio
async def test_handles_api_error(self, mock_client):
"""Test handles APIError gracefully."""
from jana_mcp.tools.trends import execute_trends
mock_client.get_air_quality.side_effect = APIError("API Error", status_code=500)
result = await execute_trends(
mock_client,
{"source": "openaq", "parameter": "pm25", "country_codes": ["USA"]},
)
error_data = json.loads(result[0].text)
assert error_data.get("success") is False
assert error_data.get("error_code") == "API_ERROR"
@pytest.mark.asyncio
async def test_handles_authentication_error(self, mock_client):
"""Test handles AuthenticationError gracefully."""
from jana_mcp.tools.trends import execute_trends
mock_client.get_air_quality.side_effect = AuthenticationError("Auth failed")
result = await execute_trends(
mock_client,
{"source": "openaq", "parameter": "pm25", "country_codes": ["USA"]},
)
error_data = json.loads(result[0].text)
assert error_data.get("success") is False
assert error_data.get("error_code") == "API_ERROR"
@pytest.mark.asyncio
async def test_handles_network_error(self, mock_client):
"""Test handles network errors gracefully."""
from jana_mcp.tools.trends import execute_trends
mock_client.get_air_quality.side_effect = httpx.RequestError("Network error")
result = await execute_trends(
mock_client,
{"source": "openaq", "parameter": "pm25", "country_codes": ["USA"]},
)
error_data = json.loads(result[0].text)
assert error_data.get("success") is False
assert error_data.get("error_code") == "NETWORK_ERROR"
@pytest.mark.asyncio
async def test_handles_data_error(self, mock_client):
"""Test handles data parsing errors."""
from jana_mcp.tools.trends import execute_trends
mock_client.get_air_quality.side_effect = KeyError("missing_key")
result = await execute_trends(
mock_client,
{"source": "openaq", "parameter": "pm25", "country_codes": ["USA"]},
)
error_data = json.loads(result[0].text)
assert error_data.get("success") is False
assert error_data.get("error_code") == "DATA_ERROR"
@pytest.mark.asyncio
async def test_validates_invalid_bbox(self, mock_client):
"""Test rejects invalid bounding box."""
from jana_mcp.tools.trends import execute_trends
result = await execute_trends(
mock_client,
{
"source": "openaq",
"parameter": "pm25",
"location_bbox": [-100.0, 40.0, -110.0, 50.0], # min_lon > max_lon
},
)
error_data = json.loads(result[0].text)
assert error_data.get("success") is False
assert error_data.get("error_code") == "VALIDATION_ERROR"
mock_client.get_air_quality.assert_not_called()
@pytest.mark.asyncio
async def test_validates_invalid_coordinates(self, mock_client):
"""Test rejects invalid coordinates."""
from jana_mcp.tools.trends import execute_trends
result = await execute_trends(
mock_client,
{
"source": "openaq",
"parameter": "pm25",
"location_point": [200.0, 40.0], # lon > 180
},
)
error_data = json.loads(result[0].text)
assert error_data.get("success") is False
assert error_data.get("error_code") == "VALIDATION_ERROR"
mock_client.get_air_quality.assert_not_called()
@pytest.mark.asyncio
async def test_validates_invalid_date(self, mock_client):
"""Test rejects invalid date format."""
from jana_mcp.tools.trends import execute_trends
result = await execute_trends(
mock_client,
{
"source": "openaq",
"parameter": "pm25",
"country_codes": ["USA"],
"date_from": "not-a-date",
},
)
error_data = json.loads(result[0].text)
assert error_data.get("success") is False
assert error_data.get("error_code") == "VALIDATION_ERROR"
mock_client.get_air_quality.assert_not_called()
@pytest.mark.asyncio
async def test_requires_location_filter(self, mock_client):
"""Test requires at least one location filter."""
from jana_mcp.tools.trends import execute_trends
result = await execute_trends(
mock_client,
{"source": "openaq", "parameter": "pm25"},
)
error_data = json.loads(result[0].text)
assert error_data.get("success") is False
mock_client.get_air_quality.assert_not_called()
@pytest.mark.asyncio
async def test_emissions_api_error(self, mock_client):
"""Test emissions source handles API error."""
from jana_mcp.tools.trends import execute_trends
mock_client.get_emissions.side_effect = APIError("API Error", status_code=500)
result = await execute_trends(
mock_client,
{"source": "climatetrace", "parameter": "co2", "country_codes": ["USA"]},
)
error_data = json.loads(result[0].text)
assert error_data.get("success") is False
assert error_data.get("error_code") == "API_ERROR"
