#!/usr/bin/env python3 """ Phase 4: Manual E2E Flow Testing for RateRight Tests all critical user flows and captures results """ import requests import json from datetime import datetime import sys import os # Test configuration BASE_URL = "http://127.0.0.1:5001" CONTRACTOR_CREDS = {"email": "contractor@test.com", "password": "test123"} WORKER_CREDS = {"email": "worker@test.com", "password": "test123"} class E2EFlowTester: def __init__(self): self.session = requests.Session() self.results = [] self.test_data = {} def log_result(self, area, scenario, status, notes=""): """Log test result""" result = { "timestamp": datetime.now().isoformat(), "area": area, "scenario": scenario, "status": status, "notes": notes } self.results.append(result) print(f"[{status:4}] {area}: {scenario} - {notes}") def test_endpoint(self, method, path, **kwargs): """Test an endpoint and return response""" url = f"{BASE_URL}{path}" try: response = self.session.request(method, url, **kwargs) return response except Exception as e: return None def test_auth_flow(self): """Test authentication flow""" print("\n=== TESTING AUTH FLOW ===") # Test registration page resp = self.test_endpoint("GET", "/auth/register") if resp and resp.status_code == 200: self.log_result("Auth", "Register page access", "PASS") else: self.log_result("Auth", "Register page access", "FAIL", f"Status: {resp.status_code if resp else 'No response'}") # Test login page resp = self.test_endpoint("GET", "/auth/login") if resp and resp.status_code == 200: self.log_result("Auth", "Login page access", "PASS") else: self.log_result("Auth", "Login page access", "FAIL", f"Status: {resp.status_code if resp else 'No response'}") # Test login with contractor credentials resp = self.test_endpoint("POST", "/auth/login", data=CONTRACTOR_CREDS, allow_redirects=False) if resp and resp.status_code in [302, 303]: self.log_result("Auth", "Contractor login", "PASS") self.test_data['contractor_logged_in'] = True else: self.log_result("Auth", "Contractor login", "FAIL", f"Status: {resp.status_code if resp else 'No response'}") # Test dashboard access resp = self.test_endpoint("GET", "/dashboard") if resp and resp.status_code == 200: self.log_result("Auth", "Dashboard access (logged in)", "PASS") else: self.log_result("Auth", "Dashboard access (logged in)", "FAIL", f"Status: {resp.status_code if resp else 'No response'}") # Test logout resp = self.test_endpoint("GET", "/auth/logout", allow_redirects=False) if resp and resp.status_code in [302, 303]: self.log_result("Auth", "Logout", "PASS") else: self.log_result("Auth", "Logout", "FAIL", f"Status: {resp.status_code if resp else 'No response'}") def test_jobs_flow(self): """Test jobs workflow""" print("\n=== TESTING JOBS FLOW ===") # Login as contractor first self.test_endpoint("POST", "/auth/login", data=CONTRACTOR_CREDS) # Test job listing resp = self.test_endpoint("GET", "/jobs") if resp and resp.status_code == 200: self.log_result("Jobs", "Job listing", "PASS") else: self.log_result("Jobs", "Job listing", "FAIL", f"Status: {resp.status_code if resp else 'No response'}") # Test job creation page resp = self.test_endpoint("GET", "/jobs/create") if resp and resp.status_code == 200: self.log_result("Jobs", "Job creation page", "PASS") else: self.log_result("Jobs", "Job creation page", "FAIL", f"Status: {resp.status_code if resp else 'No response'}") # Test job posting job_data = { "title": "Test Job", "description": "Test job description", "location": "Sydney, NSW", "budget": "1000", "duration": "1 week" } resp = self.test_endpoint("POST", "/jobs/create", data=job_data, allow_redirects=False) if resp and resp.status_code in [200, 302, 303]: self.log_result("Jobs", "Post new job", "PASS") # Extract job ID if redirected if 'Location' in resp.headers: location = resp.headers['Location'] if '/jobs/' in location: job_id = location.split('/jobs/')[-1] self.test_data['job_id'] = job_id else: self.log_result("Jobs", "Post new job", "FAIL", f"Status: {resp.status_code if resp else 'No response'}") def test_contracts_flow(self): """Test contracts workflow""" print("\n=== TESTING CONTRACTS FLOW ===") # Login as contractor self.test_endpoint("POST", "/auth/login", data=CONTRACTOR_CREDS) # Test contract listing resp = self.test_endpoint("GET", "/contracts") if resp and resp.status_code == 200: self.log_result("Contracts", "Contract listing", "PASS") else: self.log_result("Contracts", "Contract listing", "FAIL", f"Status: {resp.status_code if resp else 'No response'}") # Test contract creation contract_data = { "worker_id": "1", # Assuming worker user has ID 1 "job_id": self.test_data.get('job_id', '1'), "scope": "Test contract scope", "payment_amount": "1000", "start_date": "2025-01-20", "end_date": "2025-01-27" } resp = self.test_endpoint("POST", "/contracts/create", data=contract_data, allow_redirects=False) if resp and resp.status_code in [200, 302, 303]: self.log_result("Contracts", "Create contract", "PASS") else: self.log_result("Contracts", "Create contract", "FAIL", f"Status: {resp.status_code if resp else 'No response'}") def test_messaging_flow(self): """Test messaging workflow""" print("\n=== TESTING MESSAGING FLOW ===") # Login as worker self.test_endpoint("POST", "/auth/login", data=WORKER_CREDS) # Test messages listing resp = self.test_endpoint("GET", "/messages") if resp and resp.status_code == 200: self.log_result("Messaging", "Messages listing", "PASS") else: self.log_result("Messaging", "Messages listing", "FAIL", f"Status: {resp.status_code if resp else 'No response'}") # Test sending message message_data = { "recipient_id": "1", # Contractor ID "content": "Test message from worker" } resp = self.test_endpoint("POST", "/messages/send", json=message_data) if resp and resp.status_code in [200, 201]: self.log_result("Messaging", "Send message", "PASS") else: self.log_result("Messaging", "Send message", "FAIL", f"Status: {resp.status_code if resp else 'No response'}") def test_ratings_flow(self): """Test ratings workflow""" print("\n=== TESTING RATINGS FLOW ===") # Login as contractor self.test_endpoint("POST", "/auth/login", data=CONTRACTOR_CREDS) # Test rating submission rating_data = { "rated_user_id": "2", # Worker ID "contract_id": "1", "rating": "5", "review": "Excellent work!" } resp = self.test_endpoint("POST", "/ratings/submit", json=rating_data) if resp and resp.status_code in [200, 201]: self.log_result("Ratings", "Submit rating", "PASS") else: self.log_result("Ratings", "Submit rating", "FAIL", f"Status: {resp.status_code if resp else 'No response'}") # Test ratings display resp = self.test_endpoint("GET", "/users/2/ratings") if resp and resp.status_code == 200: self.log_result("Ratings", "View user ratings", "PASS") else: self.log_result("Ratings", "View user ratings", "FAIL", f"Status: {resp.status_code if resp else 'No response'}") def test_payments_flow(self): """Test payments workflow""" print("\n=== TESTING PAYMENTS FLOW ===") # Login as contractor self.test_endpoint("POST", "/auth/login", data=CONTRACTOR_CREDS) # Test payments page resp = self.test_endpoint("GET", "/payments") if resp and resp.status_code == 200: self.log_result("Payments", "Payments page", "PASS") else: self.log_result("Payments", "Payments page", "FAIL", f"Status: {resp.status_code if resp else 'No response'}") # Test payment intent creation payment_data = { "contract_id": "1", "amount": "1000" } resp = self.test_endpoint("POST", "/payments/create-intent", json=payment_data) if resp and resp.status_code in [200, 201]: self.log_result("Payments", "Create payment intent", "PASS") else: self.log_result("Payments", "Create payment intent", "FAIL", f"Status: {resp.status_code if resp else 'No response'}") def test_calendar_flow(self): """Test calendar sync workflow""" print("\n=== TESTING CALENDAR FLOW ===") # Login as worker self.test_endpoint("POST", "/auth/login", data=WORKER_CREDS) # Test calendar page resp = self.test_endpoint("GET", "/scheduling/calendar") if resp and resp.status_code == 200: self.log_result("Calendar", "Calendar page", "PASS") else: self.log_result("Calendar", "Calendar page", "FAIL", f"Status: {resp.status_code if resp else 'No response'}") # Test OAuth initiation (should redirect) resp = self.test_endpoint("GET", "/scheduling/google-auth", allow_redirects=False) if resp and resp.status_code in [302, 303]: self.log_result("Calendar", "OAuth initiation", "PASS") else: self.log_result("Calendar", "OAuth initiation", "FAIL", f"Status: {resp.status_code if resp else 'No response'}") def generate_report(self): """Generate test report""" print("\n" + "="*60) print("PHASE 4: MANUAL E2E FLOW TEST RESULTS") print("="*60) # Count results by area area_results = {} for result in self.results: area = result['area'] if area not in area_results: area_results[area] = {'PASS': 0, 'FAIL': 0} area_results[area][result['status']] += 1 # Print summary print("\n| Area | Scenarios | PASS | FAIL | Result |") print("|------------|-----------|------|------|--------|") overall_pass = True for area, counts in area_results.items(): total = counts['PASS'] + counts['FAIL'] result = "PASS" if counts['FAIL'] == 0 else "FAIL" if counts['FAIL'] > 0: overall_pass = False print(f"| {area:10} | {total:9} | {counts['PASS']:4} | {counts['FAIL']:4} | {result:6} |") print("\n" + "="*60) print(f"OVERALL RESULT: {'PASS' if overall_pass else 'FAIL'}") print("="*60) # Save detailed results with open("evidence/phase4/02_e2e_flow_results.json", "w") as f: json.dump(self.results, f, indent=2) return overall_pass def run_all_tests(self): """Run all E2E flow tests""" print("Starting Phase 4: Manual E2E Flow Testing") print("Testing against: " + BASE_URL) print("-" * 60) # Check if server is running try: resp = requests.get(BASE_URL, timeout=5) print(f"✓ Server is running (status: {resp.status_code})") except: print("✗ Server is not running! Please start the Flask server.") return False # Run all test flows self.test_auth_flow() self.test_jobs_flow() self.test_contracts_flow() self.test_messaging_flow() self.test_ratings_flow() self.test_payments_flow() self.test_calendar_flow() # Generate report return self.generate_report() if __name__ == "__main__": tester = E2EFlowTester() success = tester.run_all_tests() sys.exit(0 if success else 1)