from datetime import date, datetime, timedelta, timezone from fastapi.testclient import TestClient from sqlmodel import Session, select from app.db.session import engine, init_db from app.main import app from app.models import ChallengePlayer, CoinTransaction, GeoChallengeAttempt, GeoChallengeTarget from app.services.challenges import haversine_km, map_percent_to_lat_lon, score_geo_distance from tests.conftest import admin_headers def test_geo_distance_haversine(): distance = haversine_km(46.948, 7.447, 47.3769, 8.5417) assert 90 < distance < 100 def test_geo_scoring_city_gold_silver_neutral_miss(): assert score_geo_distance(8, 'city', 1)[1] == 'gold' assert score_geo_distance(20, 'city', 1)[1] == 'silver' assert score_geo_distance(40, 'city', 1)[1] == 'neutral' assert score_geo_distance(70, 'city', 1)[1] == 'far' assert score_geo_distance(120, 'city', 1)[1] == 'miss' def test_geo_scoring_lake_uses_wider_radius(): assert score_geo_distance(14, 'lake', 1)[1] == 'gold' assert score_geo_distance(30, 'lake', 1)[1] == 'silver' assert score_geo_distance(60, 'lake', 1)[1] == 'neutral' def test_geo_attempt_creates_points_and_returns_solution_after_submit(): init_db() with TestClient(app) as client: players = client.get('/api/challenges/players').json()['players'] player = players[0] today = client.get('/api/challenges/geo/today', params={'player_id': player['id']}).json() target = today['target'] assert 'latitude' not in target assert 'x_percent' not in target with Session(engine) as session: real = session.get(GeoChallengeTarget, target['id']) response = client.post('/api/challenges/geo/attempt', json={'player_id': player['id'], 'target_id': target['id'], 'selected_x_percent': real.x_percent, 'selected_y_percent': real.y_percent}) assert response.status_code == 200 payload = response.json() assert payload['target']['latitude'] == real.latitude assert payload['target']['x_percent'] == real.x_percent assert payload['selected']['x_percent'] == real.x_percent assert payload['raw_points'] >= 1 assert payload['credited_points'] >= 1 assert payload['points'] == payload['credited_points'] assert payload['daily_points_after'] >= payload['daily_points_before'] with Session(engine) as session: assert session.exec(select(GeoChallengeAttempt).where(GeoChallengeAttempt.player_id == player['id'])).first() is not None def test_hourly_attempt_limit_allows_ten_and_blocks_eleventh(): init_db() with TestClient(app) as client: player = client.get('/api/challenges/players').json()['players'][0] for _ in range(10): today = client.get('/api/challenges/geo/today', params={'player_id': player['id']}).json() target = today['target'] assert target is not None with Session(engine) as session: real = session.get(GeoChallengeTarget, target['id']) response = client.post('/api/challenges/geo/attempt', json={'player_id': player['id'], 'target_id': target['id'], 'selected_x_percent': real.x_percent, 'selected_y_percent': real.y_percent}) assert response.status_code == 200 blocked = client.get('/api/challenges/geo/today', params={'player_id': player['id']}).json() assert blocked['hourly_attempts_used'] == 10 assert blocked['hourly_attempts_remaining'] == 0 assert blocked['target'] is None with Session(engine) as session: answered = {attempt.target_id for attempt in session.exec(select(GeoChallengeAttempt).where(GeoChallengeAttempt.player_id == player['id'])).all()} extra = session.exec(select(GeoChallengeTarget).where(GeoChallengeTarget.id.not_in(answered))).first() limit = client.post('/api/challenges/geo/attempt', json={'player_id': player['id'], 'target_id': extra.id, 'selected_x_percent': extra.x_percent, 'selected_y_percent': extra.y_percent}) assert limit.status_code == 409 assert limit.json()['detail']['code'] == 'hourly_limit_reached' def test_hourly_limit_is_rolling_sixty_minutes(): init_db() now = datetime.now(timezone.utc) with Session(engine) as session: player = session.exec(select(ChallengePlayer).where(ChallengePlayer.name == 'Emilia')).one() target = session.exec(select(GeoChallengeTarget)).first() for index in range(10): lat, lon = map_percent_to_lat_lon(target.x_percent, target.y_percent) session.add(GeoChallengeAttempt(player_id=player.id, target_id=target.id + index, day=date.today(), selected_latitude=lat, selected_longitude=lon, selected_x_percent=target.x_percent, selected_y_percent=target.y_percent, distance_km=1, points=1, raw_points=1, credited_points=1, awarded_band='gold', created_at=now - timedelta(minutes=61))) session.commit() with TestClient(app) as client: player_payload = client.get('/api/challenges/players').json()['players'][0] today = client.get('/api/challenges/geo/today', params={'player_id': player_payload['id']}).json() assert today['hourly_attempts_used'] == 0 assert today['hourly_attempts_remaining'] == 10 def _add_attempts(session: Session, player: ChallengePlayer, count: int, points: float): target = session.exec(select(GeoChallengeTarget)).first() for index in range(count): lat, lon = map_percent_to_lat_lon(target.x_percent, target.y_percent) session.add(GeoChallengeAttempt(player_id=player.id, target_id=target.id, day=date.today(), selected_latitude=lat, selected_longitude=lon, selected_x_percent=target.x_percent, selected_y_percent=target.y_percent, distance_km=1, points=points, raw_points=points, credited_points=points, awarded_band='gold')) session.commit() def test_daily_points_cap_limits_credited_points_to_three(): init_db() with TestClient(app) as client: player = client.get('/api/challenges/players').json()['players'][0] credited = [] for _ in range(3): today = client.get('/api/challenges/geo/today', params={'player_id': player['id']}).json() target = today['target'] with Session(engine) as session: real = session.get(GeoChallengeTarget, target['id']) payload = client.post('/api/challenges/geo/attempt', json={'player_id': player['id'], 'target_id': target['id'], 'selected_x_percent': real.x_percent, 'selected_y_percent': real.y_percent}).json() credited.append(payload['credited_points']) today = client.get('/api/challenges/geo/today', params={'player_id': player['id']}).json() assert today['daily_points'] == 3.0 assert today['practice_only'] is True target = today['target'] with Session(engine) as session: real = session.get(GeoChallengeTarget, target['id']) practice = client.post('/api/challenges/geo/attempt', json={'player_id': player['id'], 'target_id': target['id'], 'selected_x_percent': real.x_percent, 'selected_y_percent': real.y_percent}).json() assert sum(credited) == 3.0 assert practice['raw_points'] >= 1 assert practice['credited_points'] == 0.0 assert practice['practice_only'] is True assert practice['daily_points_after'] == 3.0 def test_weekly_conversion_uses_credited_points_not_raw_points(): init_db() with Session(engine) as session: player = session.exec(select(ChallengePlayer).where(ChallengePlayer.name == 'Emilia')).one() target = session.exec(select(GeoChallengeTarget)).first() lat, lon = map_percent_to_lat_lon(target.x_percent, target.y_percent) session.add(GeoChallengeAttempt(player_id=player.id, target_id=target.id, day=date.today(), selected_latitude=lat, selected_longitude=lon, selected_x_percent=target.x_percent, selected_y_percent=target.y_percent, distance_km=1, points=1.0, raw_points=1.0, credited_points=0.0, awarded_band='gold', daily_cap_reached=True)) session.add(GeoChallengeAttempt(player_id=player.id, target_id=target.id, day=date.today(), selected_latitude=lat, selected_longitude=lon, selected_x_percent=target.x_percent, selected_y_percent=target.y_percent, distance_km=1, points=1.0, raw_points=1.0, credited_points=2.0, awarded_band='gold')) session.commit() with TestClient(app) as client: headers = admin_headers(client) response = client.post('/api/admin/challenges/weekly-convert', headers=headers) emilia = next(item for item in response.json()['conversions'] if item['player']['name'] == 'Emilia') assert emilia['challenge_points'] == 2.0 assert emilia['coins_awarded'] == 2 def test_weekly_conversion_caps_at_three_coins_and_is_idempotent(): init_db() with Session(engine) as session: player = session.exec(select(ChallengePlayer).where(ChallengePlayer.name == 'Emilia')).one() _add_attempts(session, player, 5, 1.0) with TestClient(app) as client: headers = admin_headers(client) first = client.post('/api/admin/challenges/weekly-convert', headers=headers) second = client.post('/api/admin/challenges/weekly-convert', headers=headers) emilia = next(item for item in first.json()['conversions'] if item['player']['name'] == 'Emilia') emilia2 = next(item for item in second.json()['conversions'] if item['player']['name'] == 'Emilia') assert emilia['coins_awarded'] == 3 assert emilia2['already_converted'] is True with Session(engine) as session: txs = session.exec(select(CoinTransaction).where(CoinTransaction.comment == 'Challenge-Wochenbonus Geografie')).all() assert len(txs) == 1 assert txs[0].amount == 3 def test_adults_do_not_receive_gassercoins(): init_db() with Session(engine) as session: player = session.exec(select(ChallengePlayer).where(ChallengePlayer.name == 'Marcel')).one() _add_attempts(session, player, 5, 1.0) with TestClient(app) as client: headers = admin_headers(client) response = client.post('/api/admin/challenges/weekly-convert', headers=headers) marcel = next(item for item in response.json()['conversions'] if item['player']['name'] == 'Marcel') assert marcel['coins_awarded'] == 0