93 lines
3.6 KiB
Python
93 lines
3.6 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
Test script to verify traffic data fetching and storage
|
||
|
|
"""
|
||
|
|
import asyncio
|
||
|
|
import sys
|
||
|
|
import os
|
||
|
|
from datetime import datetime, timedelta
|
||
|
|
|
||
|
|
# Add the services path to Python path
|
||
|
|
sys.path.insert(0, '/Users/urtzialfaro/Documents/bakery-ia/services/external')
|
||
|
|
|
||
|
|
async def test_traffic_storage():
|
||
|
|
"""Test traffic data fetching and storage"""
|
||
|
|
print("Testing traffic data storage workflow...")
|
||
|
|
|
||
|
|
try:
|
||
|
|
from app.services.traffic_service import TrafficService
|
||
|
|
from app.core.database import database_manager
|
||
|
|
from app.repositories.traffic_repository import TrafficRepository
|
||
|
|
|
||
|
|
# Initialize traffic service
|
||
|
|
traffic_service = TrafficService()
|
||
|
|
|
||
|
|
# Madrid coordinates (within bounds)
|
||
|
|
latitude = 40.4168
|
||
|
|
longitude = -3.7038
|
||
|
|
tenant_id = "test_tenant"
|
||
|
|
|
||
|
|
print(f"Testing location: {latitude}, {longitude}")
|
||
|
|
|
||
|
|
# Test 1: Check current traffic
|
||
|
|
print("\n1. Testing current traffic...")
|
||
|
|
current_traffic = await traffic_service.get_current_traffic(latitude, longitude, tenant_id)
|
||
|
|
if current_traffic:
|
||
|
|
print(f"Current traffic data retrieved:")
|
||
|
|
print(f" - Traffic volume: {current_traffic.get('traffic_volume')}")
|
||
|
|
print(f" - Source: {current_traffic.get('source')}")
|
||
|
|
print(f" - Congestion: {current_traffic.get('congestion_level')}")
|
||
|
|
else:
|
||
|
|
print("No current traffic data available")
|
||
|
|
|
||
|
|
# Test 2: Check historical traffic (short range to avoid large downloads)
|
||
|
|
print("\n2. Testing historical traffic storage...")
|
||
|
|
end_date = datetime.now()
|
||
|
|
start_date = end_date - timedelta(days=2) # Just 2 days to test
|
||
|
|
|
||
|
|
print(f"Requesting historical data from {start_date} to {end_date}")
|
||
|
|
|
||
|
|
historical_data = await traffic_service.get_historical_traffic(
|
||
|
|
latitude, longitude, start_date, end_date, tenant_id
|
||
|
|
)
|
||
|
|
|
||
|
|
print(f"Historical traffic records retrieved: {len(historical_data)}")
|
||
|
|
|
||
|
|
if historical_data:
|
||
|
|
print("Sample record:")
|
||
|
|
sample = historical_data[0]
|
||
|
|
for key, value in sample.items():
|
||
|
|
if key != 'raw_data': # Skip raw data to keep output clean
|
||
|
|
print(f" - {key}: {value}")
|
||
|
|
|
||
|
|
# Test 3: Check database storage
|
||
|
|
print("\n3. Checking database storage...")
|
||
|
|
async with database_manager.get_session() as session:
|
||
|
|
traffic_repo = TrafficRepository(session)
|
||
|
|
|
||
|
|
# Check what's actually stored in the database
|
||
|
|
db_records = await traffic_repo.get_by_location_and_date_range(
|
||
|
|
latitude, longitude, start_date, end_date, tenant_id
|
||
|
|
)
|
||
|
|
|
||
|
|
print(f"Records found in database: {len(db_records)}")
|
||
|
|
|
||
|
|
if db_records:
|
||
|
|
print("Sample database record:")
|
||
|
|
sample_db = db_records[0]
|
||
|
|
print(f" - ID: {sample_db.id}")
|
||
|
|
print(f" - Date: {sample_db.date}")
|
||
|
|
print(f" - Traffic volume: {sample_db.traffic_volume}")
|
||
|
|
print(f" - Location ID: {sample_db.location_id}")
|
||
|
|
print(f" - Source: {sample_db.source}")
|
||
|
|
print(f" - Tenant ID: {sample_db.tenant_id}")
|
||
|
|
|
||
|
|
print("\n✅ Traffic storage test completed successfully")
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f"\n❌ Traffic storage test failed: {e}")
|
||
|
|
import traceback
|
||
|
|
traceback.print_exc()
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
asyncio.run(test_traffic_storage())
|