epic-download/backend_test.py

#!/usr/bin/env python3
import requests
import sys
import json
from datetime import datetime

class EpicTravelAPITester:
    def __init__(self, base_url="https://world-explorer-139.preview.emergentagent.com"):
        self.base_url = base_url
        self.api_url = f"{base_url}/api"
        self.token = None
        self.tests_run = 0
        self.tests_passed = 0
        self.test_results = []

    def log_test(self, name, success, message="", response_data=None):
        """Log test results"""
        self.tests_run += 1
        if success:
            self.tests_passed += 1
            print(f"✅ {name}: PASSED - {message}")
        else:
            print(f"❌ {name}: FAILED - {message}")
        
        self.test_results.append({
            "test": name,
            "status": "PASSED" if success else "FAILED",
            "message": message,
            "response_data": response_data
        })
        return success

    def run_test(self, name, method, endpoint, expected_status, data=None, headers=None):
        """Run a single API test"""
        url = f"{self.api_url}/{endpoint}"
        test_headers = {'Content-Type': 'application/json'}
        
        if self.token:
            test_headers['Authorization'] = f'Bearer {self.token}'
        
        if headers:
            test_headers.update(headers)

        try:
            if method == 'GET':
                response = requests.get(url, headers=test_headers)
            elif method == 'POST':
                response = requests.post(url, json=data, headers=test_headers)
            elif method == 'PUT':
                response = requests.put(url, json=data, headers=test_headers)
            elif method == 'DELETE':
                response = requests.delete(url, headers=test_headers)

            success = response.status_code == expected_status
            
            if success:
                try:
                    response_data = response.json() if response.content else {}
                    return self.log_test(name, True, f"Status: {response.status_code}", response_data), response_data
                except:
                    return self.log_test(name, True, f"Status: {response.status_code}", {}), {}
            else:
                try:
                    error_data = response.json() if response.content else {}
                    return self.log_test(name, False, f"Expected {expected_status}, got {response.status_code} - {error_data}"), {}
                except:
                    return self.log_test(name, False, f"Expected {expected_status}, got {response.status_code}"), {}

        except Exception as e:
            return self.log_test(name, False, f"Error: {str(e)}"), {}

    def test_health_check(self):
        """Test API health check"""
        success, data = self.run_test("API Health Check", "GET", "", 200)
        return success

    def test_admin_login(self):
        """Test admin login"""
        login_data = {
            "email": "admin@epictravel.com",
            "password": "admin123"
        }
        success, response = self.run_test("Admin Login", "POST", "auth/login", 200, login_data)
        
        if success and 'access_token' in response:
            self.token = response['access_token']
            self.log_test("Token Extraction", True, "JWT token successfully extracted")
            return True
        else:
            self.log_test("Token Extraction", False, "Failed to extract JWT token")
            return False

    def test_get_destinations(self):
        """Test getting all destinations"""
        success, data = self.run_test("Get All Destinations", "GET", "destinations", 200)
        if success and isinstance(data, list) and len(data) > 0:
            self.log_test("Destinations Data Validation", True, f"Found {len(data)} destinations")
            return data
        else:
            self.log_test("Destinations Data Validation", False, "No destinations returned or invalid data")
            return []

    def test_get_specials(self):
        """Test getting all specials"""
        success, data = self.run_test("Get All Specials", "GET", "specials", 200)
        if success and isinstance(data, list):
            self.log_test("Specials Data Validation", True, f"Found {len(data)} specials")
            return data
        else:
            self.log_test("Specials Data Validation", False, "Failed to get specials or invalid data")
            return []

    def test_search_destinations(self):
        """Test destination search functionality"""
        # Test search by name
        success, data = self.run_test("Search Destinations by Name", "GET", "destinations?search=Paris", 200)
        if success:
            found_paris = any(dest.get('name', '').lower() == 'paris' for dest in data)
            if found_paris:
                self.log_test("Search Results Validation", True, "Paris found in search results")
            else:
                self.log_test("Search Results Validation", False, "Paris not found in search results")

        # Test filter by category
        success, data = self.run_test("Filter Destinations by Category", "GET", "destinations?category=City", 200)
        if success:
            all_city = all(dest.get('category') == 'City' for dest in data)
            if all_city:
                self.log_test("Filter Results Validation", True, "All returned destinations are City type")
            else:
                self.log_test("Filter Results Validation", False, "Filter not working correctly")

    def test_contact_form(self):
        """Test contact form submission"""
        contact_data = {
            "name": "Test User",
            "email": "test@example.com",
            "message": "This is a test contact message"
        }
        success, data = self.run_test("Contact Form Submission", "POST", "contact", 200, contact_data)
        return success

    def test_newsletter_subscription(self):
        """Test newsletter subscription"""
        newsletter_data = {
            "email": "newsletter@test.com"
        }
        success, data = self.run_test("Newsletter Subscription", "POST", "newsletter/subscribe", 200, newsletter_data)
        return success

    def test_admin_crud_operations(self):
        """Test full CRUD operations for destinations (requires authentication)"""
        if not self.token:
            self.log_test("CRUD Operations", False, "No authentication token available")
            return False

        # Test creating a new destination
        new_destination = {
            "name": "Test Destination",
            "location": "Test Country",
            "description": "A test destination for API testing",
            "image": "https://via.placeholder.com/800x600",
            "category": "City",
            "rating": 4.5,
            "price": 999,
            "currency": "USD"
        }
        
        success, created_dest = self.run_test("Create New Destination", "POST", "destinations", 200, new_destination)
        
        if not success:
            return False
            
        dest_id = created_dest.get('id')
        if not dest_id:
            self.log_test("Destination ID Extraction", False, "Failed to get destination ID from creation response")
            return False

        # Test updating the destination
        update_data = {
            "name": "Updated Test Destination",
            "price": 1299
        }
        success, updated_dest = self.run_test("Update Destination", "PUT", f"destinations/{dest_id}", 200, update_data)
        
        if success and updated_dest.get('name') == "Updated Test Destination":
            self.log_test("Update Validation", True, "Destination updated successfully")
        else:
            self.log_test("Update Validation", False, "Destination update failed or data not persisted")

        # Test getting single destination
        success, single_dest = self.run_test("Get Single Destination", "GET", f"destinations/{dest_id}", 200)
        
        # Test adding destination to specials
        special_data = {
            "destination_id": dest_id,
            "discount": 25,
            "end_date": "2025-12-31",
            "highlights": ["Test special", "Limited time", "API test"]
        }
        success, created_special = self.run_test("Add Destination to Specials", "POST", "specials", 200, special_data)
        
        special_id = created_special.get('id') if success else None
        
        if special_id:
            # Test updating special
            special_update = {
                "discount": 30,
                "end_date": "2025-11-30"
            }
            success, updated_special = self.run_test("Update Special", "PUT", f"specials/{special_id}", 200, special_update)
            
            # Test removing destination from specials
            success, _ = self.run_test("Remove from Specials", "DELETE", f"specials/destination/{dest_id}", 200)

        # Test deleting the destination (cleanup)
        success, _ = self.run_test("Delete Test Destination", "DELETE", f"destinations/{dest_id}", 200)
        
        return True

    def run_all_tests(self):
        """Run all tests in sequence"""
        print("🚀 Starting Epic Travel API Testing...")
        print("=" * 50)
        
        # Test 1: Health Check
        print("\n📋 Testing Basic API Access...")
        if not self.test_health_check():
            print("❌ Health check failed, stopping tests")
            return self.generate_summary()

        # Test 2: Authentication
        print("\n🔐 Testing Authentication...")
        if not self.test_admin_login():
            print("❌ Authentication failed, continuing with public endpoints only")

        # Test 3: Public Endpoints
        print("\n🌍 Testing Public Endpoints...")
        destinations = self.test_get_destinations()
        specials = self.test_get_specials()
        self.test_search_destinations()

        # Test 4: Forms
        print("\n📝 Testing Form Submissions...")
        self.test_contact_form()
        self.test_newsletter_subscription()

        # Test 5: Admin Operations
        if self.token:
            print("\n⚙️ Testing Admin CRUD Operations...")
            self.test_admin_crud_operations()
        else:
            print("\n⚠️ Skipping Admin CRUD tests - No authentication token")

        return self.generate_summary()

    def generate_summary(self):
        """Generate test summary"""
        print("\n" + "=" * 50)
        print("📊 TEST SUMMARY")
        print("=" * 50)
        print(f"Tests Run: {self.tests_run}")
        print(f"Tests Passed: {self.tests_passed}")
        print(f"Tests Failed: {self.tests_run - self.tests_passed}")
        print(f"Success Rate: {(self.tests_passed / self.tests_run * 100):.1f}%" if self.tests_run > 0 else "0%")
        
        failed_tests = [test for test in self.test_results if test['status'] == 'FAILED']
        if failed_tests:
            print("\n❌ FAILED TESTS:")
            for test in failed_tests:
                print(f"  - {test['test']}: {test['message']}")
        
        print("\n" + "=" * 50)
        
        return {
            "total_tests": self.tests_run,
            "passed_tests": self.tests_passed,
            "failed_tests": self.tests_run - self.tests_passed,
            "success_rate": (self.tests_passed / self.tests_run * 100) if self.tests_run > 0 else 0,
            "detailed_results": self.test_results
        }

def main():
    tester = EpicTravelAPITester()
    results = tester.run_all_tests()
    
    # Return appropriate exit code
    return 0 if results['failed_tests'] == 0 else 1

if __name__ == "__main__":
    sys.exit(main())