"""Integration tests against live Jana backend.
These tests require:
1. Jana backend running at JANA_BACKEND_URL
2. Valid credentials in .env file
Run with: pytest tests/integration/ -v
"""
import json
import os
import pytest
# Skip all tests if no credentials configured
pytestmark = pytest.mark.skipif(
not os.getenv("JANA_USERNAME") and not os.path.exists(".env"),
reason="No credentials configured for integration tests",
)
class TestLiveAuthentication:
"""Test authentication against live backend."""
@pytest.mark.asyncio
async def test_authenticate_with_credentials(self):
"""Test authentication with username/password."""
from jana_mcp.client import JanaClient
from jana_mcp.config import get_settings
settings = get_settings()
client = JanaClient(settings)
try:
token = await client.authenticate()
assert token is not None
assert len(token) > 0
print(f"✓ Authentication successful, token length: {len(token)}")
finally:
await client.close()
@pytest.mark.asyncio
async def test_health_check(self):
"""Test health check endpoint."""
from jana_mcp.client import JanaClient
from jana_mcp.config import get_settings
settings = get_settings()
client = JanaClient(settings)
try:
health = await client.check_health()
assert health["status"] in ["healthy", "ok"]
print(f"✓ Health check: {health}")
finally:
await client.close()
class TestLiveAirQuality:
"""Test air quality queries against live backend."""
@pytest.mark.asyncio
async def test_get_air_quality_by_country(self):
"""Test fetching air quality data by country."""
from jana_mcp.client import JanaClient
from jana_mcp.config import get_settings
settings = get_settings()
client = JanaClient(settings)
try:
result = await client.get_air_quality(
country_codes=["US"],
parameters=["pm25"],
limit=10,
)
# API returns paginated results with count/results structure
assert "results" in result
assert "count" in result
data = result.get("results", [])
print(f"✓ Air quality query returned {len(data)} results (total: {result.get('count', 0)})")
if data:
sample = data[0]
print(f" Sample: {json.dumps(sample, indent=2)[:200]}...")
finally:
await client.close()
@pytest.mark.asyncio
async def test_get_air_quality_by_bbox(self):
"""Test fetching air quality data by bounding box (San Francisco area)."""
from jana_mcp.client import JanaClient
from jana_mcp.config import get_settings
settings = get_settings()
client = JanaClient(settings)
try:
result = await client.get_air_quality(
bbox=[-122.5, 37.5, -122.0, 38.0], # SF Bay Area
limit=10,
)
# API returns paginated results
assert "results" in result
assert "count" in result
data = result.get("results", [])
print(f"✓ Bbox query returned {len(data)} results (total: {result.get('count', 0)})")
finally:
await client.close()
class TestLiveEmissions:
"""Test emissions queries against live backend."""
@pytest.mark.asyncio
async def test_get_emissions_by_country(self):
"""Test fetching emissions data by country."""
from jana_mcp.client import JanaClient
from jana_mcp.config import get_settings
settings = get_settings()
client = JanaClient(settings)
try:
result = await client.get_emissions(
country_codes=["USA"],
gases=["co2"],
limit=10,
)
# API returns paginated results
assert "results" in result
assert "count" in result
data = result.get("results", [])
print(f"✓ Emissions query returned {len(data)} results (total: {result.get('count', 0)})")
if data:
sample = data[0]
print(f" Sample: {json.dumps(sample, indent=2)[:200]}...")
finally:
await client.close()
@pytest.mark.asyncio
async def test_get_emissions_by_sector(self):
"""Test fetching emissions data by sector."""
from jana_mcp.client import JanaClient
from jana_mcp.config import get_settings
settings = get_settings()
client = JanaClient(settings)
try:
result = await client.get_emissions(
country_codes=["USA"],
sectors=["power"],
limit=10,
)
# API returns paginated results
assert "results" in result
assert "count" in result
data = result.get("results", [])
print(f"✓ Sector query returned {len(data)} results (total: {result.get('count', 0)})")
finally:
await client.close()
class TestLiveNearby:
"""Test nearby queries against live backend."""
@pytest.mark.asyncio
async def test_find_nearby_stations(self):
"""Test finding nearby monitoring stations."""
from jana_mcp.client import JanaClient
from jana_mcp.config import get_settings
settings = get_settings()
client = JanaClient(settings)
try:
result = await client.find_nearby(
point=[-122.4, 37.8], # San Francisco
radius_km=50,
limit=10,
)
# API returns paginated results
assert "results" in result
assert "count" in result
data = result.get("results", [])
print(f"✓ Nearby query returned {len(data)} results (total: {result.get('count', 0)})")
if data:
sample = data[0]
print(f" Sample: {json.dumps(sample, indent=2)[:200]}...")
finally:
await client.close()
class TestLiveToolExecution:
"""Test MCP tool execution against live backend."""
@pytest.mark.asyncio
async def test_execute_air_quality_tool(self):
"""Test executing air quality tool end-to-end."""
from jana_mcp.client import JanaClient
from jana_mcp.config import get_settings
from jana_mcp.tools.air_quality import execute_air_quality
settings = get_settings()
client = JanaClient(settings)
try:
result = await execute_air_quality(
client,
{
"country_codes": ["US"],
"parameters": ["pm25"],
"limit": 5,
},
)
assert len(result) == 1
data = json.loads(result[0].text)
# Should have results (may be empty) or success indicator
assert "results" in data or "count" in data or data.get("success") is True
print(f"✓ Air quality tool executed successfully")
print(f" Response keys: {list(data.keys())}")
if "count" in data:
print(f" Total records: {data['count']}")
finally:
await client.close()
@pytest.mark.asyncio
async def test_execute_emissions_tool(self):
"""Test executing emissions tool end-to-end."""
from jana_mcp.client import JanaClient
from jana_mcp.config import get_settings
from jana_mcp.tools.emissions import execute_emissions
settings = get_settings()
client = JanaClient(settings)
try:
result = await execute_emissions(
client,
{
"country_codes": ["USA"],
"gases": ["co2"],
"limit": 5,
},
)
assert len(result) == 1
data = json.loads(result[0].text)
print(f"✓ Emissions tool executed successfully")
print(f" Response keys: {list(data.keys())}")
finally:
await client.close()
@pytest.mark.asyncio
async def test_execute_system_health_tool(self):
"""Test executing system health tool end-to-end."""
from jana_mcp.client import JanaClient
from jana_mcp.config import get_settings
from jana_mcp.tools.system import execute_system_health
settings = get_settings()
client = JanaClient(settings)
try:
result = await execute_system_health(client)
assert len(result) == 1
data = json.loads(result[0].text)
assert "mcp_server" in data
assert "backend" in data
print(f"✓ System health tool executed successfully")
print(f" MCP status: {data['mcp_server']['status']}")
print(f" Backend status: {data['backend']['status']}")
finally:
await client.close()
@pytest.mark.asyncio
async def test_execute_data_summary_tool(self):
"""Test executing data summary tool end-to-end."""
from jana_mcp.client import JanaClient
from jana_mcp.config import get_settings
from jana_mcp.tools.system import execute_data_summary
settings = get_settings()
client = JanaClient(settings)
try:
result = await execute_data_summary(client)
assert len(result) == 1
data = json.loads(result[0].text)
print(f"✓ Data summary tool executed successfully")
print(f" Response keys: {list(data.keys())}")
finally:
await client.close()
class TestLiveTrends:
"""Test trend analysis against live backend."""
@pytest.mark.asyncio
async def test_execute_trends_tool(self):
"""Test executing trends tool end-to-end."""
from jana_mcp.client import JanaClient
from jana_mcp.config import get_settings
from jana_mcp.tools.trends import execute_trends
settings = get_settings()
client = JanaClient(settings)
try:
result = await execute_trends(
client,
{
"source": "openaq",
"parameter": "pm25",
"country_codes": ["US"],
"temporal_resolution": "monthly",
},
)
assert len(result) == 1
data = json.loads(result[0].text)
print(f"✓ Trends tool executed successfully")
print(f" Response keys: {list(data.keys())}")
if "periods" in data:
print(f" Periods analyzed: {len(data['periods'])}")
if "summary" in data:
print(f" Trend direction: {data['summary'].get('trend_direction', 'N/A')}")
finally:
await client.close()
