""" Comprehensive End-to-End Test Suite for RateRight Tests all features systematically and reports any issues found """ import os import sys import json import time import requests from datetime import datetime, timedelta from typing import Dict, List, Tuple, Optional # Add parent directory to path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) BASE_URL = "http://localhost:8080" TIMEOUT = 10 class RateRightE2ETester: def __init__(self): self.session = requests.Session() self.test_results = { "passed": [], "failed": [], "warnings": [], "fixes_applied": [] } self.test_data = {} def log_result(self, test_name: str, status: str, details: str = ""): """Log test results""" timestamp = datetime.now().strftime("%H:%M:%S") if status == "PASS": self.test_results["passed"].append(f"[{timestamp}] ✅ {test_name}") print(f"✅ {test_name}") elif status == "FAIL": self.test_results["failed"].append(f"[{timestamp}] ❌ {test_name}: {details}") print(f"❌ {test_name}: {details}") elif status == "WARNING": self.test_results["warnings"].append(f"[{timestamp}] ⚠️ {test_name}: {details}") print(f"⚠️ {test_name}: {details}") elif status == "FIXED": self.test_results["fixes_applied"].append(f"[{timestamp}] 🔧 {test_name}: {details}") print(f"🔧 Fixed: {test_name} - {details}") def test_server_health(self) -> bool: """Test if server is responding""" try: response = self.session.get(f"{BASE_URL}/", timeout=TIMEOUT) if response.status_code in [200, 302]: self.log_result("Server Health Check", "PASS") return True else: self.log_result("Server Health Check", "FAIL", f"Status: {response.status_code}") return False except Exception as e: self.log_result("Server Health Check", "FAIL", str(e)) return False def test_authentication(self) -> bool: """Test user registration and login""" print("\n=== Testing Authentication ===") # Test contractor registration contractor_data = { "first_name": "John", "last_name": "Contractor", "email": f"contractor_{int(time.time())}@test.com", "password": "Test123!@#", "confirm_password": "Test123!@#", "user_type": "contractor", "company_name": "Test Construction Co", "abn_number": "12345678901", "phone_number": "0412345678", "suburb": "Sydney", "state": "NSW" } try: # Register contractor response = self.session.post(f"{BASE_URL}/register", data=contractor_data, timeout=TIMEOUT) if response.status_code in [200, 302]: self.test_data["contractor_email"] = contractor_data["email"] self.test_data["contractor_password"] = contractor_data["password"] self.log_result("Contractor Registration", "PASS") else: self.log_result("Contractor Registration", "FAIL", f"Status: {response.status_code}") return False # Test contractor login login_data = { "email": contractor_data["email"], "password": contractor_data["password"] } response = self.session.post(f"{BASE_URL}/login", data=login_data, timeout=TIMEOUT) if response.status_code in [200, 302]: self.log_result("Contractor Login", "PASS") else: self.log_result("Contractor Login", "FAIL", f"Status: {response.status_code}") # Logout self.session.get(f"{BASE_URL}/logout") # Test worker registration worker_data = { "first_name": "Jane", "last_name": "Worker", "email": f"worker_{int(time.time())}@test.com", "password": "Test123!@#", "confirm_password": "Test123!@#", "user_type": "worker", "abn_number": "98765432109", "phone_number": "0498765432", "suburb": "Melbourne", "state": "VIC" } response = self.session.post(f"{BASE_URL}/register", data=worker_data, timeout=TIMEOUT) if response.status_code in [200, 302]: self.test_data["worker_email"] = worker_data["email"] self.test_data["worker_password"] = worker_data["password"] self.log_result("Worker Registration", "PASS") else: self.log_result("Worker Registration", "FAIL", f"Status: {response.status_code}") return False # Test worker login login_data = { "email": worker_data["email"], "password": worker_data["password"] } response = self.session.post(f"{BASE_URL}/login", data=login_data, timeout=TIMEOUT) if response.status_code in [200, 302]: self.log_result("Worker Login", "PASS") else: self.log_result("Worker Login", "FAIL", f"Status: {response.status_code}") # Logout self.session.get(f"{BASE_URL}/logout") return True except Exception as e: self.log_result("Authentication Tests", "FAIL", str(e)) return False def test_dashboards(self) -> bool: """Test dashboard access for both roles""" print("\n=== Testing Dashboards ===") try: # Test contractor dashboard login_data = { "email": self.test_data.get("contractor_email"), "password": self.test_data.get("contractor_password") } self.session.post(f"{BASE_URL}/login", data=login_data, timeout=TIMEOUT) response = self.session.get(f"{BASE_URL}/dashboard", timeout=TIMEOUT) if response.status_code == 200: if "contractor" in response.text.lower() or "post" in response.text.lower(): self.log_result("Contractor Dashboard Access", "PASS") else: self.log_result("Contractor Dashboard Access", "WARNING", "Dashboard loaded but content unclear") else: self.log_result("Contractor Dashboard Access", "FAIL", f"Status: {response.status_code}") self.session.get(f"{BASE_URL}/logout") # Test worker dashboard login_data = { "email": self.test_data.get("worker_email"), "password": self.test_data.get("worker_password") } self.session.post(f"{BASE_URL}/login", data=login_data, timeout=TIMEOUT) response = self.session.get(f"{BASE_URL}/dashboard", timeout=TIMEOUT) if response.status_code == 200: if "worker" in response.text.lower() or "jobs" in response.text.lower(): self.log_result("Worker Dashboard Access", "PASS") else: self.log_result("Worker Dashboard Access", "WARNING", "Dashboard loaded but content unclear") else: self.log_result("Worker Dashboard Access", "FAIL", f"Status: {response.status_code}") self.session.get(f"{BASE_URL}/logout") return True except Exception as e: self.log_result("Dashboard Tests", "FAIL", str(e)) return False def test_job_management(self) -> bool: """Test job posting and management""" print("\n=== Testing Job Management ===") try: # Login as contractor login_data = { "email": self.test_data.get("contractor_email"), "password": self.test_data.get("contractor_password") } self.session.post(f"{BASE_URL}/login", data=login_data, timeout=TIMEOUT) # Post a job job_data = { "title": f"Test Job {int(time.time())}", "description": "This is a test job for E2E testing", "category_id": 1, # Assuming category 1 exists "budget": 5000, "hourly_rate": 50, "location": "Sydney, NSW", "start_date": datetime.now().date().isoformat(), "end_date": (datetime.now() + timedelta(days=7)).date().isoformat(), "whs_required": True, "insurance_required": True } response = self.session.post(f"{BASE_URL}/jobs/create", data=job_data, timeout=TIMEOUT) if response.status_code in [200, 302]: self.log_result("Job Posting", "PASS") # Extract job ID from response or redirect if "location" in response.headers: job_id = response.headers["location"].split("/")[-1] self.test_data["job_id"] = job_id else: self.log_result("Job Posting", "FAIL", f"Status: {response.status_code}") return False # Test job listing response = self.session.get(f"{BASE_URL}/jobs", timeout=TIMEOUT) if response.status_code == 200: self.log_result("Job Listing", "PASS") else: self.log_result("Job Listing", "FAIL", f"Status: {response.status_code}") self.session.get(f"{BASE_URL}/logout") return True except Exception as e: self.log_result("Job Management Tests", "FAIL", str(e)) return False def test_applications(self) -> bool: """Test job application workflow""" print("\n=== Testing Applications ===") try: # Login as worker login_data = { "email": self.test_data.get("worker_email"), "password": self.test_data.get("worker_password") } self.session.post(f"{BASE_URL}/login", data=login_data, timeout=TIMEOUT) # Apply to job application_data = { "job_id": self.test_data.get("job_id", 1), "proposed_rate": 45, "cover_letter": "I am interested in this job and have relevant experience.", "availability": "immediate" } response = self.session.post(f"{BASE_URL}/jobs/{application_data['job_id']}/apply", data=application_data, timeout=TIMEOUT) if response.status_code in [200, 302]: self.log_result("Job Application Submission", "PASS") self.test_data["application_submitted"] = True else: self.log_result("Job Application Submission", "FAIL", f"Status: {response.status_code}") return False self.session.get(f"{BASE_URL}/logout") # Login as contractor to accept application login_data = { "email": self.test_data.get("contractor_email"), "password": self.test_data.get("contractor_password") } self.session.post(f"{BASE_URL}/login", data=login_data, timeout=TIMEOUT) # Accept application (assuming application ID is 1 for simplicity) response = self.session.post(f"{BASE_URL}/applications/1/accept", timeout=TIMEOUT) if response.status_code in [200, 302]: self.log_result("Application Acceptance", "PASS") self.test_data["contract_created"] = True else: self.log_result("Application Acceptance", "WARNING", f"Status: {response.status_code}") self.session.get(f"{BASE_URL}/logout") return True except Exception as e: self.log_result("Application Tests", "FAIL", str(e)) return False def test_contracts(self) -> bool: """Test contract workflow""" print("\n=== Testing Contracts ===") try: # Login as contractor login_data = { "email": self.test_data.get("contractor_email"), "password": self.test_data.get("contractor_password") } self.session.post(f"{BASE_URL}/login", data=login_data, timeout=TIMEOUT) # Check contracts response = self.session.get(f"{BASE_URL}/contracts", timeout=TIMEOUT) if response.status_code == 200: self.log_result("Contract Listing", "PASS") else: self.log_result("Contract Listing", "FAIL", f"Status: {response.status_code}") # Test contract signing (if contract exists) if self.test_data.get("contract_created"): response = self.session.post(f"{BASE_URL}/contracts/1/sign", timeout=TIMEOUT) if response.status_code in [200, 302]: self.log_result("Contract Signing (Contractor)", "PASS") else: self.log_result("Contract Signing (Contractor)", "WARNING", f"Status: {response.status_code}") self.session.get(f"{BASE_URL}/logout") # Login as worker and sign login_data = { "email": self.test_data.get("worker_email"), "password": self.test_data.get("worker_password") } self.session.post(f"{BASE_URL}/login", data=login_data, timeout=TIMEOUT) if self.test_data.get("contract_created"): response = self.session.post(f"{BASE_URL}/contracts/1/sign", timeout=TIMEOUT) if response.status_code in [200, 302]: self.log_result("Contract Signing (Worker)", "PASS") else: self.log_result("Contract Signing (Worker)", "WARNING", f"Status: {response.status_code}") self.session.get(f"{BASE_URL}/logout") return True except Exception as e: self.log_result("Contract Tests", "FAIL", str(e)) return False def test_messaging(self) -> bool: """Test messaging system""" print("\n=== Testing Messaging ===") try: # Login as contractor login_data = { "email": self.test_data.get("contractor_email"), "password": self.test_data.get("contractor_password") } self.session.post(f"{BASE_URL}/login", data=login_data, timeout=TIMEOUT) # Send message message_data = { "recipient_id": 2, # Assuming worker is user ID 2 "content": "Test message from contractor to worker" } response = self.session.post(f"{BASE_URL}/messages/send", data=message_data, timeout=TIMEOUT) if response.status_code in [200, 201]: self.log_result("Send Message", "PASS") else: self.log_result("Send Message", "WARNING", f"Status: {response.status_code}") # Check inbox response = self.session.get(f"{BASE_URL}/messages", timeout=TIMEOUT) if response.status_code == 200: self.log_result("Message Inbox Access", "PASS") else: self.log_result("Message Inbox Access", "FAIL", f"Status: {response.status_code}") self.session.get(f"{BASE_URL}/logout") return True except Exception as e: self.log_result("Messaging Tests", "FAIL", str(e)) return False def test_ratings(self) -> bool: """Test rating system""" print("\n=== Testing Ratings ===") try: # Login as contractor login_data = { "email": self.test_data.get("contractor_email"), "password": self.test_data.get("contractor_password") } self.session.post(f"{BASE_URL}/login", data=login_data, timeout=TIMEOUT) # Submit rating rating_data = { "contract_id": 1, "quality": 5, "communication": 4, "reliability": 5, "value": 4, "overall": 5, "comment": "Great work!", "would_work_again": True } response = self.session.post(f"{BASE_URL}/contracts/1/rate", data=rating_data, timeout=TIMEOUT) if response.status_code in [200, 201, 302]: self.log_result("Submit Rating", "PASS") else: self.log_result("Submit Rating", "WARNING", f"Status: {response.status_code}") self.session.get(f"{BASE_URL}/logout") return True except Exception as e: self.log_result("Rating Tests", "FAIL", str(e)) return False def test_payments(self) -> bool: """Test payment integration""" print("\n=== Testing Payments ===") try: # Check if Stripe is configured response = self.session.get(f"{BASE_URL}/payments/test", timeout=TIMEOUT) if response.status_code == 404: self.log_result("Payment Integration", "WARNING", "Payment routes not implemented yet") elif response.status_code == 200: self.log_result("Payment Integration", "PASS") else: self.log_result("Payment Integration", "FAIL", f"Status: {response.status_code}") return True except Exception as e: self.log_result("Payment Tests", "WARNING", str(e)) return True def test_calendar_scheduling(self) -> bool: """Test calendar and scheduling features""" print("\n=== Testing Calendar/Scheduling ===") try: # Login as worker login_data = { "email": self.test_data.get("worker_email"), "password": self.test_data.get("worker_password") } self.session.post(f"{BASE_URL}/login", data=login_data, timeout=TIMEOUT) # Test calendar view response = self.session.get(f"{BASE_URL}/calendar", timeout=TIMEOUT) if response.status_code == 200: self.log_result("Calendar View", "PASS") elif response.status_code == 404: self.log_result("Calendar View", "WARNING", "Calendar route not found") else: self.log_result("Calendar View", "FAIL", f"Status: {response.status_code}") # Test availability availability_data = { "day": "Monday", "start_time": "09:00", "end_time": "17:00" } response = self.session.post(f"{BASE_URL}/availability/set", data=availability_data, timeout=TIMEOUT) if response.status_code in [200, 201, 302]: self.log_result("Set Availability", "PASS") elif response.status_code == 404: self.log_result("Set Availability", "WARNING", "Availability route not found") else: self.log_result("Set Availability", "FAIL", f"Status: {response.status_code}") self.session.get(f"{BASE_URL}/logout") return True except Exception as e: self.log_result("Calendar/Scheduling Tests", "WARNING", str(e)) return True def test_gamification(self) -> bool: """Test gamification features""" print("\n=== Testing Gamification ===") try: # Login as worker login_data = { "email": self.test_data.get("worker_email"), "password": self.test_data.get("worker_password") } self.session.post(f"{BASE_URL}/login", data=login_data, timeout=TIMEOUT) # Check profile for points/level response = self.session.get(f"{BASE_URL}/profile", timeout=TIMEOUT) if response.status_code == 200: if "points" in response.text.lower() or "level" in response.text.lower(): self.log_result("Gamification Display", "PASS") else: self.log_result("Gamification Display", "WARNING", "Profile loaded but no gamification elements found") else: self.log_result("Gamification Display", "WARNING", f"Profile status: {response.status_code}") # Check leaderboard response = self.session.get(f"{BASE_URL}/leaderboard", timeout=TIMEOUT) if response.status_code == 200: self.log_result("Leaderboard Access", "PASS") elif response.status_code == 404: self.log_result("Leaderboard Access", "WARNING", "Leaderboard route not found") else: self.log_result("Leaderboard Access", "FAIL", f"Status: {response.status_code}") self.session.get(f"{BASE_URL}/logout") return True except Exception as e: self.log_result("Gamification Tests", "WARNING", str(e)) return True def test_notifications(self) -> bool: """Test notification system""" print("\n=== Testing Notifications ===") try: # Login as worker login_data = { "email": self.test_data.get("worker_email"), "password": self.test_data.get("worker_password") } self.session.post(f"{BASE_URL}/login", data=login_data, timeout=TIMEOUT) # Check notifications response = self.session.get(f"{BASE_URL}/notifications", timeout=TIMEOUT) if response.status_code == 200: self.log_result("Notifications Access", "PASS") elif response.status_code == 404: self.log_result("Notifications Access", "WARNING", "Notifications route not found") else: self.log_result("Notifications Access", "FAIL", f"Status: {response.status_code}") # Test notification preferences prefs_data = { "email_notifications": True, "sms_notifications": False, "push_notifications": True } response = self.session.post(f"{BASE_URL}/notifications/preferences", data=prefs_data, timeout=TIMEOUT) if response.status_code in [200, 302]: self.log_result("Notification Preferences", "PASS") elif response.status_code == 404: self.log_result("Notification Preferences", "WARNING", "Preferences route not found") else: self.log_result("Notification Preferences", "WARNING", f"Status: {response.status_code}") self.session.get(f"{BASE_URL}/logout") return True except Exception as e: self.log_result("Notification Tests", "WARNING", str(e)) return True def test_api_endpoints(self) -> bool: """Test API endpoints""" print("\n=== Testing API Endpoints ===") endpoints = [ ("/api/health", "GET", "Health Check"), ("/api/jobs", "GET", "Jobs API"), ("/api/workers", "GET", "Workers API"), ("/api/categories", "GET", "Categories API"), ("/api/stats", "GET", "Statistics API") ] for endpoint, method, name in endpoints: try: if method == "GET": response = self.session.get(f"{BASE_URL}{endpoint}", timeout=5) if response.status_code == 200: self.log_result(f"API: {name}", "PASS") elif response.status_code == 404: self.log_result(f"API: {name}", "WARNING", "Endpoint not found") else: self.log_result(f"API: {name}", "FAIL", f"Status: {response.status_code}") except Exception as e: self.log_result(f"API: {name}", "WARNING", "Timeout or error") return True def generate_report(self): """Generate comprehensive test report""" print("\n" + "="*60) print("COMPREHENSIVE E2E TEST REPORT") print("="*60) print(f"Test Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"Server: {BASE_URL}") print("\n") # Summary total_tests = len(self.test_results["passed"]) + len(self.test_results["failed"]) print(f"SUMMARY:") print(f" Total Tests: {total_tests}") print(f" ✅ Passed: {len(self.test_results['passed'])}") print(f" ❌ Failed: {len(self.test_results['failed'])}") print(f" ⚠️ Warnings: {len(self.test_results['warnings'])}") print(f" 🔧 Fixes Applied: {len(self.test_results['fixes_applied'])}") # Detailed results if self.test_results["passed"]: print("\n✅ PASSED TESTS:") for test in self.test_results["passed"]: print(f" {test}") if self.test_results["failed"]: print("\n❌ FAILED TESTS:") for test in self.test_results["failed"]: print(f" {test}") if self.test_results["warnings"]: print("\n⚠️ WARNINGS:") for warning in self.test_results["warnings"]: print(f" {warning}") if self.test_results["fixes_applied"]: print("\n🔧 FIXES APPLIED:") for fix in self.test_results["fixes_applied"]: print(f" {fix}") # Overall status print("\n" + "="*60) if len(self.test_results["failed"]) == 0: print("OVERALL STATUS: ✅ ALL CRITICAL TESTS PASSED") elif len(self.test_results["failed"]) < 3: print("OVERALL STATUS: ⚠️ MOSTLY PASSING WITH MINOR ISSUES") else: print("OVERALL STATUS: ❌ CRITICAL FAILURES DETECTED") # Save report report_data = { "timestamp": datetime.now().isoformat(), "server": BASE_URL, "results": self.test_results, "test_data": self.test_data } with open("E2E_TEST_RESULTS.json", "w") as f: json.dump(report_data, f, indent=2) print("\nDetailed report saved to: E2E_TEST_RESULTS.json") def run_all_tests(self): """Run all tests in sequence""" print("="*60) print("STARTING COMPREHENSIVE E2E TESTING") print("="*60) # Check server health first if not self.test_server_health(): print("❌ Server is not responding. Please ensure the server is running.") return # Run all test suites self.test_authentication() self.test_dashboards() self.test_job_management() self.test_applications() self.test_contracts() self.test_messaging() self.test_ratings() self.test_payments() self.test_calendar_scheduling() self.test_gamification() self.test_notifications() self.test_api_endpoints() # Generate final report self.generate_report() if __name__ == "__main__": tester = RateRightE2ETester() tester.run_all_tests()