"""Pytest configuration and fixtures."""
import os
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from pydantic import SecretStr
from jana_mcp.client import JanaClient
from jana_mcp.config import Settings
@pytest.fixture
def settings() -> Settings:
"""Create test settings with credentials."""
return Settings(
jana_backend_url="http://localhost:8000",
jana_username="testuser",
jana_password=SecretStr("testpass"),
mcp_server_port=8080,
log_level="DEBUG",
)
@pytest.fixture
def settings_with_token() -> Settings:
"""Create test settings with pre-configured token."""
return Settings(
jana_backend_url="http://localhost:8000",
jana_token=SecretStr("test-token-12345"),
mcp_server_port=8080,
log_level="DEBUG",
)
@pytest.fixture
def settings_no_auth() -> Settings:
"""Create test settings without authentication."""
return Settings(
jana_backend_url="http://localhost:8000",
mcp_server_port=8080,
log_level="DEBUG",
)
@pytest.fixture
def mock_client() -> MagicMock:
"""Create a mock Jana client."""
client = MagicMock(spec=JanaClient)
client.authenticate = AsyncMock(return_value="test-token")
client.get_air_quality = AsyncMock(return_value={"results": []})
client.get_emissions = AsyncMock(return_value={"results": []})
client.find_nearby = AsyncMock(return_value={"results": []})
client.get_summary = AsyncMock(return_value={"status": "ok"})
client.check_health = AsyncMock(return_value={"status": "healthy"})
client.close = AsyncMock()
return client
@pytest.fixture
def mock_client_with_data() -> MagicMock:
"""Create a mock Jana client with sample data responses."""
client = MagicMock(spec=JanaClient)
client.authenticate = AsyncMock(return_value="test-token")
# Air quality sample data
client.get_air_quality = AsyncMock(return_value={
"results": [
{
"value": 25.5,
"unit": "µg/m³",
"parameter": "pm25",
"datetime": "2024-01-15T12:00:00Z",
"location": {
"name": "Test Station",
"coordinates": {"latitude": 37.7, "longitude": -122.4},
},
}
]
})
# Emissions sample data
client.get_emissions = AsyncMock(return_value={
"results": [
{
"source": "climatetrace",
"emissions": 1000000,
"unit": "tonnes CO2e",
"gas": "co2",
"sector": "power",
"year": 2023,
}
]
})
# Nearby sample data
client.find_nearby = AsyncMock(return_value={
"results": [
{
"type": "station",
"name": "Nearby Station",
"distance_km": 5.2,
"source": "openaq",
}
]
})
client.get_summary = AsyncMock(return_value={
"total_records": 1000000,
"sources": {
"openaq": {"records": 500000},
"climatetrace": {"records": 300000},
"edgar": {"records": 200000},
},
})
client.check_health = AsyncMock(return_value={
"status": "healthy",
"backend": {"status": "ok", "response_time_ms": 50},
})
client.close = AsyncMock()
return client
@pytest.fixture(autouse=True)
def clear_settings_cache():
"""Clear settings cache before each test."""
from jana_mcp.config import get_settings
get_settings.cache_clear()
yield
get_settings.cache_clear()
@pytest.fixture
def clean_env():
"""Provide a clean environment without Jana variables."""
env_vars_to_remove = [
"JANA_BACKEND_URL",
"JANA_USERNAME",
"JANA_PASSWORD",
"JANA_TOKEN",
"JANA_TIMEOUT",
"MCP_SERVER_HOST",
"MCP_SERVER_PORT",
"LOG_LEVEL",
]
original_env = {k: os.environ.get(k) for k in env_vars_to_remove}
for var in env_vars_to_remove:
if var in os.environ:
del os.environ[var]
yield
# Restore original environment
for var, value in original_env.items():
if value is not None:
os.environ[var] = value
elif var in os.environ:
del os.environ[var]
