60 lines
3.9 KiB
Python
60 lines
3.9 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import create_engine, func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database import Base
|
|
from app.models import CatchReport, DataSource, ExternalObservation, Fish, ModerationEvent, ModerationStatus, SourceType, SubmissionAttempt, Waterbody
|
|
from app.retention import apply_retention
|
|
|
|
|
|
NOW = datetime(2026, 9, 6, tzinfo=timezone.utc)
|
|
|
|
|
|
def test_retention_dry_run_then_apply() -> 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)
|
|
waterbody = Waterbody(slug="lake", name_ru="Озеро", unlock_level=1)
|
|
source = DataSource(key="rf4db", name="RF4DB", base_url="https://rf4db.com", default_confidence=60, enabled=False)
|
|
pending = CatchReport(
|
|
fish=fish, waterbody=waterbody, weight_g=1000, reported_at=NOW - timedelta(days=31),
|
|
player_name="Private", source_type=SourceType.user, source_confidence=60,
|
|
moderation_status=ModerationStatus.pending, raw_payload={"comment": "private"},
|
|
screenshot_key="reports/old.jpg", screenshot_upload_token_hash="a" * 64,
|
|
)
|
|
approved = CatchReport(
|
|
fish=fish, waterbody=waterbody, weight_g=2000, reported_at=NOW - timedelta(days=181),
|
|
player_name="Old winner", source_type=SourceType.user, source_confidence=60,
|
|
moderation_status=ModerationStatus.approved, raw_payload={"comment": "old"},
|
|
)
|
|
fresh = CatchReport(
|
|
fish=fish, waterbody=waterbody, weight_g=3000, reported_at=NOW - timedelta(days=10),
|
|
player_name="Fresh", source_type=SourceType.user, source_confidence=60,
|
|
moderation_status=ModerationStatus.pending, raw_payload={"comment": "fresh"},
|
|
)
|
|
db.add_all([source, pending, approved, fresh])
|
|
db.flush()
|
|
db.add(SubmissionAttempt(client_hash="x" * 64, created_at=NOW - timedelta(days=2)))
|
|
db.add(ModerationEvent(catch_report=approved, created_at=NOW - timedelta(days=366), previous_status=ModerationStatus.pending, new_status=ModerationStatus.approved, moderator="admin"))
|
|
db.add(ExternalObservation(source=source, source_external_id="stale", source_url="https://rf4db.com/1", fish_name="Щука", waterbody_name="Озеро", first_seen_at=NOW - timedelta(days=100), last_seen_at=NOW - timedelta(days=100), status="staged", payload={"raw": True}))
|
|
db.add(ExternalObservation(source=source, source_external_id="published", source_url="https://rf4db.com/2", fish_name="Щука", waterbody_name="Озеро", first_seen_at=NOW - timedelta(days=400), last_seen_at=NOW - timedelta(days=400), status="published", payload={"raw": True}, catch_report=approved))
|
|
db.commit()
|
|
|
|
preview = apply_retention(db, now=NOW)
|
|
assert preview == {"submission_attempts": 1, "user_reports_anonymized": 2, "pending_reports_expired": 1, "screenshots_deleted": 1, "staging_observations_deleted": 1, "published_payloads_cleared": 1, "moderation_events_deleted": 1}
|
|
assert db.get(CatchReport, pending.id).player_name == "Private"
|
|
|
|
deleted: list[str] = []
|
|
assert apply_retention(db, now=NOW, dry_run=False, delete_object=deleted.append) == preview
|
|
assert deleted == ["reports/old.jpg"]
|
|
assert db.get(CatchReport, pending.id).moderation_status == ModerationStatus.rejected
|
|
assert db.get(CatchReport, pending.id).player_name is None
|
|
assert db.get(CatchReport, approved.id).raw_payload is None
|
|
assert db.get(CatchReport, fresh.id).player_name == "Fresh"
|
|
assert db.scalar(select(func.count()).select_from(SubmissionAttempt)) == 0
|
|
assert db.scalar(select(func.count()).select_from(ExternalObservation)) == 1
|
|
assert db.scalar(select(ExternalObservation)).payload == {}
|
|
assert db.scalar(select(func.count()).select_from(ModerationEvent)) == 1
|