115 lines
4.0 KiB
Python
115 lines
4.0 KiB
Python
# ================================================================
|
|
# Integration Tests: tests/integration/test_forecasting_flow.py
|
|
# ================================================================
|
|
"""
|
|
Integration tests for complete forecasting flow
|
|
"""
|
|
|
|
import pytest
|
|
import httpx
|
|
import asyncio
|
|
from datetime import date, timedelta
|
|
import json
|
|
|
|
class TestForecastingFlow:
|
|
"""Test complete forecasting workflow"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_complete_forecast_flow(self):
|
|
"""Test complete flow from training to forecasting"""
|
|
|
|
base_url = "http://localhost:8000" # API Gateway
|
|
|
|
# Test data
|
|
tenant_id = "test-tenant-123"
|
|
product_name = "Pan Integral"
|
|
location = "madrid_centro"
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
# 1. Check if model exists
|
|
model_response = await client.get(
|
|
f"{base_url}/api/v1/training/models/latest",
|
|
params={
|
|
"tenant_id": tenant_id,
|
|
"product_name": product_name,
|
|
"location": location
|
|
}
|
|
)
|
|
|
|
# 2. Generate forecast
|
|
forecast_request = {
|
|
"tenant_id": tenant_id,
|
|
"product_name": product_name,
|
|
"location": location,
|
|
"forecast_date": (date.today() + timedelta(days=1)).isoformat(),
|
|
"business_type": "individual",
|
|
"include_weather": True,
|
|
"include_traffic": True,
|
|
"confidence_level": 0.8
|
|
}
|
|
|
|
forecast_response = await client.post(
|
|
f"{base_url}/api/v1/forecasting/single",
|
|
json=forecast_request
|
|
)
|
|
|
|
assert forecast_response.status_code == 200
|
|
forecast_data = forecast_response.json()
|
|
|
|
# Verify forecast structure
|
|
assert "id" in forecast_data
|
|
assert "predicted_demand" in forecast_data
|
|
assert "confidence_lower" in forecast_data
|
|
assert "confidence_upper" in forecast_data
|
|
assert forecast_data["product_name"] == product_name
|
|
|
|
# 3. Get forecast list
|
|
list_response = await client.get(
|
|
f"{base_url}/api/v1/forecasting/list",
|
|
params={"location": location}
|
|
)
|
|
|
|
assert list_response.status_code == 200
|
|
forecasts = list_response.json()
|
|
assert len(forecasts) > 0
|
|
|
|
# 4. Check for alerts
|
|
alerts_response = await client.get(
|
|
f"{base_url}/api/v1/forecasting/alerts"
|
|
)
|
|
|
|
assert alerts_response.status_code == 200
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_batch_forecasting(self):
|
|
"""Test batch forecasting functionality"""
|
|
|
|
base_url = "http://localhost:8000"
|
|
|
|
batch_request = {
|
|
"tenant_id": "test-tenant-123",
|
|
"batch_name": "Weekly Forecast Batch",
|
|
"products": ["Pan Integral", "Croissant", "Café con Leche"],
|
|
"location": "madrid_centro",
|
|
"forecast_days": 7,
|
|
"business_type": "individual",
|
|
"include_weather": True,
|
|
"include_traffic": True,
|
|
"confidence_level": 0.8
|
|
}
|
|
|
|
async with httpx.AsyncClient(timeout=60.0) as client:
|
|
response = await client.post(
|
|
f"{base_url}/api/v1/forecasting/batch",
|
|
json=batch_request
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
batch_data = response.json()
|
|
|
|
assert "id" in batch_data
|
|
assert batch_data["batch_name"] == "Weekly Forecast Batch"
|
|
assert batch_data["total_products"] == 21 # 3 products * 7 days
|
|
assert batch_data["status"] in ["completed", "partial"]
|
|
|