test_location_api.py

#!/usr/bin/env python3
"""
Test script for location tracking API
"""

import requests
import time
from generate_test_token import generate_test_token

# Configuration
BASE_URL = "http://localhost:8003"
MERCHANT_ID = "550e8400-e29b-41d4-a716-446655440000"
EMPLOYEE_ID = "660e8400-e29b-41d4-a716-446655440001"

def test_location_upload():
    """Test batch location upload endpoint"""
    
    print("="*70)
    print("🧪 Testing Location Tracking API")
    print("="*70)
    print()
    
    # Generate JWT token
    print("🔑 Generating JWT token...")
    token = generate_test_token()
    print(f"✅ Token generated\n")
    
    # Prepare headers
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    # Test 1: Health check
    print("1️⃣  Testing health endpoint...")
    response = requests.get(f"{BASE_URL}/tracker/tracking/health")
    print(f"   Status: {response.status_code}")
    print(f"   Response: {response.json()}\n")
    
    # Test 2: Upload single location point
    print("2️⃣  Testing single location point upload...")
    current_time_ms = int(time.time() * 1000)
    print(f"   Response: {current_time_ms}\n")
    single_point_data = {
        "points": [
            {
                "latitude": 40.7589,
                "longitude": -73.9851,
                "timestamp": current_time_ms,
                "accuracy": 15.5,
                "speed": 3.2,
                "heading": 90.0,
                "batteryLevel": 75
            }
        ]
    }
    
    response = requests.post(
        f"{BASE_URL}/tracker/tracking/points",
        json=single_point_data,
        headers=headers
    )
    
    print(f"   Status: {response.status_code}")
    print(f"   Response: {response.json()}\n")
    
    # Test 3: Upload batch of location points
    print("3️⃣  Testing batch location upload (5 points)...")
    
    batch_points = []
    for i in range(5):
        batch_points.append({
            "latitude": 40.7589 + (i * 0.001),
            "longitude": -73.9851 + (i * 0.001),
            "timestamp": current_time_ms + (i * 60000),  # 1 minute apart
            "accuracy": 10.0 + i,
            "speed": 2.5 + (i * 0.5),
            "heading": 90.0 + (i * 10),
            "batteryLevel": 75 - i
        })
    
    batch_data = {"points": batch_points}
    
    response = requests.post(
        f"{BASE_URL}/tracker/tracking/points",
        json=batch_data,
        headers=headers
    )
    
    print(f"   Status: {response.status_code}")
    print(f"   Response: {response.json()}\n")
    
    # Test 4: Invalid data - missing required field
    print("4️⃣  Testing validation (missing latitude)...")
    
    invalid_data = {
        "points": [
            {
                "longitude": -73.9851,
                "timestamp": current_time_ms
            }
        ]
    }
    
    response = requests.post(
        f"{BASE_URL}/tracker/tracking/points",
        json=invalid_data,
        headers=headers
    )
    
    print(f"   Status: {response.status_code}")
    print(f"   Response: {response.json()}\n")
    
    # Test 5: Invalid data - out of order timestamps
    print("5️⃣  Testing validation (out of order timestamps)...")
    
    out_of_order_data = {
        "points": [
            {
                "latitude": 40.7589,
                "longitude": -73.9851,
                "timestamp": current_time_ms
            },
            {
                "latitude": 40.7590,
                "longitude": -73.9852,
                "timestamp": current_time_ms - 60000  # Earlier timestamp
            }
        ]
    }
    
    response = requests.post(
        f"{BASE_URL}/tracker/tracking/points",
        json=out_of_order_data,
        headers=headers
    )
    
    print(f"   Status: {response.status_code}")
    print(f"   Response: {response.json()}\n")
    
    # Test 6: Unauthorized request (no token)
    print("6️⃣  Testing authentication (no token)...")
    
    response = requests.post(
        f"{BASE_URL}/tracker/tracking/points",
        json=single_point_data
    )
    
    print(f"   Status: {response.status_code}")
    print(f"   Response: {response.json()}\n")
    
    print("="*70)
    print("✅ Location Tracking API Tests Complete!")
    print("="*70)
    print()
    print("📊 Summary:")
    print("   • Health check: Working")
    print("   • Single point upload: Working")
    print("   • Batch upload: Working")
    print("   • Validation: Working")
    print("   • Authentication: Working")
    print()
    print("💡 Next steps:")
    print("   1. Check database: SELECT * FROM trans.scm_location_points;")
    print("   2. Verify TimescaleDB chunks are created")
    print("   3. Test with mobile app")


if __name__ == "__main__":
    try:
        test_location_upload()
    except requests.exceptions.ConnectionError:
        print("❌ Error: Could not connect to server")
        print("   Make sure the server is running: python3 -m uvicorn app.main:app --port 8003")
    except Exception as e:
        print(f"❌ Error: {e}")