314 lines
9.1 KiB
Python
314 lines
9.1 KiB
Python
# services/external/tests/conftest.py
|
|
"""
|
|
Pytest configuration and fixtures for External Service tests
|
|
"""
|
|
|
|
import pytest
|
|
import asyncio
|
|
from datetime import datetime, timezone
|
|
from typing import AsyncGenerator
|
|
from uuid import uuid4, UUID
|
|
|
|
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
from app.core.config import settings
|
|
from app.core.database import Base, get_db
|
|
from app.models.weather import WeatherData, WeatherStation
|
|
from app.models.traffic import TrafficData, TrafficMeasurementPoint
|
|
|
|
|
|
# Test database configuration
|
|
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def event_loop():
|
|
"""Create event loop for the test session"""
|
|
loop = asyncio.new_event_loop()
|
|
yield loop
|
|
loop.close()
|
|
|
|
|
|
@pytest.fixture
|
|
async def test_engine():
|
|
"""Create test database engine"""
|
|
engine = create_async_engine(
|
|
TEST_DATABASE_URL,
|
|
poolclass=StaticPool,
|
|
connect_args={"check_same_thread": False}
|
|
)
|
|
|
|
# Create tables
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
yield engine
|
|
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.fixture
|
|
async def test_db_session(test_engine) -> AsyncGenerator[AsyncSession, None]:
|
|
"""Create test database session"""
|
|
async_session = async_sessionmaker(
|
|
test_engine, class_=AsyncSession, expire_on_commit=False
|
|
)
|
|
|
|
async with async_session() as session:
|
|
yield session
|
|
|
|
|
|
@pytest.fixture
|
|
def test_client():
|
|
"""Create test client"""
|
|
return TestClient(app)
|
|
|
|
|
|
@pytest.fixture
|
|
async def override_get_db(test_db_session):
|
|
"""Override get_db dependency for testing"""
|
|
async def _override_get_db():
|
|
yield test_db_session
|
|
|
|
app.dependency_overrides[get_db] = _override_get_db
|
|
yield
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
# Test data fixtures
|
|
@pytest.fixture
|
|
def sample_tenant_id() -> UUID:
|
|
"""Sample tenant ID for testing"""
|
|
return uuid4()
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_weather_data() -> dict:
|
|
"""Sample weather data for testing"""
|
|
return {
|
|
"city": "madrid",
|
|
"location_id": "40.4168,-3.7038",
|
|
"date": datetime.now(timezone.utc),
|
|
"temperature": 18.5,
|
|
"humidity": 65.0,
|
|
"pressure": 1013.2,
|
|
"wind_speed": 10.2,
|
|
"condition": "partly_cloudy",
|
|
"description": "Parcialmente nublado",
|
|
"source": "aemet",
|
|
"data_type": "current",
|
|
"is_forecast": False,
|
|
"data_quality_score": 95.0
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_traffic_data() -> dict:
|
|
"""Sample traffic data for testing"""
|
|
return {
|
|
"city": "madrid",
|
|
"location_id": "PM_M30_001",
|
|
"date": datetime.now(timezone.utc),
|
|
"measurement_point_id": "PM_M30_001",
|
|
"measurement_point_name": "M-30 Norte - Nudo Norte",
|
|
"measurement_point_type": "M30",
|
|
"traffic_volume": 850,
|
|
"average_speed": 65.2,
|
|
"congestion_level": "medium",
|
|
"occupation_percentage": 45.8,
|
|
"latitude": 40.4501,
|
|
"longitude": -3.6919,
|
|
"district": "Chamartín",
|
|
"source": "madrid_opendata",
|
|
"data_quality_score": 92.0,
|
|
"is_synthetic": False
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_weather_forecast() -> list[dict]:
|
|
"""Sample weather forecast data"""
|
|
base_date = datetime.now(timezone.utc)
|
|
return [
|
|
{
|
|
"city": "madrid",
|
|
"location_id": "40.4168,-3.7038",
|
|
"date": base_date,
|
|
"forecast_date": base_date,
|
|
"temperature": 20.0,
|
|
"temperature_min": 15.0,
|
|
"temperature_max": 25.0,
|
|
"precipitation": 0.0,
|
|
"humidity": 60.0,
|
|
"wind_speed": 12.0,
|
|
"condition": "sunny",
|
|
"description": "Soleado",
|
|
"source": "aemet",
|
|
"data_type": "forecast",
|
|
"is_forecast": True,
|
|
"data_quality_score": 85.0
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
async def populated_weather_db(test_db_session: AsyncSession, sample_weather_data: dict):
|
|
"""Database populated with weather test data"""
|
|
weather_record = WeatherData(**sample_weather_data)
|
|
test_db_session.add(weather_record)
|
|
await test_db_session.commit()
|
|
yield test_db_session
|
|
|
|
|
|
@pytest.fixture
|
|
async def populated_traffic_db(test_db_session: AsyncSession, sample_traffic_data: dict):
|
|
"""Database populated with traffic test data"""
|
|
traffic_record = TrafficData(**sample_traffic_data)
|
|
test_db_session.add(traffic_record)
|
|
await test_db_session.commit()
|
|
yield test_db_session
|
|
|
|
|
|
# Mock external API fixtures
|
|
@pytest.fixture
|
|
def mock_aemet_response():
|
|
"""Mock AEMET API response"""
|
|
return {
|
|
"date": datetime.now(timezone.utc),
|
|
"temperature": 18.5,
|
|
"humidity": 65.0,
|
|
"pressure": 1013.2,
|
|
"wind_speed": 10.2,
|
|
"description": "Parcialmente nublado",
|
|
"source": "aemet"
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_madrid_traffic_xml():
|
|
"""Mock Madrid Open Data traffic XML"""
|
|
return """<?xml version="1.0" encoding="UTF-8"?>
|
|
<pms>
|
|
<pm codigo="PM_M30_001" nombre="M-30 Norte - Nudo Norte">
|
|
<intensidad>850</intensidad>
|
|
<ocupacion>45</ocupacion>
|
|
<velocidad>65</velocidad>
|
|
<fechahora>2024-01-15T10:30:00</fechahora>
|
|
</pm>
|
|
<pm codigo="PM_URB_002" nombre="Gran Vía - Plaza España">
|
|
<intensidad>320</intensidad>
|
|
<ocupacion>78</ocupacion>
|
|
<velocidad>25</velocidad>
|
|
<fechahora>2024-01-15T10:30:00</fechahora>
|
|
</pm>
|
|
</pms>"""
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_messaging():
|
|
"""Mock messaging service"""
|
|
class MockMessaging:
|
|
def __init__(self):
|
|
self.published_events = []
|
|
|
|
async def publish_weather_updated(self, data):
|
|
self.published_events.append(("weather_updated", data))
|
|
return True
|
|
|
|
async def publish_traffic_updated(self, data):
|
|
self.published_events.append(("traffic_updated", data))
|
|
return True
|
|
|
|
async def publish_collection_job_started(self, data):
|
|
self.published_events.append(("job_started", data))
|
|
return True
|
|
|
|
async def publish_collection_job_completed(self, data):
|
|
self.published_events.append(("job_completed", data))
|
|
return True
|
|
|
|
return MockMessaging()
|
|
|
|
|
|
# Mock external clients
|
|
@pytest.fixture
|
|
def mock_aemet_client():
|
|
"""Mock AEMET client"""
|
|
class MockAEMETClient:
|
|
async def get_current_weather(self, lat, lon):
|
|
return {
|
|
"date": datetime.now(timezone.utc),
|
|
"temperature": 18.5,
|
|
"humidity": 65.0,
|
|
"pressure": 1013.2,
|
|
"wind_speed": 10.2,
|
|
"description": "Parcialmente nublado",
|
|
"source": "aemet"
|
|
}
|
|
|
|
async def get_forecast(self, lat, lon, days):
|
|
return [
|
|
{
|
|
"forecast_date": datetime.now(timezone.utc),
|
|
"temperature": 20.0,
|
|
"temperature_min": 15.0,
|
|
"temperature_max": 25.0,
|
|
"precipitation": 0.0,
|
|
"humidity": 60.0,
|
|
"wind_speed": 12.0,
|
|
"description": "Soleado",
|
|
"source": "aemet"
|
|
}
|
|
]
|
|
|
|
return MockAEMETClient()
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_madrid_client():
|
|
"""Mock Madrid traffic client"""
|
|
class MockMadridClient:
|
|
async def fetch_current_traffic_xml(self):
|
|
return """<?xml version="1.0" encoding="UTF-8"?>
|
|
<pms>
|
|
<pm codigo="PM_TEST_001" nombre="Test Point">
|
|
<intensidad>500</intensidad>
|
|
<ocupacion>50</ocupacion>
|
|
<velocidad>50</velocidad>
|
|
<fechahora>2024-01-15T10:30:00</fechahora>
|
|
</pm>
|
|
</pms>"""
|
|
|
|
return MockMadridClient()
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_madrid_processor():
|
|
"""Mock Madrid traffic processor"""
|
|
class MockMadridProcessor:
|
|
async def process_current_traffic_xml(self, xml_content):
|
|
return [
|
|
{
|
|
"city": "madrid",
|
|
"location_id": "PM_TEST_001",
|
|
"date": datetime.now(timezone.utc),
|
|
"measurement_point_id": "PM_TEST_001",
|
|
"measurement_point_name": "Test Point",
|
|
"measurement_point_type": "TEST",
|
|
"traffic_volume": 500,
|
|
"average_speed": 50.0,
|
|
"congestion_level": "medium",
|
|
"occupation_percentage": 50.0,
|
|
"latitude": 40.4168,
|
|
"longitude": -3.7038,
|
|
"district": "Centro",
|
|
"source": "madrid_opendata",
|
|
"data_quality_score": 90.0,
|
|
"is_synthetic": False
|
|
}
|
|
]
|
|
|
|
return MockMadridProcessor() |