"""Integration tests - test complete user workflows""" import pytest from datetime import datetime, date, timedelta from app import create_app from app.extensions import db from app.models import User, Job, Contract, Application from app.models.safety import Review from app.models.notification import Notification from app.models.message import Message from app.models.gamification import PointActivity from app.services.rating_service import RatingService from app.services.notification_service import NotificationService from app.services.message_service import MessageService from app.utils.gamification import award_points, calculate_level from app.utils.notification_helpers import NotificationTriggers @pytest.fixture def app(): """Create and configure test app""" app = create_app() app.config.update({ 'TESTING': True, 'SQLALCHEMY_DATABASE_URI': 'sqlite:///:memory:', 'SECRET_KEY': 'test-secret-key', 'MAIL_SUPPRESS_SEND': True, 'SMS_SUPPRESS_SEND': True, 'FEATURES': { 'gamification_leaderboards': True, 'achievement_system': True, 'notification_system': True, 'messaging_system': True } }) with app.app_context(): db.create_all() yield app db.session.remove() db.drop_all() @pytest.fixture def client(app): """Create test client""" return app.test_client() @pytest.fixture def complete_users(app): """Create complete user profiles for integration testing""" contractor = User( email='contractor@integration.com', first_name='John', last_name='Builder', role='contractor', phone_number='0400111111', location='Sydney, NSW', abn_number='11111111111', is_active=True, privacy_consent=True, terms_accepted=True, terms_accepted_date=datetime.utcnow(), total_points=0, current_level=1, seasonal_league='bronze', jobs_completed=0, average_rating=0.0 ) contractor.set_password('contractor123') worker = User( email='worker@integration.com', first_name='Sarah', last_name='Carpenter', role='worker', phone_number='0400222222', location='Melbourne, VIC', abn_number='22222222222', is_active=True, privacy_consent=True, terms_accepted=True, terms_accepted_date=datetime.utcnow(), total_points=0, current_level=1, seasonal_league='bronze', jobs_completed=0, average_rating=0.0 ) worker.set_password('worker123') db.session.add_all([contractor, worker]) db.session.commit() return { 'contractor': contractor, 'worker': worker } class TestIntegrationWorkflows: """Test complete integrated workflows""" def test_complete_job_workflow(self, app, complete_users): """Test complete job posting to completion workflow""" with app.app_context(): contractor = complete_users['contractor'] worker = complete_users['worker'] # 1. Create and post job job = Job( title='Kitchen Renovation Project', description='Complete kitchen renovation including plumbing and electrical work', contractor_id=contractor.id, location='Sydney, NSW', budget_min=5000.00, budget_max=8000.00, whs_requirements='White card required, public liability insurance', white_card_required=True, insurance_required=True, status='open' ) db.session.add(job) db.session.commit() # 2. Worker applies for job application = Application( job_id=job.id, worker_id=worker.id, proposed_rate=7500.00, cover_letter='I have 5 years experience in kitchen renovations...', abn_verified=True, insurance_verified=True, status='pending' ) db.session.add(application) db.session.commit() # 3. Contractor accepts application and creates contract application.status = 'accepted' job.status = 'assigned' contract = Contract( job_id=job.id, contractor_id=contractor.id, worker_id=worker.id, agreed_rate=7500.00, rate_type='total', start_date=date.today() + timedelta(days=7), end_date=date.today() + timedelta(days=37), scope_of_work=job.description, status='pending_agreement' ) db.session.add(contract) db.session.commit() # 4. Both parties sign contract contract.contractor_signed = True contract.contractor_signed_date = datetime.utcnow() contract.worker_signed = True contract.worker_signed_date = datetime.utcnow() contract.status = 'active' db.session.commit() # 5. Work is completed contract.status = 'completed' db.session.commit() # 6. Submit rating (triggers all systems) rating_service = RatingService() notification_service = NotificationService() message_service = MessageService() triggers = NotificationTriggers() # Contractor rates worker review = Review( reviewer_id=contractor.id, reviewee_id=worker.id, job_id=job.id, contract_id=contract.id, overall_rating=5, quality_rating=5, communication_rating=4, safety_rating=5, comment='Excellent work, highly recommended!', would_work_again=True, verified_completion=True ) db.session.add(review) db.session.commit() # 7. Update worker's average rating avg_rating = rating_service.calculate_average_rating(worker.id) worker.average_rating = avg_rating worker.jobs_completed += 1 db.session.commit() # 8. Award points for contract completion points_awarded = award_points(worker.id, 'contract_completed', points=100) # 9. Update user level based on points worker.current_level = calculate_level(worker.total_points) db.session.commit() # 10. Trigger rating notification triggers.rating_received( rated_user_id=worker.id, rating_user_id=contractor.id, rating_value=5, contract_id=contract.id ) # 11. Send completion message completion_message = message_service.send_message( sender_id=contractor.id, receiver_id=worker.id, content='Great job on the kitchen renovation! Looking forward to working with you again.', message_type='text' ) # Verify all systems updated correctly # Check rating system assert worker.average_rating == 4.75 # (5+5+4+5)/4 assert worker.jobs_completed == 1 # Check gamification system assert points_awarded is True assert worker.total_points > 0 point_activity = PointActivity.query.filter_by(user_id=worker.id).first() assert point_activity is not None assert point_activity.activity_type == 'contract_completed' # Check notification system notification = Notification.query.filter_by( user_id=worker.id, type='rating_received' ).first() assert notification is not None # Check messaging system assert completion_message is not None assert completion_message.content == 'Great job on the kitchen renovation! Looking forward to working with you again.' # Check contract workflow assert contract.status == 'completed' assert job.status == 'assigned' assert application.status == 'accepted' def test_rating_triggers_all_systems(self, app, complete_users): """Test that rating submission triggers all related systems""" with app.app_context(): contractor = complete_users['contractor'] worker = complete_users['worker'] # Create completed contract job = Job( title='Plumbing Repair', description='Fix kitchen sink leak', contractor_id=contractor.id, location='Sydney, NSW', budget_min=500.00, budget_max=800.00, status='assigned' ) db.session.add(job) db.session.flush() contract = Contract( job_id=job.id, contractor_id=contractor.id, worker_id=worker.id, agreed_rate=650.00, rate_type='total', start_date=date.today() - timedelta(days=7), end_date=date.today() - timedelta(days=1), scope_of_work='Plumbing repair work', status='completed' ) db.session.add(contract) db.session.commit() # Track initial states initial_points = worker.total_points initial_notifications = Notification.query.count() initial_messages = Message.query.count() # Submit rating - this should trigger cascading effects review = Review( reviewer_id=contractor.id, reviewee_id=worker.id, job_id=job.id, contract_id=contract.id, overall_rating=4, quality_rating=4, communication_rating=5, safety_rating=4, comment='Good work, completed on time', verified_completion=True ) db.session.add(review) db.session.commit() # Manually trigger all the systems (in real app this would be automatic) # 1. Update rating average rating_service = RatingService() avg_rating = rating_service.calculate_average_rating(worker.id) worker.average_rating = avg_rating worker.jobs_completed += 1 # 2. Award gamification points award_points(worker.id, 'rating_received', points=25) award_points(worker.id, 'job_completed', points=75) # 3. Send notification triggers = NotificationTriggers() triggers.rating_received( rated_user_id=worker.id, rating_user_id=contractor.id, rating_value=4, contract_id=contract.id ) # 4. Send follow-up message message_service = MessageService() message_service.send_message( sender_id=contractor.id, receiver_id=worker.id, content='Thanks for the good work! I left you a 4-star rating.' ) db.session.commit() # Verify all systems were triggered # Rating system db.session.refresh(worker) assert worker.average_rating == 4.25 # (4+4+5+4)/4 assert worker.jobs_completed == 1 # Gamification system assert worker.total_points > initial_points point_activities = PointActivity.query.filter_by(user_id=worker.id).all() assert len(point_activities) >= 2 # rating_received + job_completed # Notification system final_notifications = Notification.query.count() assert final_notifications > initial_notifications rating_notification = Notification.query.filter_by( user_id=worker.id, type='rating_received' ).first() assert rating_notification is not None # Messaging system final_messages = Message.query.count() assert final_messages > initial_messages follow_up_message = Message.query.filter_by( sender_id=contractor.id, receiver_id=worker.id ).first() assert follow_up_message is not None def test_user_journey_from_registration_to_completion(self, app): """Test complete user journey from registration to job completion""" with app.app_context(): # 1. User Registration new_contractor = User( email='newbie@contractor.com', first_name='New', last_name='Contractor', role='contractor', phone_number='0400999999', location='Brisbane, QLD', abn_number='99999999999', privacy_consent=True, terms_accepted=True, terms_accepted_date=datetime.utcnow() ) new_contractor.set_password('newbie123') new_worker = User( email='newbie@worker.com', first_name='New', last_name='Worker', role='worker', phone_number='0400888888', location='Brisbane, QLD', abn_number='88888888888', privacy_consent=True, terms_accepted=True, terms_accepted_date=datetime.utcnow() ) new_worker.set_password('newbie123') db.session.add_all([new_contractor, new_worker]) db.session.commit() # 2. Award initial points for profile completion award_points(new_contractor.id, 'profile_completed', points=50) award_points(new_worker.id, 'profile_completed', points=50) # 3. Contractor posts first job first_job = Job( title='Bathroom Renovation', description='Complete bathroom renovation project', contractor_id=new_contractor.id, location='Brisbane, QLD', budget_min=3000.00, budget_max=5000.00, status='open' ) db.session.add(first_job) db.session.commit() # Award points for posting first job award_points(new_contractor.id, 'first_job_posted', points=25) # 4. Worker applies application = Application( job_id=first_job.id, worker_id=new_worker.id, proposed_rate=4500.00, cover_letter='This is my first application on the platform', status='pending' ) db.session.add(application) db.session.commit() # 5. Send initial messages message_service = MessageService() intro_message = message_service.send_message( sender_id=new_worker.id, receiver_id=new_contractor.id, content='Hi! I applied for your bathroom renovation job. I have great references!' ) reply_message = message_service.send_message( sender_id=new_contractor.id, receiver_id=new_worker.id, content='Thanks for applying! Your profile looks good. I will review and get back to you.' ) # 6. Accept application and create contract application.status = 'accepted' first_job.status = 'assigned' contract = Contract( job_id=first_job.id, contractor_id=new_contractor.id, worker_id=new_worker.id, agreed_rate=4500.00, rate_type='total', start_date=date.today() + timedelta(days=3), end_date=date.today() + timedelta(days=20), scope_of_work=first_job.description, status='active' ) db.session.add(contract) db.session.commit() # 7. Complete work and submit ratings contract.status = 'completed' # Mutual ratings contractor_review = Review( reviewer_id=new_contractor.id, reviewee_id=new_worker.id, job_id=first_job.id, contract_id=contract.id, overall_rating=5, quality_rating=5, communication_rating=5, safety_rating=5, comment='Outstanding work! Exceeded expectations.', verified_completion=True ) worker_review = Review( reviewer_id=new_worker.id, reviewee_id=new_contractor.id, job_id=first_job.id, contract_id=contract.id, overall_rating=4, quality_rating=4, communication_rating=5, safety_rating=4, comment='Good client, clear instructions and prompt payment.', verified_completion=True ) db.session.add_all([contractor_review, worker_review]) db.session.commit() # 8. Update user stats and award completion points rating_service = RatingService() # Update worker stats worker_avg = rating_service.calculate_average_rating(new_worker.id) new_worker.average_rating = worker_avg new_worker.jobs_completed += 1 # Update contractor stats contractor_avg = rating_service.calculate_average_rating(new_contractor.id) new_contractor.average_rating = contractor_avg new_contractor.jobs_completed += 1 # Award completion points award_points(new_worker.id, 'first_job_completed', points=100) award_points(new_contractor.id, 'first_job_completed', points=100) # Update levels new_worker.current_level = calculate_level(new_worker.total_points) new_contractor.current_level = calculate_level(new_contractor.total_points) db.session.commit() # 9. Send completion notifications triggers = NotificationTriggers() triggers.rating_received( rated_user_id=new_worker.id, rating_user_id=new_contractor.id, rating_value=5, contract_id=contract.id ) triggers.rating_received( rated_user_id=new_contractor.id, rating_user_id=new_worker.id, rating_value=4, contract_id=contract.id ) # 10. Verify complete user journey # Check both users have progressed db.session.refresh(new_worker) db.session.refresh(new_contractor) # Worker progress assert new_worker.jobs_completed == 1 assert new_worker.average_rating == 5.0 assert new_worker.total_points >= 150 # profile + first_job_completed assert new_worker.current_level >= 2 # Contractor progress assert new_contractor.jobs_completed == 1 assert new_contractor.average_rating == 4.25 # (4+4+5+4)/4 assert new_contractor.total_points >= 175 # profile + first_job_posted + first_job_completed assert new_contractor.current_level >= 2 # Check messages exist messages = Message.query.filter( ((Message.sender_id == new_worker.id) & (Message.receiver_id == new_contractor.id)) | ((Message.sender_id == new_contractor.id) & (Message.receiver_id == new_worker.id)) ).all() assert len(messages) >= 2 # Check notifications were sent notifications = Notification.query.filter( (Notification.user_id == new_worker.id) | (Notification.user_id == new_contractor.id) ).all() assert len(notifications) >= 2 # Check point activities recorded worker_activities = PointActivity.query.filter_by(user_id=new_worker.id).all() contractor_activities = PointActivity.query.filter_by(user_id=new_contractor.id).all() assert len(worker_activities) >= 2 # profile + completion assert len(contractor_activities) >= 3 # profile + first_job_posted + completion def test_performance_under_concurrent_operations(self, app, complete_users): """Test system handles multiple simultaneous operations""" with app.app_context(): contractor = complete_users['contractor'] worker = complete_users['worker'] # Create multiple completed contracts contracts = [] for i in range(10): job = Job( title=f'Performance Test Job {i+1}', description=f'Performance testing job {i+1}', contractor_id=contractor.id, location='Sydney, NSW', budget_min=1000.00, budget_max=2000.00, status='assigned' ) db.session.add(job) db.session.flush() contract = Contract( job_id=job.id, contractor_id=contractor.id, worker_id=worker.id, agreed_rate=1500.00, rate_type='total', start_date=date.today() - timedelta(days=10), end_date=date.today() - timedelta(days=1), scope_of_work=f'Performance test work {i+1}', status='completed' ) db.session.add(contract) contracts.append(contract) db.session.commit() # Simulate concurrent operations start_time = datetime.utcnow() # Concurrent rating submissions for i, contract in enumerate(contracts): rating = 3 + (i % 3) # Ratings between 3-5 review = Review( reviewer_id=contractor.id, reviewee_id=worker.id, job_id=contract.job_id, contract_id=contract.id, overall_rating=rating, quality_rating=rating, communication_rating=rating, safety_rating=rating, comment=f'Performance test rating {i+1}', verified_completion=True ) db.session.add(review) # Award points concurrently award_points(worker.id, 'contract_completed', points=50) db.session.commit() end_time = datetime.utcnow() processing_time = (end_time - start_time).total_seconds() # Performance assertions assert processing_time < 5.0 # Should complete within 5 seconds # Verify data integrity reviews_count = Review.query.filter_by(reviewee_id=worker.id).count() assert reviews_count == 10 point_activities = PointActivity.query.filter_by(user_id=worker.id).count() assert point_activities == 10 # Calculate final rating average rating_service = RatingService() final_average = rating_service.calculate_average_rating(worker.id) expected_average = 4.0 # Should be average of 3,4,5,3,4,5,3,4,5,3 assert abs(final_average - expected_average) < 0.1 def test_error_handling_and_rollback(self, app, complete_users): """Test that errors in one system don't corrupt others""" with app.app_context(): contractor = complete_users['contractor'] worker = complete_users['worker'] # Create contract job = Job( title='Error Handling Test', description='Test error scenarios', contractor_id=contractor.id, location='Sydney, NSW', budget_min=1000.00, budget_max=2000.00, status='assigned' ) db.session.add(job) db.session.flush() contract = Contract( job_id=job.id, contractor_id=contractor.id, worker_id=worker.id, agreed_rate=1500.00, rate_type='total', start_date=date.today() - timedelta(days=7), end_date=date.today() - timedelta(days=1), scope_of_work='Error handling test work', status='completed' ) db.session.add(contract) db.session.commit() # Record initial state initial_points = worker.total_points initial_notifications = Notification.query.count() # Attempt operations that might fail try: # This should succeed review = Review( reviewer_id=contractor.id, reviewee_id=worker.id, job_id=job.id, contract_id=contract.id, overall_rating=5, quality_rating=5, communication_rating=5, safety_rating=5, comment='Error test rating', verified_completion=True ) db.session.add(review) # This should succeed award_points(worker.id, 'contract_completed', points=100) # Try to send notification to non-existent user (might fail gracefully) notification_service = NotificationService() try: notification_service.send_notification( user_id=99999, # Non-existent user type='rating_received', title='Test', message='This should fail gracefully' ) except Exception as e: # Notification failure shouldn't affect other systems pass db.session.commit() except Exception as e: db.session.rollback() # Even if something fails, we should be able to continue # Verify that successful operations completed despite failures final_reviews = Review.query.filter_by( reviewer_id=contractor.id, reviewee_id=worker.id ).count() assert final_reviews >= 1 # Rating should have succeeded # Points might have been awarded final_points = worker.total_points assert final_points >= initial_points