"""Tests for air quality MCP tool.
Requirements covered:
- DA-001: Retrieve air quality measurements from OpenAQ data
- DA-002: Support filtering by pollutant parameters (PM2.5, PM10, O3, NO2, SO2, CO)
- DA-003: Return measurement values with units, timestamps, and location information
- DA-008: Support bounding box queries
- DA-009: Support point + radius queries
- DA-010: Support country code filtering
- DA-011: Support date range filtering
- 4.2: Tool parameters specification for get_air_quality
"""
import json
from unittest.mock import AsyncMock, MagicMock
import httpx
import pytest
from jana_mcp.client import APIError, AuthenticationError
from jana_mcp.tools.air_quality import AIR_QUALITY_TOOL
class TestAirQualityToolDefinition:
"""Test air quality tool definition and schema."""
def test_tool_name(self):
"""Test tool has correct name."""
assert AIR_QUALITY_TOOL.name == "get_air_quality"
def test_tool_has_description(self):
"""Test tool has description."""
assert AIR_QUALITY_TOOL.description is not None
assert len(AIR_QUALITY_TOOL.description) > 0
assert "air quality" in AIR_QUALITY_TOOL.description.lower()
def test_tool_mentions_openaq(self):
"""Test tool description mentions OpenAQ (DA-001)."""
assert "OpenAQ" in AIR_QUALITY_TOOL.description
def test_tool_mentions_pollutants(self):
"""Test tool description mentions pollutant types (DA-002)."""
description = AIR_QUALITY_TOOL.description.lower()
assert "pm2.5" in description or "pm25" in description
def test_input_schema_has_location_bbox(self):
"""Test schema includes location_bbox parameter (DA-008)."""
props = AIR_QUALITY_TOOL.inputSchema["properties"]
assert "location_bbox" in props
assert props["location_bbox"]["type"] == "array"
assert props["location_bbox"]["minItems"] == 4
assert props["location_bbox"]["maxItems"] == 4
def test_input_schema_has_location_point(self):
"""Test schema includes location_point parameter (DA-009)."""
props = AIR_QUALITY_TOOL.inputSchema["properties"]
assert "location_point" in props
assert props["location_point"]["type"] == "array"
assert props["location_point"]["minItems"] == 2
assert props["location_point"]["maxItems"] == 2
def test_input_schema_has_radius_km(self):
"""Test schema includes radius_km parameter (DA-009)."""
props = AIR_QUALITY_TOOL.inputSchema["properties"]
assert "radius_km" in props
assert props["radius_km"]["type"] == "number"
assert props["radius_km"]["minimum"] == 0.1
assert props["radius_km"]["maximum"] == 500
def test_input_schema_has_country_codes(self):
"""Test schema includes country_codes parameter (DA-010)."""
props = AIR_QUALITY_TOOL.inputSchema["properties"]
assert "country_codes" in props
assert props["country_codes"]["type"] == "array"
def test_input_schema_has_parameters(self):
"""Test schema includes parameters filter (DA-002)."""
props = AIR_QUALITY_TOOL.inputSchema["properties"]
assert "parameters" in props
assert props["parameters"]["type"] == "array"
# Check pollutant enum values
param_items = props["parameters"]["items"]
assert "enum" in param_items
pollutants = param_items["enum"]
assert "pm25" in pollutants
assert "pm10" in pollutants
assert "o3" in pollutants
assert "no2" in pollutants
assert "so2" in pollutants
assert "co" in pollutants
def test_input_schema_has_date_from(self):
"""Test schema includes date_from parameter (DA-011)."""
props = AIR_QUALITY_TOOL.inputSchema["properties"]
assert "date_from" in props
assert props["date_from"]["type"] == "string"
def test_input_schema_has_date_to(self):
"""Test schema includes date_to parameter (DA-011)."""
props = AIR_QUALITY_TOOL.inputSchema["properties"]
assert "date_to" in props
assert props["date_to"]["type"] == "string"
def test_input_schema_has_limit(self):
"""Test schema includes limit parameter."""
props = AIR_QUALITY_TOOL.inputSchema["properties"]
assert "limit" in props
assert props["limit"]["type"] == "integer"
assert props["limit"]["minimum"] == 1
assert props["limit"]["maximum"] == 1000
assert props["limit"]["default"] == 100
def test_no_required_parameters(self):
"""Test no parameters are strictly required (location validation is runtime)."""
required = AIR_QUALITY_TOOL.inputSchema.get("required", [])
assert required == []
class TestAirQualityToolExecution:
"""Test air quality tool execution."""
@pytest.mark.asyncio
async def test_execute_with_bbox(self, mock_client):
"""Test execution with bounding box (DA-008)."""
from jana_mcp.tools.air_quality import execute_air_quality
mock_client.get_air_quality.return_value = {
"results": [
{"value": 25.5, "parameter": "pm25", "location": "Test", "datetime": "2024-01-01"}
]
}
result = await execute_air_quality(
mock_client,
{"location_bbox": [-122.5, 37.5, -122.0, 38.0]},
)
mock_client.get_air_quality.assert_called_once()
call_kwargs = mock_client.get_air_quality.call_args[1]
assert call_kwargs["bbox"] == [-122.5, 37.5, -122.0, 38.0]
# Check result is valid JSON
assert len(result) == 1
assert result[0].type == "text"
data = json.loads(result[0].text)
assert "results" in data
@pytest.mark.asyncio
async def test_execute_with_point_radius(self, mock_client):
"""Test execution with point + radius (DA-009)."""
from jana_mcp.tools.air_quality import execute_air_quality
mock_client.get_air_quality.return_value = {"results": []}
result = await execute_air_quality(
mock_client,
{"location_point": [-122.4, 37.7], "radius_km": 10},
)
call_kwargs = mock_client.get_air_quality.call_args[1]
assert call_kwargs["point"] == [-122.4, 37.7]
assert call_kwargs["radius_km"] == 10
@pytest.mark.asyncio
async def test_execute_with_country_codes(self, mock_client):
"""Test execution with country codes (DA-010)."""
from jana_mcp.tools.air_quality import execute_air_quality
mock_client.get_air_quality.return_value = {"results": []}
result = await execute_air_quality(
mock_client,
{"country_codes": ["USA", "GBR"]},
)
call_kwargs = mock_client.get_air_quality.call_args[1]
assert call_kwargs["country_codes"] == ["USA", "GBR"]
@pytest.mark.asyncio
async def test_execute_with_parameters(self, mock_client):
"""Test execution with pollutant parameters (DA-002)."""
from jana_mcp.tools.air_quality import execute_air_quality
mock_client.get_air_quality.return_value = {"results": []}
result = await execute_air_quality(
mock_client,
{"country_codes": ["USA"], "parameters": ["pm25", "o3"]},
)
call_kwargs = mock_client.get_air_quality.call_args[1]
assert call_kwargs["parameters"] == ["pm25", "o3"]
@pytest.mark.asyncio
async def test_execute_with_date_range(self, mock_client):
"""Test execution with date range (DA-011)."""
from jana_mcp.tools.air_quality import execute_air_quality
mock_client.get_air_quality.return_value = {"results": []}
result = await execute_air_quality(
mock_client,
{
"country_codes": ["USA"],
"date_from": "2024-01-01",
"date_to": "2024-06-30",
},
)
call_kwargs = mock_client.get_air_quality.call_args[1]
assert call_kwargs["date_from"] == "2024-01-01"
assert call_kwargs["date_to"] == "2024-06-30"
@pytest.mark.asyncio
async def test_execute_requires_location_filter(self, mock_client):
"""Test execution fails without location filter."""
from jana_mcp.tools.air_quality import execute_air_quality
result = await execute_air_quality(mock_client, {})
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 error_data["error"].lower() and "required" in error_data["error"].lower()
mock_client.get_air_quality.assert_not_called()
@pytest.mark.asyncio
async def test_execute_handles_api_error(self, mock_client):
"""Test execution handles API errors gracefully (EH-001)."""
from jana_mcp.tools.air_quality import execute_air_quality
import httpx
mock_client.get_air_quality.side_effect = httpx.RequestError("API Error")
result = await execute_air_quality(
mock_client,
{"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_default_limit(self, mock_client):
"""Test execution uses default limit."""
from jana_mcp.tools.air_quality import execute_air_quality
mock_client.get_air_quality.return_value = {"results": []}
result = await execute_air_quality(
mock_client,
{"country_codes": ["USA"]},
)
call_kwargs = mock_client.get_air_quality.call_args[1]
assert call_kwargs["limit"] == 100
@pytest.mark.asyncio
async def test_execute_custom_limit(self, mock_client):
"""Test execution with custom limit."""
from jana_mcp.tools.air_quality import execute_air_quality
mock_client.get_air_quality.return_value = {"results": []}
result = await execute_air_quality(
mock_client,
{"country_codes": ["USA"], "limit": 50},
)
call_kwargs = mock_client.get_air_quality.call_args[1]
assert call_kwargs["limit"] == 50
@pytest.mark.asyncio
async def test_result_format_json(self, mock_client):
"""Test result is properly formatted JSON (DA-003)."""
from jana_mcp.tools.air_quality import execute_air_quality
mock_client.get_air_quality.return_value = {
"results": [
{
"value": 25.5,
"unit": "µg/m³",
"parameter": "pm25",
"datetime": "2024-01-01T12:00:00Z",
"location": {
"name": "Test Station",
"coordinates": {"latitude": 37.7, "longitude": -122.4},
},
}
]
}
result = await execute_air_quality(
mock_client,
{"country_codes": ["USA"]},
)
data = json.loads(result[0].text)
measurement = data["results"][0]
# Verify measurement contains required fields (DA-003)
assert "value" in measurement
assert "datetime" in measurement
assert "location" in measurement
class TestAirQualityToolValidation:
"""Test input validation for air quality tool."""
@pytest.mark.asyncio
async def test_accepts_valid_bbox(self, mock_client):
"""Test accepts valid bounding box."""
from jana_mcp.tools.air_quality import execute_air_quality
result = await execute_air_quality(
mock_client,
{"location_bbox": [-180, -90, 180, 90]},
)
mock_client.get_air_quality.assert_called_once()
@pytest.mark.asyncio
async def test_accepts_valid_point(self, mock_client):
"""Test accepts valid point coordinates."""
from jana_mcp.tools.air_quality import execute_air_quality
result = await execute_air_quality(
mock_client,
{"location_point": [0, 0], "radius_km": 10},
)
mock_client.get_air_quality.assert_called_once()
@pytest.mark.asyncio
async def test_accepts_valid_country_codes(self, mock_client):
"""Test accepts valid ISO-3 country codes."""
from jana_mcp.tools.air_quality import execute_air_quality
result = await execute_air_quality(
mock_client,
{"country_codes": ["USA", "CAN", "MEX"]},
)
mock_client.get_air_quality.assert_called_once()
@pytest.mark.asyncio
async def test_bbox_or_point_or_country_required(self, mock_client):
"""Test at least one location filter is required."""
from jana_mcp.tools.air_quality import execute_air_quality
# Only date filters, no location
result = await execute_air_quality(
mock_client,
{"date_from": "2024-01-01", "parameters": ["pm25"]},
)
import json
error_data = json.loads(result[0].text)
assert error_data.get("success") is False
assert "error" in error_data
mock_client.get_air_quality.assert_not_called()
class TestAirQualityToolErrorPaths:
"""Test error handling paths for air quality tool."""
@pytest.mark.asyncio
async def test_handles_api_error(self, mock_client):
"""Test handles APIError gracefully."""
from jana_mcp.tools.air_quality import execute_air_quality
mock_client.get_air_quality.side_effect = APIError("API Error", status_code=500)
result = await execute_air_quality(
mock_client,
{"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.air_quality import execute_air_quality
mock_client.get_air_quality.side_effect = AuthenticationError("Auth failed")
result = await execute_air_quality(
mock_client,
{"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.air_quality import execute_air_quality
mock_client.get_air_quality.side_effect = httpx.RequestError("Network error")
result = await execute_air_quality(
mock_client,
{"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_key_error(self, mock_client):
"""Test handles KeyError in response parsing."""
from jana_mcp.tools.air_quality import execute_air_quality
mock_client.get_air_quality.side_effect = KeyError("missing_key")
result = await execute_air_quality(
mock_client,
{"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_handles_value_error(self, mock_client):
"""Test handles ValueError in response parsing."""
from jana_mcp.tools.air_quality import execute_air_quality
mock_client.get_air_quality.side_effect = ValueError("Invalid value")
result = await execute_air_quality(
mock_client,
{"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_handles_type_error(self, mock_client):
"""Test handles TypeError in response parsing."""
from jana_mcp.tools.air_quality import execute_air_quality
mock_client.get_air_quality.side_effect = TypeError("Type error")
result = await execute_air_quality(
mock_client,
{"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.air_quality import execute_air_quality
# bbox with min > max
result = await execute_air_quality(
mock_client,
{"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.air_quality import execute_air_quality
# longitude out of bounds
result = await execute_air_quality(
mock_client,
{"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_from(self, mock_client):
"""Test rejects invalid date_from format."""
from jana_mcp.tools.air_quality import execute_air_quality
result = await execute_air_quality(
mock_client,
{"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_validates_invalid_date_to(self, mock_client):
"""Test rejects invalid date_to format."""
from jana_mcp.tools.air_quality import execute_air_quality
result = await execute_air_quality(
mock_client,
{"country_codes": ["USA"], "date_to": "invalid"},
)
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()
