"""Tests for emissions MCP tool.
Requirements covered:
- DA-004: Retrieve GHG emissions data from Climate TRACE
- DA-005: Retrieve national emissions data from EDGAR
- DA-006: Support filtering by sector (power, steel, cement, etc.)
- DA-007: Support filtering by gas type (CO2, CH4, N2O)
- DA-013: Support unified queries across multiple data sources
- DA-014: Return data with proper source attribution
- 4.2: Tool parameters specification for get_emissions
"""
import json
from unittest.mock import AsyncMock, MagicMock
import httpx
import pytest
from jana_mcp.client import APIError, AuthenticationError
from jana_mcp.tools.emissions import EMISSIONS_TOOL
class TestEmissionsToolDefinition:
"""Test emissions tool definition and schema."""
def test_tool_name(self):
"""Test tool has correct name."""
assert EMISSIONS_TOOL.name == "get_emissions"
def test_tool_has_description(self):
"""Test tool has description."""
assert EMISSIONS_TOOL.description is not None
assert len(EMISSIONS_TOOL.description) > 0
assert "emissions" in EMISSIONS_TOOL.description.lower()
def test_tool_mentions_climate_trace(self):
"""Test tool description mentions Climate TRACE (DA-004)."""
assert "Climate TRACE" in EMISSIONS_TOOL.description
def test_tool_mentions_edgar(self):
"""Test tool description mentions EDGAR (DA-005)."""
assert "EDGAR" in EMISSIONS_TOOL.description
def test_input_schema_has_sources(self):
"""Test schema includes sources parameter (DA-004, DA-005, DA-013)."""
props = EMISSIONS_TOOL.inputSchema["properties"]
assert "sources" in props
assert props["sources"]["type"] == "array"
# Check source enum values
source_items = props["sources"]["items"]
assert "enum" in source_items
sources = source_items["enum"]
assert "climatetrace" in sources
assert "edgar" in sources
def test_input_schema_has_sectors(self):
"""Test schema includes sectors parameter (DA-006)."""
props = EMISSIONS_TOOL.inputSchema["properties"]
assert "sectors" in props
assert props["sectors"]["type"] == "array"
def test_input_schema_has_gases(self):
"""Test schema includes gases parameter (DA-007)."""
props = EMISSIONS_TOOL.inputSchema["properties"]
assert "gases" in props
assert props["gases"]["type"] == "array"
# Check gas enum values
gas_items = props["gases"]["items"]
assert "enum" in gas_items
gases = gas_items["enum"]
assert "co2" in gases
assert "ch4" in gases
assert "n2o" in gases
def test_input_schema_has_location_bbox(self):
"""Test schema includes location_bbox parameter."""
props = EMISSIONS_TOOL.inputSchema["properties"]
assert "location_bbox" in props
def test_input_schema_has_location_point(self):
"""Test schema includes location_point parameter."""
props = EMISSIONS_TOOL.inputSchema["properties"]
assert "location_point" in props
def test_input_schema_has_radius_km(self):
"""Test schema includes radius_km parameter."""
props = EMISSIONS_TOOL.inputSchema["properties"]
assert "radius_km" in props
def test_input_schema_has_country_codes(self):
"""Test schema includes country_codes parameter."""
props = EMISSIONS_TOOL.inputSchema["properties"]
assert "country_codes" in props
def test_input_schema_has_date_from(self):
"""Test schema includes date_from parameter."""
props = EMISSIONS_TOOL.inputSchema["properties"]
assert "date_from" in props
def test_input_schema_has_date_to(self):
"""Test schema includes date_to parameter."""
props = EMISSIONS_TOOL.inputSchema["properties"]
assert "date_to" in props
def test_input_schema_has_limit(self):
"""Test schema includes limit parameter."""
props = EMISSIONS_TOOL.inputSchema["properties"]
assert "limit" in props
class TestEmissionsToolExecution:
"""Test emissions tool execution."""
@pytest.mark.asyncio
async def test_execute_with_country_codes(self, mock_client):
"""Test execution with country codes."""
from jana_mcp.tools.emissions import execute_emissions
mock_client.get_emissions.return_value = {
"results": [
{"source": "climatetrace", "emissions": 1000000, "gas": "co2"}
]
}
result = await execute_emissions(
mock_client,
{"country_codes": ["USA"]},
)
mock_client.get_emissions.assert_called_once()
call_kwargs = mock_client.get_emissions.call_args[1]
assert call_kwargs["country_codes"] == ["USA"]
@pytest.mark.asyncio
async def test_execute_with_sources(self, mock_client):
"""Test execution with data sources (DA-013)."""
from jana_mcp.tools.emissions import execute_emissions
mock_client.get_emissions.return_value = {"results": []}
result = await execute_emissions(
mock_client,
{"country_codes": ["USA"], "sources": ["climatetrace", "edgar"]},
)
call_kwargs = mock_client.get_emissions.call_args[1]
assert call_kwargs["sources"] == ["climatetrace", "edgar"]
@pytest.mark.asyncio
async def test_execute_with_climatetrace_only(self, mock_client):
"""Test execution with Climate TRACE only (DA-004)."""
from jana_mcp.tools.emissions import execute_emissions
mock_client.get_emissions.return_value = {"results": []}
result = await execute_emissions(
mock_client,
{"country_codes": ["USA"], "sources": ["climatetrace"]},
)
call_kwargs = mock_client.get_emissions.call_args[1]
assert call_kwargs["sources"] == ["climatetrace"]
@pytest.mark.asyncio
async def test_execute_with_edgar_only(self, mock_client):
"""Test execution with EDGAR only (DA-005)."""
from jana_mcp.tools.emissions import execute_emissions
mock_client.get_emissions.return_value = {"results": []}
result = await execute_emissions(
mock_client,
{"country_codes": ["USA"], "sources": ["edgar"]},
)
call_kwargs = mock_client.get_emissions.call_args[1]
assert call_kwargs["sources"] == ["edgar"]
@pytest.mark.asyncio
async def test_execute_with_sectors(self, mock_client):
"""Test execution with sector filter (DA-006)."""
from jana_mcp.tools.emissions import execute_emissions
mock_client.get_emissions.return_value = {"results": []}
result = await execute_emissions(
mock_client,
{"country_codes": ["USA"], "sectors": ["power", "steel", "cement"]},
)
call_kwargs = mock_client.get_emissions.call_args[1]
assert call_kwargs["sectors"] == ["power", "steel", "cement"]
@pytest.mark.asyncio
async def test_execute_with_gases(self, mock_client):
"""Test execution with gas filter (DA-007)."""
from jana_mcp.tools.emissions import execute_emissions
mock_client.get_emissions.return_value = {"results": []}
result = await execute_emissions(
mock_client,
{"country_codes": ["USA"], "gases": ["co2", "ch4"]},
)
call_kwargs = mock_client.get_emissions.call_args[1]
assert call_kwargs["gases"] == ["co2", "ch4"]
@pytest.mark.asyncio
async def test_execute_with_bbox(self, mock_client):
"""Test execution with bounding box."""
from jana_mcp.tools.emissions import execute_emissions
mock_client.get_emissions.return_value = {"results": []}
result = await execute_emissions(
mock_client,
{"location_bbox": [-122.5, 37.5, -122.0, 38.0]},
)
call_kwargs = mock_client.get_emissions.call_args[1]
assert call_kwargs["bbox"] == [-122.5, 37.5, -122.0, 38.0]
@pytest.mark.asyncio
async def test_execute_with_point_radius(self, mock_client):
"""Test execution with point and radius."""
from jana_mcp.tools.emissions import execute_emissions
mock_client.get_emissions.return_value = {"results": []}
result = await execute_emissions(
mock_client,
{"location_point": [-122.4, 37.7], "radius_km": 50},
)
call_kwargs = mock_client.get_emissions.call_args[1]
assert call_kwargs["point"] == [-122.4, 37.7]
assert call_kwargs["radius_km"] == 50
@pytest.mark.asyncio
async def test_execute_requires_location_filter(self, mock_client):
"""Test execution fails without location filter."""
from jana_mcp.tools.emissions import execute_emissions
result = await execute_emissions(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 result[0].text.lower() and "required" in result[0].text.lower()
mock_client.get_emissions.assert_not_called()
@pytest.mark.asyncio
async def test_execute_handles_api_error(self, mock_client):
"""Test execution handles API errors gracefully."""
from jana_mcp.tools.emissions import execute_emissions
import httpx
mock_client.get_emissions.side_effect = httpx.RequestError("API Error")
result = await execute_emissions(
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_result_includes_source_attribution(self, mock_client):
"""Test result includes source attribution (DA-014)."""
from jana_mcp.tools.emissions import execute_emissions
mock_client.get_emissions.return_value = {
"results": [
{
"source": "climatetrace",
"emissions": 1000000,
"gas": "co2",
"sector": "power",
},
{
"source": "edgar",
"emissions": 500000,
"gas": "co2",
"sector": "power",
},
]
}
result = await execute_emissions(
mock_client,
{"country_codes": ["USA"]},
)
data = json.loads(result[0].text)
# Verify source attribution in results
for item in data["results"]:
assert "source" in item
@pytest.mark.asyncio
async def test_execute_with_date_range(self, mock_client):
"""Test execution with date range."""
from jana_mcp.tools.emissions import execute_emissions
mock_client.get_emissions.return_value = {"results": []}
result = await execute_emissions(
mock_client,
{
"country_codes": ["USA"],
"date_from": "2020-01-01",
"date_to": "2023-12-31",
},
)
call_kwargs = mock_client.get_emissions.call_args[1]
assert call_kwargs["date_from"] == "2020-01-01"
assert call_kwargs["date_to"] == "2023-12-31"
@pytest.mark.asyncio
async def test_execute_default_limit(self, mock_client):
"""Test execution uses default limit."""
from jana_mcp.tools.emissions import execute_emissions
mock_client.get_emissions.return_value = {"results": []}
result = await execute_emissions(
mock_client,
{"country_codes": ["USA"]},
)
call_kwargs = mock_client.get_emissions.call_args[1]
assert call_kwargs["limit"] == 100
class TestEmissionsToolValidation:
"""Test input validation for emissions tool."""
@pytest.mark.asyncio
async def test_accepts_valid_sources(self, mock_client):
"""Test accepts valid data sources."""
from jana_mcp.tools.emissions import execute_emissions
result = await execute_emissions(
mock_client,
{"country_codes": ["USA"], "sources": ["climatetrace"]},
)
mock_client.get_emissions.assert_called_once()
@pytest.mark.asyncio
async def test_accepts_multiple_gas_types(self, mock_client):
"""Test accepts multiple gas types."""
from jana_mcp.tools.emissions import execute_emissions
result = await execute_emissions(
mock_client,
{"country_codes": ["USA"], "gases": ["co2", "ch4", "n2o"]},
)
mock_client.get_emissions.assert_called_once()
@pytest.mark.asyncio
async def test_accepts_multiple_sectors(self, mock_client):
"""Test accepts multiple sectors."""
from jana_mcp.tools.emissions import execute_emissions
result = await execute_emissions(
mock_client,
{"country_codes": ["USA"], "sectors": ["power", "steel", "cement", "oil-and-gas"]},
)
mock_client.get_emissions.assert_called_once()
class TestEmissionsToolErrorPaths:
"""Test error handling paths for emissions tool."""
@pytest.mark.asyncio
async def test_handles_api_error(self, mock_client):
"""Test handles APIError gracefully."""
from jana_mcp.tools.emissions import execute_emissions
mock_client.get_emissions.side_effect = APIError("API Error", status_code=500)
result = await execute_emissions(
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.emissions import execute_emissions
mock_client.get_emissions.side_effect = AuthenticationError("Auth failed")
result = await execute_emissions(
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.emissions import execute_emissions
mock_client.get_emissions.side_effect = httpx.RequestError("Network error")
result = await execute_emissions(
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_data_error(self, mock_client):
"""Test handles data parsing errors."""
from jana_mcp.tools.emissions import execute_emissions
mock_client.get_emissions.side_effect = KeyError("missing_key")
result = await execute_emissions(
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.emissions import execute_emissions
result = await execute_emissions(
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_emissions.assert_not_called()
@pytest.mark.asyncio
async def test_validates_invalid_coordinates(self, mock_client):
"""Test rejects invalid coordinates."""
from jana_mcp.tools.emissions import execute_emissions
result = await execute_emissions(
mock_client,
{"location_point": [200.0, 40.0]},
)
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_emissions.assert_not_called()
@pytest.mark.asyncio
async def test_validates_invalid_date(self, mock_client):
"""Test rejects invalid date format."""
from jana_mcp.tools.emissions import execute_emissions
result = await execute_emissions(
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_emissions.assert_not_called()
