Spaces:
Sleeping
Sleeping
| """ | |
| Test script for Trips API endpoints. | |
| Tests trip start functionality with JWT authentication. | |
| """ | |
| import requests | |
| import time | |
| from generate_test_token import generate_test_token | |
| # Configuration | |
| BASE_URL = "http://localhost:8003" | |
| TRIPS_ENDPOINT = f"{BASE_URL}/tracker/trips" | |
| # Test data | |
| TEST_MERCHANT_ID = "550e8400-e29b-41d4-a716-446655440000" | |
| TEST_USER_ID = "660e8400-e29b-41d4-a716-446655440001" | |
| # Bangkok coordinates | |
| TEST_LOCATION = { | |
| "latitude": 13.7563, | |
| "longitude": 100.5018 | |
| } | |
| def test_start_trip(): | |
| """Test starting a new trip""" | |
| print("\n" + "="*60) | |
| print("TEST: Start Trip") | |
| print("="*60) | |
| # Generate JWT token | |
| token = generate_test_token( | |
| user_id=TEST_USER_ID, | |
| merchant_id=TEST_MERCHANT_ID | |
| ) | |
| headers = { | |
| "Authorization": f"Bearer {token}", | |
| "Content-Type": "application/json" | |
| } | |
| # Prepare request | |
| current_time = int(time.time() * 1000) | |
| payload = { | |
| "startedAt": current_time, | |
| "startLocation": TEST_LOCATION | |
| } | |
| print(f"\n📍 Starting trip at: {TEST_LOCATION}") | |
| print(f"⏰ Timestamp: {current_time}") | |
| # Make request | |
| response = requests.post( | |
| f"{TRIPS_ENDPOINT}/start", | |
| json=payload, | |
| headers=headers | |
| ) | |
| print(f"\n📊 Status Code: {response.status_code}") | |
| print(f"📄 Response: {response.json()}") | |
| if response.status_code == 201: | |
| print("\n✅ Trip started successfully!") | |
| trip_id = response.json().get("tripId") | |
| print(f"🆔 Trip ID: {trip_id}") | |
| return trip_id, current_time | |
| elif response.status_code == 409: | |
| print("\n⚠️ Active trip already exists") | |
| return None, None | |
| else: | |
| print(f"\n❌ Failed to start trip") | |
| return None, None | |
| def test_start_trip_duplicate(): | |
| """Test starting a trip when one is already active""" | |
| print("\n" + "="*60) | |
| print("TEST: Start Trip (Duplicate - Should Fail)") | |
| print("="*60) | |
| # Generate JWT token | |
| token = generate_test_token( | |
| user_id=TEST_USER_ID, | |
| merchant_id=TEST_MERCHANT_ID | |
| ) | |
| headers = { | |
| "Authorization": f"Bearer {token}", | |
| "Content-Type": "application/json" | |
| } | |
| # Prepare request | |
| current_time = int(time.time() * 1000) | |
| payload = { | |
| "startedAt": current_time, | |
| "startLocation": TEST_LOCATION | |
| } | |
| print(f"\n📍 Attempting to start another trip...") | |
| # Make request | |
| response = requests.post( | |
| f"{TRIPS_ENDPOINT}/start", | |
| json=payload, | |
| headers=headers | |
| ) | |
| print(f"\n📊 Status Code: {response.status_code}") | |
| print(f"📄 Response: {response.json()}") | |
| if response.status_code == 409: | |
| print("\n✅ Correctly rejected duplicate trip!") | |
| else: | |
| print(f"\n⚠️ Unexpected response") | |
| def test_start_trip_invalid_location(): | |
| """Test starting a trip with invalid coordinates""" | |
| print("\n" + "="*60) | |
| print("TEST: Start Trip (Invalid Location - Should Fail)") | |
| print("="*60) | |
| # Generate JWT token | |
| token = generate_test_token( | |
| user_id=TEST_USER_ID, | |
| merchant_id=TEST_MERCHANT_ID | |
| ) | |
| headers = { | |
| "Authorization": f"Bearer {token}", | |
| "Content-Type": "application/json" | |
| } | |
| # Invalid latitude (> 90) | |
| current_time = int(time.time() * 1000) | |
| payload = { | |
| "startedAt": current_time, | |
| "startLocation": { | |
| "latitude": 95.0, # Invalid | |
| "longitude": 100.5018 | |
| } | |
| } | |
| print(f"\n📍 Attempting with invalid latitude: 95.0") | |
| # Make request | |
| response = requests.post( | |
| f"{TRIPS_ENDPOINT}/start", | |
| json=payload, | |
| headers=headers | |
| ) | |
| print(f"\n📊 Status Code: {response.status_code}") | |
| print(f"📄 Response: {response.json()}") | |
| if response.status_code == 422: | |
| print("\n✅ Correctly rejected invalid coordinates!") | |
| else: | |
| print(f"\n⚠️ Unexpected response") | |
| def test_start_trip_no_auth(): | |
| """Test starting a trip without authentication""" | |
| print("\n" + "="*60) | |
| print("TEST: Start Trip (No Auth - Should Fail)") | |
| print("="*60) | |
| # No authorization header | |
| headers = { | |
| "Content-Type": "application/json" | |
| } | |
| current_time = int(time.time() * 1000) | |
| payload = { | |
| "startedAt": current_time, | |
| "startLocation": TEST_LOCATION | |
| } | |
| print(f"\n🔒 Attempting without JWT token...") | |
| # Make request | |
| response = requests.post( | |
| f"{TRIPS_ENDPOINT}/start", | |
| json=payload, | |
| headers=headers | |
| ) | |
| print(f"\n📊 Status Code: {response.status_code}") | |
| print(f"📄 Response: {response.json()}") | |
| if response.status_code == 401: | |
| print("\n✅ Correctly rejected unauthorized request!") | |
| else: | |
| print(f"\n⚠️ Unexpected response") | |
| def test_stop_trip(trip_id, start_time): | |
| """Test stopping an active trip""" | |
| print("\n" + "="*60) | |
| print("TEST: Stop Trip") | |
| print("="*60) | |
| if not trip_id: | |
| print("\n⚠️ No active trip to stop") | |
| return False | |
| # Generate JWT token | |
| token = generate_test_token( | |
| user_id=TEST_USER_ID, | |
| merchant_id=TEST_MERCHANT_ID | |
| ) | |
| headers = { | |
| "Authorization": f"Bearer {token}", | |
| "Content-Type": "application/json" | |
| } | |
| # Prepare request (simulate 15km trip over 30 minutes) | |
| current_time = int(time.time() * 1000) | |
| payload = { | |
| "tripId": trip_id, | |
| "stoppedAt": current_time, | |
| "distanceMeters": 15000, # 15km | |
| "endLocation": { | |
| "latitude": 13.7600, # Slightly different location | |
| "longitude": 100.5100 | |
| } | |
| } | |
| print(f"\n🆔 Trip ID: {trip_id}") | |
| print(f"📍 End location: {payload['endLocation']}") | |
| print(f"📏 Distance: {payload['distanceMeters']} meters") | |
| print(f"⏰ Stop timestamp: {current_time}") | |
| # Make request | |
| response = requests.post( | |
| f"{TRIPS_ENDPOINT}/stop", | |
| json=payload, | |
| headers=headers | |
| ) | |
| print(f"\n📊 Status Code: {response.status_code}") | |
| print(f"📄 Response: {response.json()}") | |
| if response.status_code == 200: | |
| print("\n✅ Trip stopped successfully!") | |
| result = response.json() | |
| print(f"📏 Distance: {result.get('distanceMeters')} meters") | |
| print(f"⏱️ Duration: {result.get('durationMinutes')} minutes") | |
| return True | |
| else: | |
| print(f"\n❌ Failed to stop trip") | |
| return False | |
| def test_stop_trip_duplicate(trip_id): | |
| """Test stopping an already stopped trip""" | |
| print("\n" + "="*60) | |
| print("TEST: Stop Trip (Duplicate - Should Fail)") | |
| print("="*60) | |
| if not trip_id: | |
| print("\n⚠️ No trip ID provided") | |
| return | |
| # Generate JWT token | |
| token = generate_test_token( | |
| user_id=TEST_USER_ID, | |
| merchant_id=TEST_MERCHANT_ID | |
| ) | |
| headers = { | |
| "Authorization": f"Bearer {token}", | |
| "Content-Type": "application/json" | |
| } | |
| current_time = int(time.time() * 1000) | |
| payload = { | |
| "tripId": trip_id, | |
| "stoppedAt": current_time, | |
| "distanceMeters": 15000, | |
| "endLocation": { | |
| "latitude": 13.7600, | |
| "longitude": 100.5100 | |
| } | |
| } | |
| print(f"\n🆔 Attempting to stop already stopped trip: {trip_id}") | |
| # Make request | |
| response = requests.post( | |
| f"{TRIPS_ENDPOINT}/stop", | |
| json=payload, | |
| headers=headers | |
| ) | |
| print(f"\n📊 Status Code: {response.status_code}") | |
| print(f"📄 Response: {response.json()}") | |
| if response.status_code == 400: | |
| print("\n✅ Correctly rejected duplicate stop!") | |
| else: | |
| print(f"\n⚠️ Unexpected response") | |
| def test_health_check(): | |
| """Test trip service health check""" | |
| print("\n" + "="*60) | |
| print("TEST: Health Check") | |
| print("="*60) | |
| response = requests.get(f"{TRIPS_ENDPOINT}/health") | |
| print(f"\n📊 Status Code: {response.status_code}") | |
| print(f"📄 Response: {response.json()}") | |
| if response.status_code == 200: | |
| print("\n✅ Service is healthy!") | |
| else: | |
| print(f"\n❌ Service health check failed") | |
| def main(): | |
| """Run all tests""" | |
| print("\n" + "="*60) | |
| print("🚀 TRIPS API TEST SUITE") | |
| print("="*60) | |
| print(f"Base URL: {BASE_URL}") | |
| print(f"Merchant ID: {TEST_MERCHANT_ID}") | |
| print(f"User ID: {TEST_USER_ID}") | |
| try: | |
| # Test 1: Health check | |
| test_health_check() | |
| # Test 2: Start trip (should succeed) | |
| trip_id, start_time = test_start_trip() | |
| # Test 3: Try to start another trip (should fail with 409) | |
| if trip_id: | |
| test_start_trip_duplicate() | |
| # Test 4: Invalid location (should fail with 422) | |
| test_start_trip_invalid_location() | |
| # Test 5: No authentication (should fail with 401) | |
| test_start_trip_no_auth() | |
| # Test 6: Stop trip (should succeed) | |
| if trip_id: | |
| # Wait 2 seconds to ensure duration > 0 | |
| print("\n⏳ Waiting 2 seconds before stopping trip...") | |
| time.sleep(2) | |
| stopped = test_stop_trip(trip_id, start_time) | |
| # Test 7: Try to stop again (should fail with 400) | |
| if stopped: | |
| test_stop_trip_duplicate(trip_id) | |
| print("\n" + "="*60) | |
| print("✅ ALL TESTS COMPLETED") | |
| print("="*60) | |
| except requests.exceptions.ConnectionError: | |
| print("\n❌ ERROR: Could not connect to server") | |
| print(f"Make sure the server is running at {BASE_URL}") | |
| except Exception as e: | |
| print(f"\n❌ ERROR: {str(e)}") | |
| if __name__ == "__main__": | |
| main() | |