132 lines
4.3 KiB
Python
132 lines
4.3 KiB
Python
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) == []
|