95 lines
3.1 KiB
Python
95 lines
3.1 KiB
Python
# ================================================================
|
|
# services/data/tests/test_data.py
|
|
# ================================================================
|
|
"""Integration tests for data service"""
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from datetime import datetime, timedelta
|
|
import uuid
|
|
|
|
from app.services.sales_service import SalesService
|
|
from app.services.weather_service import WeatherService
|
|
from app.services.traffic_service import TrafficService
|
|
from app.schemas.sales import SalesDataCreate, SalesDataQuery
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_sales_record(db: AsyncSession, test_tenant_id, test_sales_data):
|
|
"""Test creating a sales record"""
|
|
sales_data = SalesDataCreate(
|
|
tenant_id=test_tenant_id,
|
|
**test_sales_data
|
|
)
|
|
|
|
record = await SalesService.create_sales_record(sales_data, db)
|
|
|
|
assert record.id is not None
|
|
assert record.tenant_id == test_tenant_id
|
|
assert record.product_name == test_sales_data["product_name"]
|
|
assert record.quantity_sold == test_sales_data["quantity_sold"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_query_sales_data(db: AsyncSession, test_tenant_id, test_sales_data):
|
|
"""Test querying sales data"""
|
|
# Create test record
|
|
sales_data = SalesDataCreate(
|
|
tenant_id=test_tenant_id,
|
|
**test_sales_data
|
|
)
|
|
await SalesService.create_sales_record(sales_data, db)
|
|
|
|
# Query data
|
|
query = SalesDataQuery(
|
|
tenant_id=test_tenant_id,
|
|
start_date=datetime.now() - timedelta(days=1),
|
|
end_date=datetime.now() + timedelta(days=1)
|
|
)
|
|
|
|
records = await SalesService.get_sales_data(query, db)
|
|
|
|
assert len(records) == 1
|
|
assert records[0].product_name == test_sales_data["product_name"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_csv_import():
|
|
"""Test CSV data import"""
|
|
csv_content = """date,product,quantity,revenue
|
|
2024-01-15,Pan Integral,25,37.50
|
|
2024-01-15,Croissant,15,22.50"""
|
|
|
|
# This would normally use the database, but we're testing the parsing logic
|
|
import io
|
|
import pandas as pd
|
|
|
|
csv_file = io.StringIO(csv_content)
|
|
df = pd.read_csv(csv_file)
|
|
|
|
assert len(df) == 2
|
|
assert df.iloc[0]['product'] == 'Pan Integral'
|
|
assert df.iloc[0]['quantity'] == 25
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_weather_service():
|
|
"""Test weather service"""
|
|
weather_service = WeatherService()
|
|
|
|
# Test current weather (should return synthetic data)
|
|
current = await weather_service.get_current_weather(40.4168, -3.7038)
|
|
|
|
assert current is not None
|
|
assert current.temperature is not None
|
|
assert current.source in ['aemet', 'synthetic']
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_traffic_service():
|
|
"""Test traffic service"""
|
|
traffic_service = TrafficService()
|
|
|
|
# Test current traffic (should return synthetic data)
|
|
current = await traffic_service.get_current_traffic(40.4168, -3.7038)
|
|
|
|
assert current is not None
|
|
assert current.traffic_volume is not None
|
|
assert current.congestion_level in ['low', 'medium', 'high']
|