48 lines
2.0 KiB
Python
48 lines
2.0 KiB
Python
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.catalog_audit import audit_catalog, audit_waterbody_catalog
|
|
from app.database import Base
|
|
from app.models import CatchReport, Fish, ModerationStatus, SourceType, Spot, Waterbody
|
|
|
|
|
|
def test_catalog_audit_checks_the_whole_catalog() -> None:
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as db:
|
|
fish = Fish(slug="pike", name_ru="Щука", trophy_weight_g=10_000)
|
|
water = Waterbody(slug="lake", name_ru="Озеро", unlock_level=1)
|
|
spot = Spot(waterbody=water, x=10, y=20)
|
|
db.add(CatchReport(fish=fish, waterbody=water, spot=spot, weight_g=1000, reported_at=datetime.now(timezone.utc), source_type=SourceType.user, source_confidence=60, moderation_status=ModerationStatus.approved))
|
|
db.commit()
|
|
assert audit_catalog(db)["failures"] == 0
|
|
spot.x = 10_001
|
|
db.commit()
|
|
result = audit_catalog(db)
|
|
assert result["reports"] == 1
|
|
assert result["invalid_coordinates"] == 1
|
|
assert result["failures"] == 1
|
|
|
|
|
|
def test_waterbody_catalog_audit_reports_snapshot_gaps_without_legacy_rows() -> None:
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as db:
|
|
now = datetime.now(timezone.utc)
|
|
db.add_all([
|
|
Waterbody(
|
|
slug="lake", name_ru="Озеро", source_system="rf4db",
|
|
source_external_id="level_001_lake", source_url="https://rf4db.com/ru/maps/level_001_lake",
|
|
source_checked_at=now,
|
|
),
|
|
Waterbody(slug="legacy", name_ru="Старое озеро"),
|
|
])
|
|
db.commit()
|
|
result = audit_waterbody_catalog(db, {"level_001_lake", "level_002_river"})
|
|
assert result["expected"] == 2
|
|
assert result["observed"] == 1
|
|
assert result["missing_source_external_ids"] == ["level_002_river"]
|
|
assert result["failures"] == 1
|