oedb-backend/test_api_params.py

#!/usr/bin/env python3
# Script to test the API query parameters
# This script tests each query parameter to ensure it works as expected

import os
import sys
import json
import requests
from datetime import datetime, timedelta
import time

# Function to load environment variables from .env file
def load_env_from_file():
    if os.path.exists('.env'):
        print("Loading environment variables from .env file...")
        with open('.env', 'r') as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith('#'):
                    key, value = line.split('=', 1)
                    os.environ[key] = value

# Load environment variables from .env file
load_env_from_file()

# API base URL
API_BASE_URL = "http://localhost:8080"

# Current date (2025-09-15 as specified in the issue)
CURRENT_DATE = datetime(2025, 9, 15, 23, 0)

# Sample locations (Paris, Lyon, Marseille, Toulouse, Bordeaux, Lille)
SAMPLE_LOCATIONS = [
    {"name": "Paris", "lon": 2.3522, "lat": 48.8566},
    {"name": "Lyon", "lon": 4.8357, "lat": 45.7640},
    {"name": "Marseille", "lon": 5.3698, "lat": 43.2965},
    {"name": "Toulouse", "lon": 1.4442, "lat": 43.6047},
    {"name": "Bordeaux", "lon": -0.5792, "lat": 44.8378},
    {"name": "Lille", "lon": 3.0573, "lat": 50.6292}
]

# Event categories
EVENT_CATEGORIES = ["traffic", "nature", "weather", "sport", "conference", "party"]

# Function to make API request and handle errors
def make_api_request(endpoint, params=None):
    url = f"{API_BASE_URL}/{endpoint}"
    try:
        response = requests.get(url, params=params)
        response.raise_for_status()  # Raise exception for 4XX/5XX responses
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"Error making request to {url}: {e}")
        return None

# Function to print test results
def print_test_results(test_name, success, results=None, error=None):
    print(f"\n--- Test: {test_name} ---")
    if success:
        print("✅ SUCCESS")
        if results:
            if isinstance(results, dict) and 'features' in results:
                print(f"Found {len(results['features'])} events")
                for i, feature in enumerate(results['features'][:3]):  # Show first 3 events
                    props = feature['properties']
                    print(f"  {i+1}. {props.get('name', 'Unnamed')} ({props.get('id', 'No ID')})")
                if len(results['features']) > 3:
                    print(f"  ... and {len(results['features']) - 3} more events")
            else:
                print(json.dumps(results, indent=2))
    else:
        print("❌ FAILED")
        if error:
            print(f"Error: {error}")

# Test 1: Basic event retrieval (no parameters)
def test_basic_retrieval():
    print("\n=== Testing Basic Event Retrieval ===")
    results = make_api_request("event")
    success = results is not None and 'features' in results
    print_test_results("Basic Event Retrieval", success, results)
    return success

# Test 2: Filter by 'what' parameter
def test_what_parameter():
    print("\n=== Testing 'what' Parameter ===")
    all_success = True
    
    for category in EVENT_CATEGORIES:
        results = make_api_request("event", {"what": category})
        success = results is not None and 'features' in results
        
        # Verify that all returned events have the correct category
        if success and results['features']:
            for feature in results['features']:
                if 'events_what' in feature['properties'] and feature['properties']['events_what'] != category:
                    success = False
                    break
        
        print_test_results(f"Filter by what={category}", success, results)
        all_success = all_success and success
    
    return all_success

# Test 3: Filter by 'when' parameter
def test_when_parameter():
    print("\n=== Testing 'when' Parameter ===")
    all_success = True
    
    # Test with 'now'
    results = make_api_request("event", {"when": "now"})
    success = results is not None and 'features' in results
    print_test_results("Filter by when=now", success, results)
    all_success = all_success and success
    
    # Test with 'today'
    results = make_api_request("event", {"when": "today"})
    success = results is not None and 'features' in results
    print_test_results("Filter by when=today", success, results)
    all_success = all_success and success
    
    # Test with 'yesterday'
    results = make_api_request("event", {"when": "yesterday"})
    success = results is not None and 'features' in results
    print_test_results("Filter by when=yesterday", success, results)
    all_success = all_success and success
    
    # Test with 'tomorrow'
    results = make_api_request("event", {"when": "tomorrow"})
    success = results is not None and 'features' in results
    print_test_results("Filter by when=tomorrow", success, results)
    all_success = all_success and success
    
    # Test with 'lasthour'
    results = make_api_request("event", {"when": "lasthour"})
    success = results is not None and 'features' in results
    print_test_results("Filter by when=lasthour", success, results)
    all_success = all_success and success
    
    # Test with 'nexthour'
    results = make_api_request("event", {"when": "nexthour"})
    success = results is not None and 'features' in results
    print_test_results("Filter by when=nexthour", success, results)
    all_success = all_success and success
    
    # Test with ISO8601 date
    iso_date = CURRENT_DATE.isoformat()
    results = make_api_request("event", {"when": iso_date})
    success = results is not None and 'features' in results
    print_test_results(f"Filter by when={iso_date}", success, results)
    all_success = all_success and success
    
    return all_success

