from __future__ import annotations import math from datetime import datetime, timedelta, timezone import pytest from sqlalchemy import create_engine from sqlalchemy.orm import Session from sqlalchemy.pool import StaticPool from app.activity import activity_rows from app.database import Base from app.models import CatchReport, Fish, ModerationStatus, SourceType, Spot, Waterbody NOW = datetime(2026, 9, 3, 6, 0, tzinfo=timezone.utc) @pytest.fixture def db() -> Session: engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) Base.metadata.create_all(engine) with Session(engine) as session: yield session engine.dispose() def add_report( db: Session, *, age_hours: float, player: str | None = "Player", confidence: int = 100, status: ModerationStatus = ModerationStatus.approved, deleted: bool = False, source: SourceType = SourceType.user, weight_g: int = 5_000, ) -> CatchReport: waterbody = db.query(Waterbody).first() fish = db.query(Fish).first() spot = db.query(Spot).first() if waterbody is None: waterbody = Waterbody(slug="test-lake", name_ru="Тестовое озеро", unlock_level=1) fish = Fish(slug="pike", name_ru="Щука", trophy_weight_g=10_000) spot = Spot(waterbody=waterbody, x=10, y=20, description=None) db.add_all([waterbody, fish, spot]) db.flush() report = CatchReport( fish=fish, spot=spot, waterbody=waterbody, bait=None, weight_g=weight_g, fishing_method="spinning", caught_at=NOW - timedelta(hours=age_hours), reported_at=NOW - timedelta(hours=age_hours), player_name=player, source_type=source, source_confidence=confidence, moderation_status=status, deleted_at=NOW if deleted else None, ) db.add(report) db.commit() return report def test_freshness_and_source_confidence_follow_documented_formula(db: Session) -> None: add_report(db, age_hours=18, confidence=80) row = activity_rows(db, hours=24, now=NOW)[0] weighted = math.exp(-1) * 0.8 expected_activity = round(55 * weighted / 12 + 25 / 6) expected_confidence = round(45 / 10 + 35 / 5 + 20 * 0.8) assert row.activity_score == expected_activity assert row.confidence_score == expected_confidence def test_repeated_reports_from_one_player_do_not_add_unique_player_weight(db: Session) -> None: add_report(db, age_hours=1, player=" Same Player ") add_report(db, age_hours=1, player="same player") add_report(db, age_hours=1, player=None) row = activity_rows(db, hours=6, now=NOW)[0] assert row.catches == 3 assert row.unique_players == 1 assert "3 свежих улова от 1 игрока" in row.explanation def test_explanation_describes_low_data_and_confidence(db: Session) -> None: add_report(db, age_hours=2, player="Player", confidence=50) row = activity_rows(db, hours=6, now=NOW)[0] assert row.explanation.startswith("Низкая активность: 1 свежий улов от 1 игрока.") assert "Уверенность низкая." in row.explanation assert "Данных мало: нужно хотя бы 3 наблюдения." in row.explanation def test_pending_rejected_and_deleted_reports_are_excluded(db: Session) -> None: add_report(db, age_hours=1) add_report(db, age_hours=1, status=ModerationStatus.pending) add_report(db, age_hours=1, status=ModerationStatus.rejected) add_report(db, age_hours=1, deleted=True) row = activity_rows(db, hours=6, now=NOW)[0] assert row.catches == 1 @pytest.mark.parametrize( ("hours", "expected_catches"), [(6, 1), (12, 2), (24, 3), (72, 4)], ) def test_supported_windows_are_deterministic(db: Session, hours: int, expected_catches: int) -> None: for age in (5, 11, 23, 71, 73): add_report(db, age_hours=age, player=f"Player {age}") row = activity_rows(db, hours=hours, now=NOW)[0] assert row.catches == expected_catches def test_reports_without_coordinates_do_not_create_activity_group(db: Session) -> None: report = add_report(db, age_hours=1, source=SourceType.official_record) report.spot = None db.commit() assert activity_rows(db, hours=72, now=NOW) == [] def test_confidence_capped_at_50_with_single_player() -> None: """D06: One player cannot artificially inflate confidence above 50%.""" engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) Base.metadata.create_all(engine) with Session(engine) as session: waterbody = Waterbody(slug="test-lake", name_ru="Тестовое озеро", unlock_level=1) fish = Fish(slug="pike", name_ru="Щука", trophy_weight_g=10_000) spot = Spot(waterbody=waterbody, x=10, y=20) session.add_all([waterbody, fish, spot]) session.flush() # 10 reports from 1 player, all max confidence for _ in range(10): report = CatchReport( fish=fish, spot=spot, waterbody=waterbody, bait=None, weight_g=5_000, fishing_method="spinning", caught_at=NOW - timedelta(hours=1), reported_at=NOW - timedelta(hours=1), player_name="Single Player", source_type=SourceType.user, source_confidence=100, moderation_status=ModerationStatus.approved, ) session.add(report) session.commit() row = activity_rows(session, hours=24, now=NOW)[0] assert row.unique_players == 1 assert row.confidence_score <= 50, f"Expected max 50 with 1 player, got {row.confidence_score}" def test_confidence_capped_at_65_with_two_players() -> None: """D06: Two players cannot get confidence above 65%.""" engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) Base.metadata.create_all(engine) with Session(engine) as session: waterbody = Waterbody(slug="test-lake", name_ru="Тестовое озеро", unlock_level=1) fish = Fish(slug="pike", name_ru="Щука", trophy_weight_g=10_000) spot = Spot(waterbody=waterbody, x=10, y=20) session.add_all([waterbody, fish, spot]) session.flush() # 10 reports from 2 players, all max confidence for i in range(10): report = CatchReport( fish=fish, spot=spot, waterbody=waterbody, bait=None, weight_g=5_000, fishing_method="spinning", caught_at=NOW - timedelta(hours=1), reported_at=NOW - timedelta(hours=1), player_name=f"Player {i % 2}", source_type=SourceType.user, source_confidence=100, moderation_status=ModerationStatus.approved, ) session.add(report) session.commit() row = activity_rows(session, hours=24, now=NOW)[0] assert row.unique_players == 2 assert row.confidence_score <= 65, f"Expected max 65 with 2 players, got {row.confidence_score}"