""" End-to-End Contract Workflow Test Tests the complete lifecycle: Registration → Job Posting → Application → Contract Creation → Signing → Completion → Closeout → Rating """ import pytest import time import requests from datetime import datetime, timezone, timedelta from flask import url_for from app import create_app from app.extensions import db from app.models.user import User from app.models.job import Job from app.models.contract import Contract from app.models.rating import Rating from app.models.category import Category from app.models.gamification import Achievement from app.models.enums import JobStatus, ContractStatus, ApplicationStatus from werkzeug.security import generate_password_hash import logging # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class TestE2EContractWorkflow: """Complete end-to-end workflow test""" @pytest.fixture(scope='class') def app(self): """Create application instance""" app = create_app() app.config['TESTING'] = True app.config['WTF_CSRF_ENABLED'] = False app.config['SERVER_NAME'] = 'localhost:5000' app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:' with app.app_context(): db.create_all() self._setup_categories() self._setup_badges() yield app db.session.remove() db.drop_all() @pytest.fixture(scope='class') def client(self, app): """Create test client""" return app.test_client() def _setup_categories(self): """Setup job categories""" categories = [ ('General Labor', 'General construction and labor work', 'low'), ('Renovation', 'Kitchen, bathroom, and home renovations', 'medium'), ('Electrical', 'Electrical installations and repairs', 'high'), ('Plumbing', 'Plumbing installations and repairs', 'high'), ('Carpentry', 'Carpentry and woodwork', 'medium') ] for name, desc, risk in categories: if not Category.query.filter_by(name=name).first(): category = Category( name=name, description=desc, whs_risk_level=risk, white_card_required=True, license_required=(risk == 'high') ) db.session.add(category) db.session.commit() logger.info(f"✓ Set up {len(categories)} categories") def _setup_badges(self): """Setup gamification badges - placeholder for now""" # The current Achievement model is different from what we expected # It's user-specific achievements, not badge templates logger.info("✓ Badge setup skipped (Achievement model is user-specific)") def test_complete_workflow(self, client, app): """Test the complete contract workflow""" workflow_report = [] issues_found = [] fixes_applied = [] with app.app_context(): try: # Step 1: Register Users logger.info("\n=== STEP 1: USER REGISTRATION ===") # Register contractor contractor_data = { 'first_name': 'Test', 'last_name': 'Contractor', 'email': 'contractor_test@example.com', 'password': 'TestPass123!', 'confirm_password': 'TestPass123!', 'abn_number': '12345678901', 'role': 'contractor', 'phone_number': '0412345678', 'location': 'Sydney' } response = client.post('/register', data=contractor_data, follow_redirects=True) if response.status_code == 200: workflow_report.append("✓ Contractor registered successfully") else: issues_found.append(f"Contractor registration failed: {response.status_code}") # Register worker worker_data = { 'first_name': 'Test', 'last_name': 'Worker', 'email': 'worker_test@example.com', 'password': 'TestPass123!', 'confirm_password': 'TestPass123!', 'abn_number': '98765432109', 'role': 'worker', 'phone_number': '0498765432', 'location': 'Sydney' } response = client.post('/register', data=worker_data, follow_redirects=True) if response.status_code == 200: workflow_report.append("✓ Worker registered successfully") else: issues_found.append(f"Worker registration failed: {response.status_code}") # Step 2: Login as Contractor and Post Job logger.info("\n=== STEP 2: JOB POSTING ===") response = client.post('/login', data={ 'email': 'contractor_test@example.com', 'password': 'TestPass123!' }, follow_redirects=True) if b'Dashboard' in response.data or b'dashboard' in response.data: workflow_report.append("✓ Contractor logged in successfully") else: issues_found.append("Contractor login failed") # Post a job job_data = { 'title': 'Test Kitchen Renovation', 'category_id': 2, # Renovation category 'location': 'Sydney, NSW', 'hourly_rate': 50.00, 'budget_min': 1000, 'budget_max': 5000, 'description': 'Complete kitchen renovation including cabinets and countertops', 'whs_requirements': 'White Card and Insurance required', 'white_card_required': 'on', 'insurance_required': 'on' } response = client.post('/jobs/post', data=job_data, follow_redirects=True) if response.status_code == 200: workflow_report.append("✓ Job posted successfully") job = Job.query.filter_by(title='Test Kitchen Renovation').first() if job: workflow_report.append(f" - Job ID: {job.id}") job_id = job.id else: issues_found.append("Job not found in database after posting") job_id = None else: issues_found.append(f"Job posting failed: {response.status_code}") job_id = None # Logout contractor client.get('/logout') # Step 3: Login as Worker and Apply logger.info("\n=== STEP 3: JOB APPLICATION ===") response = client.post('/login', data={ 'email': 'worker_test@example.com', 'password': 'TestPass123!' }, follow_redirects=True) if b'Dashboard' in response.data or b'dashboard' in response.data: workflow_report.append("✓ Worker logged in successfully") else: issues_found.append("Worker login failed") if job_id: # Apply for the job application_data = { 'cover_letter': 'I have 10 years experience in kitchen renovations', 'proposed_rate': 48.00 } response = client.post(f'/jobs/{job_id}/apply', data=application_data, follow_redirects=True) if response.status_code == 200: workflow_report.append("✓ Applied for job successfully") else: issues_found.append(f"Job application failed: {response.status_code}") # Logout worker client.get('/logout') # Step 4: Login as Contractor and Accept Application logger.info("\n=== STEP 4: APPLICATION ACCEPTANCE & CONTRACT CREATION ===") client.post('/login', data={ 'email': 'contractor_test@example.com', 'password': 'TestPass123!' }, follow_redirects=True) if job_id: # Get the application from app.models.job import Application application = Application.query.filter_by(job_id=job_id).first() if application: workflow_report.append(f" - Application ID: {application.id}") # Accept application response = client.post(f'/applications/{application.id}/accept', follow_redirects=True) if response.status_code == 200: workflow_report.append("✓ Application accepted") else: issues_found.append(f"Application acceptance failed: {response.status_code}") # Create contract contract_data = { 'start_date': datetime.now(timezone.utc).date().isoformat(), 'end_date': (datetime.now(timezone.utc) + timedelta(days=14)).date().isoformat(), 'hourly_rate': 48.00, 'estimated_hours': 40, 'terms': 'Standard renovation contract terms apply', 'safety_requirements': 'White Card and Insurance required' } response = client.post(f'/contracts/create/{application.id}', data=contract_data, follow_redirects=True) if response.status_code == 200: workflow_report.append("✓ Contract created successfully") contract = Contract.query.filter_by(job_id=job_id).first() if contract: workflow_report.append(f" - Contract ID: {contract.id}") contract_id = contract.id else: issues_found.append("Contract not found after creation") contract_id = None else: issues_found.append(f"Contract creation failed: {response.status_code}") contract_id = None else: issues_found.append("No application found for job") contract_id = None # Step 5: Sign Contract (Contractor) logger.info("\n=== STEP 5: CONTRACT SIGNING ===") if 'contract_id' in locals() and contract_id: response = client.post(f'/contracts/{contract_id}/sign', follow_redirects=True) if response.status_code == 200: workflow_report.append("✓ Contract signed by contractor") else: issues_found.append(f"Contractor signing failed: {response.status_code}") # Logout contractor client.get('/logout') # Login as worker and sign client.post('/login', data={ 'email': 'worker_test@example.com', 'password': 'TestPass123!' }, follow_redirects=True) if 'contract_id' in locals() and contract_id: response = client.post(f'/contracts/{contract_id}/sign', follow_redirects=True) if response.status_code == 200: workflow_report.append("✓ Contract signed by worker") # Check contract status contract = Contract.query.get(contract_id) if contract and contract.status == ContractStatus.ACTIVE: workflow_report.append("✓ Contract is now ACTIVE") else: issues_found.append(f"Contract status not active: {contract.status if contract else 'Not found'}") else: issues_found.append(f"Worker signing failed: {response.status_code}") # Step 6: Mark Contract as Complete logger.info("\n=== STEP 6: CONTRACT COMPLETION ===") # Logout worker and login as contractor client.get('/logout') client.post('/login', data={ 'email': 'contractor_test@example.com', 'password': 'TestPass123!' }, follow_redirects=True) if 'contract_id' in locals() and contract_id: response = client.post(f'/contracts/{contract_id}/complete', follow_redirects=True) if response.status_code == 200: workflow_report.append("✓ Contract marked as complete") contract = Contract.query.get(contract_id) if contract and contract.status == ContractStatus.COMPLETED: workflow_report.append("✓ Contract status is COMPLETED") else: issues_found.append(f"Contract not completed: {contract.status if contract else 'Not found'}") else: issues_found.append(f"Contract completion failed: {response.status_code}") # Step 7: Contract Closeout and Rating logger.info("\n=== STEP 7: CONTRACT CLOSEOUT & RATING ===") if 'contract_id' in locals() and contract_id: # Navigate to closeout page response = client.get(f'/contracts/{contract_id}/closeout') if response.status_code == 200: workflow_report.append("✓ Closeout page accessible") else: issues_found.append(f"Closeout page not accessible: {response.status_code}") # Submit contractor's rating of worker rating_data = { 'quality_rating': 5, 'timeliness_rating': 5, 'communication_rating': 4, 'professionalism_rating': 5, 'review_text': 'Excellent work on the kitchen renovation!' } response = client.post(f'/contracts/{contract_id}/rate', data=rating_data, follow_redirects=True) if response.status_code == 200: workflow_report.append("✓ Contractor rated worker successfully") else: issues_found.append(f"Contractor rating failed: {response.status_code}") # Logout contractor and login as worker client.get('/logout') client.post('/login', data={ 'email': 'worker_test@example.com', 'password': 'TestPass123!' }, follow_redirects=True) if 'contract_id' in locals() and contract_id: # Submit worker's rating of contractor rating_data = { 'payment_promptness': 5, 'scope_clarity': 4, 'communication_rating': 5, 'professionalism_rating': 5, 'review_text': 'Great client, clear requirements and prompt payment!' } response = client.post(f'/contracts/{contract_id}/rate', data=rating_data, follow_redirects=True) if response.status_code == 200: workflow_report.append("✓ Worker rated contractor successfully") else: issues_found.append(f"Worker rating failed: {response.status_code}") # Verify ratings in database ratings = Rating.query.filter_by(contract_id=contract_id).all() if len(ratings) == 2: workflow_report.append(f"✓ Both ratings found in database ({len(ratings)} ratings)") # Check overall scores for rating in ratings: if rating.reviewer_id: reviewer = User.query.get(rating.reviewer_id) if reviewer: workflow_report.append(f" - Rating by {reviewer.name}: {rating.overall_score}/5.0") else: issues_found.append(f"Expected 2 ratings, found {len(ratings)}") # Check if contract is now closed out contract = Contract.query.get(contract_id) if contract and contract.status == ContractStatus.CLOSED_OUT: workflow_report.append("✓ Contract successfully CLOSED OUT") else: issues_found.append(f"Contract not closed out: {contract.status if contract else 'Not found'}") # Step 8: Verify Gamification logger.info("\n=== STEP 8: GAMIFICATION VERIFICATION ===") contractor = User.query.filter_by(email='contractor_test@example.com').first() worker = User.query.filter_by(email='worker_test@example.com').first() if contractor: workflow_report.append(f"Contractor points: {contractor.total_points}") achievements = Achievement.query.filter_by(user_id=contractor.id).all() if achievements: workflow_report.append(f"Contractor achievements earned: {len(achievements)}") if worker: workflow_report.append(f"Worker points: {worker.total_points}") achievements = Achievement.query.filter_by(user_id=worker.id).all() if achievements: workflow_report.append(f"Worker achievements earned: {len(achievements)}") except Exception as e: issues_found.append(f"Unexpected error: {str(e)}") logger.error(f"Error during workflow: {e}", exc_info=True) # Generate Final Report logger.info("\n" + "="*60) logger.info("END-TO-END WORKFLOW TEST REPORT") logger.info("="*60) logger.info("\n✅ SUCCESSFUL STEPS:") for item in workflow_report: logger.info(f" {item}") if issues_found: logger.info("\n❌ ISSUES FOUND:") for issue in issues_found: logger.info(f" - {issue}") else: logger.info("\n🎉 ALL STEPS COMPLETED SUCCESSFULLY!") if fixes_applied: logger.info("\n🔧 FIXES APPLIED:") for fix in fixes_applied: logger.info(f" - {fix}") # Assert for test result assert len(issues_found) == 0, f"Workflow had {len(issues_found)} issues" assert len(workflow_report) >= 15, f"Expected at least 15 successful steps, got {len(workflow_report)}" if __name__ == "__main__": # Run the test directly import sys sys.exit(pytest.main([__file__, '-v', '-s']))