# Test 4: Filter by 'start' and 'stop' parameters
def test_start_stop_parameters():
    print("\n=== Testing 'start' and 'stop' Parameters ===")
    all_success = True
    
    # Test with start only
    start_date = (CURRENT_DATE - timedelta(days=10)).isoformat()
    results = make_api_request("event", {"start": start_date})
    success = results is not None and 'features' in results
    print_test_results(f"Filter by start={start_date}", success, results)
    all_success = all_success and success
    
    # Test with stop only
    stop_date = (CURRENT_DATE + timedelta(days=10)).isoformat()
    results = make_api_request("event", {"stop": stop_date})
    success = results is not None and 'features' in results
    print_test_results(f"Filter by stop={stop_date}", success, results)
    all_success = all_success and success
    
    # Test with both start and stop
    results = make_api_request("event", {"start": start_date, "stop": stop_date})
    success = results is not None and 'features' in results
    print_test_results(f"Filter by start={start_date} and stop={stop_date}", success, results)
    all_success = all_success and success
    
    # Test with relative keywords
    results = make_api_request("event", {"start": "last7days", "stop": "next30days"})
    success = results is not None and 'features' in results
    print_test_results("Filter by start=last7days and stop=next30days", success, results)
    all_success = all_success and success
    
    return all_success

# Test 5: Filter by 'bbox' parameter
def test_bbox_parameter():
    print("\n=== Testing 'bbox' Parameter ===")
    
    # Create a bounding box around France
    bbox = ["-5.0", "41.0", "10.0", "52.0"]  # [E, S, W, N]
    results = make_api_request("event", {"bbox": bbox})
    success = results is not None and 'features' in results
    print_test_results(f"Filter by bbox={','.join(bbox)}", success, results)
    
    return success

# Test 6: Filter by 'near' parameter
def test_near_parameter():
    print("\n=== Testing 'near' Parameter ===")
    all_success = True
    
    # Test with Paris coordinates
    paris = SAMPLE_LOCATIONS[0]
    near_params = [paris["lon"], paris["lat"], 50000]  # 50km radius
    results = make_api_request("event", {"near": near_params})
    success = results is not None and 'features' in results
    print_test_results(f"Filter by near={paris['lon']},{paris['lat']},50000", success, results)
    all_success = all_success and success
    
    # Test with Lyon coordinates
    lyon = SAMPLE_LOCATIONS[1]
    near_params = [lyon["lon"], lyon["lat"], 30000]  # 30km radius
    results = make_api_request("event", {"near": near_params})
    success = results is not None and 'features' in results
    print_test_results(f"Filter by near={lyon['lon']},{lyon['lat']},30000", success, results)
    all_success = all_success and success
    
    return all_success

# Test 7: Combine multiple parameters
def test_combined_parameters():
    print("\n=== Testing Combined Parameters ===")
    
    # Combine 'what' and 'when' parameters
    params = {
        "what": "traffic",
        "when": "now"
    }
    results = make_api_request("event", params)
    success = results is not None and 'features' in results
    print_test_results("Filter by what=traffic and when=now", success, results)
    
    return success

# Main function to run all tests
def main():
    print("Starting API parameter tests...")
    
    # Check if the server is running
    try:
        response = requests.get(f"{API_BASE_URL}/stats")
        if response.status_code != 200:
            print(f"❌ Server is not responding correctly. Status code: {response.status_code}")
            return False
        print("✅ Server is running and responding to requests.")
    except requests.exceptions.ConnectionError:
        print("❌ Cannot connect to the server. Make sure it's running on http://localhost:8080")
        return False
    
    # Run all tests
    tests = [
        ("Basic Event Retrieval", test_basic_retrieval),
        ("'what' Parameter", test_what_parameter),
        ("'when' Parameter", test_when_parameter),
        ("'start' and 'stop' Parameters", test_start_stop_parameters),
        ("'bbox' Parameter", test_bbox_parameter),
        ("'near' Parameter", test_near_parameter),
        ("Combined Parameters", test_combined_parameters)
    ]
    
    results = {}
    all_success = True
    
    for test_name, test_func in tests:
        print(f"\n{'=' * 50}")
        print(f"Running test: {test_name}")
        try:
            success = test_func()
            results[test_name] = success
            all_success = all_success and success
        except Exception as e:
            print(f"❌ Test failed with exception: {e}")
            results[test_name] = False
            all_success = False
    
    # Print summary
    print("\n\n" + "=" * 50)
    print("TEST SUMMARY")
    print("=" * 50)
    for test_name, success in results.items():
        status = "✅ PASSED" if success else "❌ FAILED"
        print(f"{test_name}: {status}")
    
    overall_status = "✅ ALL TESTS PASSED" if all_success else "❌ SOME TESTS FAILED"
    print(f"\nOverall: {overall_status}")
    
    return all_success

if __name__ == "__main__":
    main()