118 lines
5.6 KiB
Python
118 lines
5.6 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database import Base
|
|
from app.models import CatchReport, CatchTackleComponent, Fish, ModerationStatus, SourceType, Spot, Waterbody
|
|
from app.routers.analytics import tackle_combinations
|
|
from app.schemas import TackleCombinationOut
|
|
|
|
|
|
def test_tackle_recommendation_requires_samples_and_independent_players() -> None:
|
|
now = datetime.now(timezone.utc)
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as db:
|
|
waterbody = Waterbody(slug="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)
|
|
db.add_all([waterbody, fish, spot])
|
|
db.flush()
|
|
for index, player in enumerate(("One", "One", "Two")):
|
|
report = CatchReport(
|
|
fish=fish, waterbody=waterbody, spot=spot, weight_g=1000,
|
|
caught_at=now - timedelta(hours=1), reported_at=now - timedelta(hours=1),
|
|
player_name=player, source_type=SourceType.user, source_confidence=80,
|
|
moderation_status=ModerationStatus.approved,
|
|
)
|
|
report.tackle_components.append(CatchTackleComponent(role="lure", position=0, raw_value="Spinner #1"))
|
|
db.add(report)
|
|
db.commit()
|
|
|
|
rows = tackle_combinations(db, waterbody="lake", fish="pike", method=None, hours=72, min_samples=3, min_players=2)
|
|
assert len(rows) == 1
|
|
assert (rows[0].status, rows[0].catches, rows[0].unique_players) == ("recommendation", 3, 2)
|
|
|
|
rows = tackle_combinations(db, waterbody="lake", fish="pike", method=None, hours=72, min_samples=3, min_players=3)
|
|
assert rows[0].status == "insufficient_data"
|
|
|
|
engine.dispose()
|
|
|
|
|
|
def test_tackle_analytics_handles_empty_and_multicomponent_observations() -> None:
|
|
now = datetime.now(timezone.utc)
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as db:
|
|
waterbody = Waterbody(slug="empty-check", name_ru="Проверка", unlock_level=1)
|
|
fish = Fish(slug="perch", name_ru="Окунь", trophy_weight_g=5_000)
|
|
spot = Spot(waterbody=waterbody, x=1, y=2)
|
|
report = CatchReport(
|
|
fish=fish, waterbody=waterbody, spot=spot, weight_g=500,
|
|
caught_at=now, reported_at=now, player_name="Player",
|
|
source_type=SourceType.user, source_confidence=80,
|
|
moderation_status=ModerationStatus.approved,
|
|
)
|
|
report.tackle_components.extend([
|
|
CatchTackleComponent(role="lure", position=0, raw_value="Spinner #1"),
|
|
CatchTackleComponent(role="rig", position=1, raw_value="Rig #1"),
|
|
CatchTackleComponent(role="lure", position=2, raw_value="Spinner #1"),
|
|
CatchTackleComponent(role="lure", position=3, raw_value=" "),
|
|
])
|
|
db.add(report)
|
|
db.commit()
|
|
|
|
rows = tackle_combinations(db, waterbody="empty-check", fish="perch", method=None, hours=72, min_samples=1, min_players=1)
|
|
assert {(row.role, row.value, row.catches) for row in rows} == {
|
|
("lure", "Spinner #1", 1), ("rig", "Rig #1", 1),
|
|
}
|
|
assert tackle_combinations(db, waterbody="missing", fish=None, method=None, hours=72) == []
|
|
|
|
engine.dispose()
|
|
|
|
|
|
def test_tackle_analytics_exposes_decay_and_prefers_fresher_equal_samples() -> None:
|
|
now = datetime.now(timezone.utc)
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as db:
|
|
waterbody = Waterbody(slug="decay-check", name_ru="Свежесть", unlock_level=1)
|
|
fish = Fish(slug="pike", name_ru="Щука", trophy_weight_g=10_000)
|
|
spot = Spot(waterbody=waterbody, x=3, y=4)
|
|
db.add_all([waterbody, fish, spot])
|
|
db.flush()
|
|
for value, age in (("Fresh spinner", 1), ("Old spinner", 48)):
|
|
for index, player in enumerate(("One", "One", "Two")):
|
|
report = CatchReport(
|
|
fish=fish, waterbody=waterbody, spot=spot, weight_g=1000,
|
|
caught_at=now - timedelta(hours=age), reported_at=now - timedelta(hours=age),
|
|
player_name=player, source_type=SourceType.user, source_confidence=80,
|
|
moderation_status=ModerationStatus.approved,
|
|
)
|
|
report.tackle_components.append(CatchTackleComponent(role="lure", position=index, raw_value=value))
|
|
db.add(report)
|
|
db.commit()
|
|
|
|
rows = tackle_combinations(db, waterbody="decay-check", fish="pike", method=None, hours=72, min_samples=3, min_players=2)
|
|
assert [row.value for row in rows] == ["Fresh spinner", "Old spinner"]
|
|
assert rows[0].freshness_score > rows[1].freshness_score
|
|
assert rows[0].freshness_score <= 100
|
|
|
|
engine.dispose()
|
|
|
|
|
|
def test_tackle_freshness_score_contract_rejects_out_of_range_values() -> None:
|
|
base = {
|
|
"role": "lure", "value": "Spinner", "catches": 3, "unique_players": 2,
|
|
"last_seen_at": datetime.now(timezone.utc), "status": "recommendation", "explanation": "ok",
|
|
}
|
|
assert TackleCombinationOut(**base, freshness_score=0).freshness_score == 0
|
|
assert TackleCombinationOut(**base, freshness_score=100).freshness_score == 100
|
|
for value in (-1, 101):
|
|
try:
|
|
TackleCombinationOut(**base, freshness_score=value)
|
|
except ValueError:
|
|
continue
|
|
raise AssertionError(f"freshness score {value} was accepted")
|