"""Test gamification system - points and achievements""" import pytest from datetime import datetime, timedelta from app import create_app from app.extensions import db from app.models import User, Contract, Job from app.models.gamification import Leaderboard, Achievement, PointActivity from app.utils.gamification import award_points, calculate_level, get_user_rank_in_league, check_achievement_triggers from app.services.leaderboard_service import LeaderboardService from app.config.achievements import check_user_achievements, get_user_achievement_progress @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', 'FEATURES': { 'gamification_leaderboards': True, 'achievement_system': True } }) with app.app_context(): db.create_all() yield app db.session.remove() db.drop_all() @pytest.fixture def sample_users(app): """Create sample users for testing""" users = [] for i in range(5): user = User( email=f'gamification_test_user_{i}_{datetime.utcnow().timestamp()}@test.com', first_name=f'User{i}', last_name='Test', role='worker', phone_number=f'0400000{i:03d}', location='Sydney, NSW', abn_number=f'{12345678900 + i}', # Ensure unique 11-digit ABN is_active=True, privacy_consent=True, terms_accepted=True, terms_accepted_date=datetime.utcnow(), total_points=0, current_level=1, seasonal_league='bronze' ) user.set_password('password123') users.append(user) db.session.add_all(users) db.session.commit() return users class TestGamificationSystem: """Test gamification functionality""" def test_award_points_basic(self, app, sample_users): """Test basic point awarding""" with app.app_context(): user = sample_users[0] initial_points = user.total_points # Award points success = award_points(user.id, 'contract_completed', points=100) assert success is True # Check points were recorded activity = PointActivity.query.filter_by(user_id=user.id).first() assert activity is not None assert activity.activity_type == 'contract_completed' assert activity.points_earned == 100 # Check user total updated by re-querying the user updated_user = User.query.get(user.id) assert updated_user.total_points == initial_points + 100 def test_award_points_default_amounts(self, app, sample_users): """Test point awarding with default amounts""" with app.app_context(): user = sample_users[0] # Award points without specifying amount (should use default) success = award_points(user.id, 'contract_completed') assert success is True activity = PointActivity.query.filter_by(user_id=user.id).first() assert activity.points_earned == 100 # Default for contract_completed def test_level_calculation(self): """Test level progression calculation""" # Test level thresholds assert calculate_level(0) == 1 assert calculate_level(50) == 1 assert calculate_level(100) == 2 assert calculate_level(150) == 3 assert calculate_level(300) == 4 assert calculate_level(500) == 5 assert calculate_level(700) == 6 assert calculate_level(1000) == 7 assert calculate_level(1500) == 8 assert calculate_level(2500) == 9 assert calculate_level(5000) == 10 assert calculate_level(10000) == 10 # Max level def test_leaderboard_weekly(self, app, sample_users): """Test weekly leaderboard generation""" with app.app_context(): service = LeaderboardService() # Award different points to users points = [500, 300, 700, 150, 450] for i, user in enumerate(sample_users): award_points(user.id, 'contract_completed', points=points[i]) # Update user level user.current_level = calculate_level(points[i]) user.total_points = points[i] db.session.commit() # Get weekly leaderboard leaderboard = service.get_weekly_leaderboard(limit=5) assert len(leaderboard['leaderboard']) == 5 # Should be ordered by points (descending) expected_order = [700, 500, 450, 300, 150] actual_points = [entry['points'] for entry in leaderboard['leaderboard']] assert actual_points == expected_order def test_leaderboard_monthly(self, app, sample_users): """Test monthly leaderboard generation""" with app.app_context(): service = LeaderboardService() # Award points in current month for i, user in enumerate(sample_users[:3]): points = (i + 1) * 100 award_points(user.id, 'job_completion', points=points) user.total_points = points db.session.commit() # Get monthly leaderboard leaderboard = service.get_monthly_leaderboard(limit=10) assert len(leaderboard['leaderboard']) >= 3 assert leaderboard['period']['type'] == 'monthly' def test_league_leaderboard(self, app, sample_users): """Test league-specific leaderboard""" with app.app_context(): service = LeaderboardService() # Set users to different leagues leagues = ['bronze', 'bronze', 'silver', 'silver', 'gold'] for i, user in enumerate(sample_users): user.seasonal_league = leagues[i] user.total_points = (i + 1) * 100 db.session.commit() # Get bronze league leaderboard leaderboard = service.get_league_leaderboard('bronze', limit=10) # Should only have bronze league players bronze_users = [entry for entry in leaderboard['leaderboard'] if entry['league'] == 'bronze'] assert len(bronze_users) == 2 def test_user_rank_in_league(self, app, sample_users): """Test getting user's rank within their league""" with app.app_context(): # Set up bronze league users with different points for i, user in enumerate(sample_users[:3]): user.seasonal_league = 'bronze' user.total_points = (3 - i) * 100 # 300, 200, 100 user.current_level = calculate_level(user.total_points) db.session.commit() # Check rank of middle user (should be rank 2) rank_info = get_user_rank_in_league(sample_users[1].id) assert rank_info['rank'] == 2 assert rank_info['league'] == 'bronze' assert rank_info['total_in_league'] == 3 def test_check_achievement_triggers(self, app, sample_users): """Test achievement trigger checking""" with app.app_context(): user = sample_users[0] # Simulate completing first contract user.jobs_completed = 1 db.session.commit() # Check achievements new_achievements = check_user_achievements(user.id) # Should unlock "First Job" achievement if it exists assert isinstance(new_achievements, list) def test_multiple_point_activities_same_user(self, app, sample_users): """Test multiple point activities for same user""" with app.app_context(): user = sample_users[0] # Award points multiple times activities = [ ('contract_completed', 100), ('job_posted', 50), ('profile_completed', 25), ('first_rating', 75) ] for activity_type, points in activities: award_points(user.id, activity_type, points=points) # Check total points by re-querying the user updated_user = User.query.get(user.id) expected_total = sum(points for _, points in activities) assert updated_user.total_points == expected_total # Check individual activities recorded user_activities = PointActivity.query.filter_by(user_id=user.id).all() assert len(user_activities) == len(activities) def test_level_updates_with_points(self, app, sample_users): """Test that user level updates as points increase""" with app.app_context(): user = sample_users[0] # Start at level 1 assert user.current_level == 1 # Award enough points for level 4 (300 points) success = award_points(user.id, 'contract_completed', points=300) assert success is True # Get updated user object from database updated_user = User.query.get(user.id) # The award_points function should have already updated the level assert updated_user.current_level == 4 # 300 points = level 4 assert updated_user.total_points == 300 def test_leaderboard_service_get_user_position(self, app, sample_users): """Test getting specific user's position in leaderboard""" with app.app_context(): service = LeaderboardService() # Give users different points points = [100, 200, 300, 400, 500] for i, user in enumerate(sample_users): award_points(user.id, 'contract_completed', points=points[i]) user.total_points = points[i] db.session.commit() # Check position of user with 300 points (should be 3rd) position = service.get_user_position(sample_users[2].id, 'all_time') assert position['position'] == 3 assert position['points'] == 300 def test_league_promotion_logic(self, app, sample_users): """Test league promotion based on points""" with app.app_context(): user = sample_users[0] # Start in bronze assert user.seasonal_league == 'bronze' # Award enough points for silver (1000+ points typically) award_points(user.id, 'contract_completed', points=1000) user.total_points = 1000 # Manually promote (in real system this would be automatic) if user.total_points >= 1000: user.seasonal_league = 'silver' db.session.commit() assert user.seasonal_league == 'silver' def test_achievement_progress_tracking(self, app, sample_users): """Test tracking progress towards achievements""" with app.app_context(): user = sample_users[0] # Get initial progress progress = get_user_achievement_progress(user.id) # Should return progress data structure assert isinstance(progress, list) def test_point_activity_timestamps(self, app, sample_users): """Test that point activities have correct timestamps""" with app.app_context(): user = sample_users[0] before_time = datetime.utcnow() award_points(user.id, 'contract_completed', points=100) after_time = datetime.utcnow() activity = PointActivity.query.filter_by(user_id=user.id).first() assert activity.date_earned is not None assert before_time <= activity.date_earned <= after_time def test_weekly_leaderboard_date_filtering(self, app, sample_users): """Test that weekly leaderboard only includes recent activities""" with app.app_context(): service = LeaderboardService() user = sample_users[0] # Create an old activity (more than 7 days ago) old_activity = PointActivity( user_id=user.id, activity_type='contract_completed', points_earned=100, date_earned=datetime.utcnow() - timedelta(days=10) ) db.session.add(old_activity) # Create a recent activity award_points(user.id, 'job_posted', points=50) db.session.commit() # Weekly leaderboard should only count recent points leaderboard = service.get_weekly_leaderboard() if leaderboard['leaderboard']: # Find user in leaderboard user_entry = next( (entry for entry in leaderboard['leaderboard'] if entry['user_id'] == user.id), None ) if user_entry: # Should only count recent 50 points, not old 100 points assert user_entry['points'] == 50 def test_invalid_user_id_award_points(self, app): """Test awarding points to non-existent user""" with app.app_context(): # Try to award points to non-existent user success = award_points(99999, 'contract_completed', points=100) assert success is False def test_zero_points_award(self, app, sample_users): """Test awarding zero points""" with app.app_context(): user = sample_users[0] initial_points = user.total_points # Award zero points success = award_points(user.id, 'contract_completed', points=0) # Should still record activity but not change total assert success is True activity = PointActivity.query.filter_by(user_id=user.id).first() assert activity.points_earned == 0 updated_user = User.query.get(user.id) assert updated_user.total_points == initial_points def test_leaderboard_empty_database(self, app): """Test leaderboard with no users""" with app.app_context(): service = LeaderboardService() leaderboard = service.get_weekly_leaderboard() assert leaderboard['leaderboard'] == [] assert leaderboard['period']['type'] == 'weekly' def test_negative_points_handling(self, app, sample_users): """Test handling of negative points (penalties allowed but logged)""" with app.app_context(): user = sample_users[0] initial_points = user.total_points # Award negative points (penalty) success = award_points(user.id, 'penalty', points=-50) # Should succeed as penalties are now allowed assert success is True # Check points were deducted updated_user = User.query.get(user.id) assert updated_user.total_points == initial_points - 50 # Check activity was recorded activity = PointActivity.query.filter_by(user_id=user.id).first() assert activity.points_earned == -50