mirror of
https://github.com/pacnpal/thrillwiki_django_no_react.git
synced 2025-12-20 13:31:08 -05:00
Add comprehensive test scripts for various models and services
- Implement tests for RideLocation and CompanyHeadquarters models to verify functionality and data integrity. - Create a manual trigger test script for trending content calculation endpoint, including authentication and unauthorized access tests. - Develop a manufacturer sync test to ensure ride manufacturers are correctly associated with ride models. - Add tests for ParkLocation model, including coordinate setting and distance calculations between parks. - Implement a RoadTripService test suite covering geocoding, route calculation, park discovery, and error handling. - Create a unified map service test script to validate map functionality, API endpoints, and performance metrics.
This commit is contained in:
pacnpal
182
tests/test_manual_trigger.py
Normal file
182
tests/test_manual_trigger.py
Normal file
@@ -0,0 +1,182 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test script for the manual trending content calculation trigger endpoint.
|
||||
"""
|
||||
|
||||
import requests
|
||||
import json
|
||||
import time
|
||||
from datetime import datetime
|
||||
|
||||
# Configuration
|
||||
BASE_URL = "http://localhost:8000"
|
||||
ADMIN_USERNAME = "admin"
|
||||
ADMIN_PASSWORD = "admin" # We'll need to check what the password is
|
||||
|
||||
|
||||
def login_and_get_token():
|
||||
"""Login and get authentication token."""
|
||||
login_url = f"{BASE_URL}/api/v1/auth/login/"
|
||||
|
||||
login_data = {
|
||||
"username": ADMIN_USERNAME,
|
||||
"password": ADMIN_PASSWORD
|
||||
}
|
||||
|
||||
print(f"🔐 Attempting to login as {ADMIN_USERNAME}...")
|
||||
response = requests.post(login_url, json=login_data)
|
||||
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
token = data.get('token')
|
||||
print(f"✅ Login successful! Token: {token[:20]}...")
|
||||
return token
|
||||
else:
|
||||
print(f"❌ Login failed: {response.status_code}")
|
||||
print(f"Response: {response.text}")
|
||||
return None
|
||||
|
||||
|
||||
def test_trigger_endpoint(token):
|
||||
"""Test the manual trigger endpoint."""
|
||||
trigger_url = f"{BASE_URL}/api/v1/trending/calculate/"
|
||||
|
||||
headers = {
|
||||
"Authorization": f"Bearer {token}",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
|
||||
print(f"\n🚀 Testing manual trigger endpoint...")
|
||||
print(f"URL: {trigger_url}")
|
||||
|
||||
response = requests.post(trigger_url, headers=headers)
|
||||
|
||||
print(f"Status Code: {response.status_code}")
|
||||
print(f"Response Headers: {dict(response.headers)}")
|
||||
|
||||
try:
|
||||
response_data = response.json()
|
||||
print(f"Response Body: {json.dumps(response_data, indent=2)}")
|
||||
|
||||
if response.status_code == 202:
|
||||
print("✅ Manual trigger successful!")
|
||||
return response_data
|
||||
else:
|
||||
print(f"❌ Manual trigger failed with status {response.status_code}")
|
||||
return None
|
||||
|
||||
except json.JSONDecodeError:
|
||||
print(f"❌ Invalid JSON response: {response.text}")
|
||||
return None
|
||||
|
||||
|
||||
def test_trending_endpoints():
|
||||
"""Test the trending content endpoints to see the results."""
|
||||
print(f"\n📊 Testing trending content endpoints...")
|
||||
|
||||
# Test trending content endpoint
|
||||
trending_url = f"{BASE_URL}/api/v1/trending/content/"
|
||||
print(f"Testing: {trending_url}")
|
||||
|
||||
response = requests.get(trending_url)
|
||||
print(f"Trending Content Status: {response.status_code}")
|
||||
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
print(f"Trending Parks: {len(data.get('trending_parks', []))}")
|
||||
print(f"Trending Rides: {len(data.get('trending_rides', []))}")
|
||||
|
||||
# Show first few trending items
|
||||
if data.get('trending_parks'):
|
||||
print(
|
||||
f"First trending park: {data['trending_parks'][0].get('name', 'Unknown')}")
|
||||
if data.get('trending_rides'):
|
||||
print(
|
||||
f"First trending ride: {data['trending_rides'][0].get('name', 'Unknown')}")
|
||||
|
||||
# Test new content endpoint
|
||||
new_content_url = f"{BASE_URL}/api/v1/trending/new/"
|
||||
print(f"\nTesting: {new_content_url}")
|
||||
|
||||
response = requests.get(new_content_url)
|
||||
print(f"New Content Status: {response.status_code}")
|
||||
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
print(f"Recently Added: {len(data.get('recently_added', []))}")
|
||||
print(f"Newly Opened: {len(data.get('newly_opened', []))}")
|
||||
print(f"Upcoming: {len(data.get('upcoming', []))}")
|
||||
|
||||
# Show the newly_opened structure to verify our changes
|
||||
if data.get('newly_opened'):
|
||||
print(f"\n🎢 First newly opened item structure:")
|
||||
first_item = data['newly_opened'][0]
|
||||
print(f" Name: {first_item.get('name')}")
|
||||
# Should be park name, not location
|
||||
print(f" Park: {first_item.get('park')}")
|
||||
# Should be date_opened, not location
|
||||
print(f" Date Opened: {first_item.get('date_opened')}")
|
||||
print(f" Category: {first_item.get('category')}")
|
||||
print(f" Slug: {first_item.get('slug')}")
|
||||
|
||||
# Verify location field is NOT present
|
||||
if 'location' in first_item:
|
||||
print(
|
||||
f" ❌ ERROR: 'location' field still present: {first_item['location']}")
|
||||
else:
|
||||
print(f" ✅ SUCCESS: 'location' field removed as requested")
|
||||
|
||||
|
||||
def test_unauthorized_access():
|
||||
"""Test that non-admin users cannot access the trigger endpoint."""
|
||||
print(f"\n🔒 Testing unauthorized access...")
|
||||
|
||||
trigger_url = f"{BASE_URL}/api/v1/trending/calculate/"
|
||||
|
||||
# Test without authentication
|
||||
print("Testing without authentication...")
|
||||
response = requests.post(trigger_url)
|
||||
print(f"No auth status: {response.status_code}")
|
||||
|
||||
# Test with invalid token
|
||||
print("Testing with invalid token...")
|
||||
headers = {"Authorization": "Bearer invalid_token_123"}
|
||||
response = requests.post(trigger_url, headers=headers)
|
||||
print(f"Invalid token status: {response.status_code}")
|
||||
|
||||
if response.status_code in [401, 403]:
|
||||
print("✅ Unauthorized access properly blocked")
|
||||
else:
|
||||
print(f"❌ Unauthorized access not properly blocked: {response.status_code}")
|
||||
|
||||
|
||||
def main():
|
||||
"""Main test function."""
|
||||
print("🧪 ThrillWiki Manual Trigger Endpoint Test")
|
||||
print("=" * 50)
|
||||
|
||||
# First test unauthorized access
|
||||
test_unauthorized_access()
|
||||
|
||||
# Try to login and get token
|
||||
token = login_and_get_token()
|
||||
|
||||
if not token:
|
||||
print("❌ Cannot proceed without authentication token")
|
||||
return
|
||||
|
||||
# Test the manual trigger endpoint
|
||||
trigger_result = test_trigger_endpoint(token)
|
||||
|
||||
if trigger_result:
|
||||
print(f"\n⏳ Waiting 10 seconds for tasks to process...")
|
||||
time.sleep(10)
|
||||
|
||||
# Test the trending endpoints to see results
|
||||
test_trending_endpoints()
|
||||
|
||||
print(f"\n🏁 Test completed at {datetime.now()}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user