136 lines
5.1 KiB
Python
136 lines
5.1 KiB
Python
# ================================================================
|
|
# services/forecasting/tests/test_forecasting.py
|
|
# ================================================================
|
|
"""
|
|
Tests for forecasting service
|
|
"""
|
|
|
|
import pytest
|
|
import asyncio
|
|
from datetime import date, datetime, timedelta
|
|
from unittest.mock import Mock, AsyncMock, patch
|
|
import uuid
|
|
|
|
from app.services.forecasting_service import ForecastingService
|
|
from app.schemas.forecasts import ForecastRequest, BusinessType
|
|
from app.models.forecasts import Forecast
|
|
|
|
|
|
class TestForecastingService:
|
|
"""Test cases for ForecastingService"""
|
|
|
|
@pytest.fixture
|
|
def forecasting_service(self):
|
|
return ForecastingService()
|
|
|
|
@pytest.fixture
|
|
def sample_forecast_request(self):
|
|
return ForecastRequest(
|
|
tenant_id=str(uuid.uuid4()),
|
|
product_name="Pan Integral",
|
|
location="madrid_centro",
|
|
forecast_date=date.today() + timedelta(days=1),
|
|
business_type=BusinessType.INDIVIDUAL,
|
|
include_weather=True,
|
|
include_traffic=True,
|
|
confidence_level=0.8
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_forecast_success(self, forecasting_service, sample_forecast_request):
|
|
"""Test successful forecast generation"""
|
|
|
|
# Mock database session
|
|
mock_db = AsyncMock()
|
|
|
|
# Mock external dependencies
|
|
with patch.object(forecasting_service, '_get_latest_model') as mock_get_model, \
|
|
patch.object(forecasting_service, '_prepare_forecast_features') as mock_prepare_features, \
|
|
patch.object(forecasting_service.prediction_service, 'predict') as mock_predict, \
|
|
patch.object(forecasting_service, '_check_and_create_alerts') as mock_check_alerts:
|
|
|
|
# Setup mocks
|
|
mock_get_model.return_value = {
|
|
"model_id": str(uuid.uuid4()),
|
|
"version": "1.0.0",
|
|
"algorithm": "prophet"
|
|
}
|
|
|
|
mock_prepare_features.return_value = {
|
|
"date": "2024-01-16",
|
|
"day_of_week": 1,
|
|
"is_weekend": False,
|
|
"is_holiday": False,
|
|
"temperature": 15.0,
|
|
"precipitation": 0.0
|
|
}
|
|
|
|
mock_predict.return_value = {
|
|
"demand": 85.5,
|
|
"lower_bound": 70.2,
|
|
"upper_bound": 100.8
|
|
}
|
|
|
|
# Execute test
|
|
result = await forecasting_service.generate_forecast(sample_forecast_request, mock_db)
|
|
|
|
# Assertions
|
|
assert isinstance(result, Forecast)
|
|
assert result.product_name == "Pan Integral"
|
|
assert result.predicted_demand == 85.5
|
|
assert result.confidence_lower == 70.2
|
|
assert result.confidence_upper == 100.8
|
|
|
|
# Verify mocks were called
|
|
mock_get_model.assert_called_once()
|
|
mock_prepare_features.assert_called_once()
|
|
mock_predict.assert_called_once()
|
|
mock_check_alerts.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_forecast_no_model(self, forecasting_service, sample_forecast_request):
|
|
"""Test forecast generation when no model is found"""
|
|
|
|
mock_db = AsyncMock()
|
|
|
|
with patch.object(forecasting_service, '_get_latest_model') as mock_get_model:
|
|
mock_get_model.return_value = None
|
|
|
|
# Should raise ValueError
|
|
with pytest.raises(ValueError, match="No trained model found"):
|
|
await forecasting_service.generate_forecast(sample_forecast_request, mock_db)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prepare_forecast_features(self, forecasting_service, sample_forecast_request):
|
|
"""Test feature preparation for forecasting"""
|
|
|
|
with patch.object(forecasting_service, '_is_spanish_holiday') as mock_holiday, \
|
|
patch.object(forecasting_service, '_get_weather_forecast') as mock_weather, \
|
|
patch.object(forecasting_service, '_get_traffic_forecast') as mock_traffic:
|
|
|
|
# Setup mocks
|
|
mock_holiday.return_value = False
|
|
mock_weather.return_value = {
|
|
"temperature": 18.5,
|
|
"precipitation": 0.0,
|
|
"humidity": 65.0,
|
|
"weather_description": "Clear"
|
|
}
|
|
mock_traffic.return_value = {
|
|
"traffic_volume": 1200,
|
|
"pedestrian_count": 850
|
|
}
|
|
|
|
# Execute test
|
|
features = await forecasting_service._prepare_forecast_features(sample_forecast_request)
|
|
|
|
# Assertions
|
|
assert "date" in features
|
|
assert "day_of_week" in features
|
|
assert "is_weekend" in features
|
|
assert "is_holiday" in features
|
|
assert features["business_type"] == "individual"
|
|
assert features["temperature"] == 18.5
|
|
assert features["traffic_volume"] == 1200
|
|
|