472 lines
26 KiB
Python
472 lines
26 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from uuid import UUID
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine, delete, select
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app.database import Base, get_session
|
|
from app.community_importer import stage_observations
|
|
from app.importer import ImportAlreadyRunning
|
|
from app.main import app
|
|
from app.models import Bait, BaitKind, CatchReport, DataSource, ExternalEntityAlias, ExternalObservation, Fish, ImportStatus, ModerationEvent, ModerationStatus, OfficialRecordImport, SourceType, Spot, SubmissionAttempt, Waterbody
|
|
|
|
|
|
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
|
|
Base.metadata.create_all(engine)
|
|
|
|
|
|
def override_session():
|
|
with Session(engine) as session:
|
|
yield session
|
|
|
|
|
|
app.dependency_overrides[get_session] = override_session
|
|
client = TestClient(app)
|
|
|
|
|
|
def setup_module() -> None:
|
|
with Session(engine) as db:
|
|
waterbody = Waterbody(slug="test-lake", name_ru="Тестовое озеро", unlock_level=1)
|
|
fish = Fish(slug="pike", name_ru="Щука", trophy_weight_g=10_000)
|
|
bait = Bait(name="Тестовая приманка", normalized_name="тестовая приманка", kind=BaitKind.lure)
|
|
spot = Spot(waterbody=waterbody, x=10, y=20, description="Тестовая точка")
|
|
db.add_all([waterbody, fish, bait, spot])
|
|
now = datetime.now(timezone.utc)
|
|
for index in range(3):
|
|
db.add(CatchReport(fish=fish, spot=spot, waterbody=waterbody, bait=bait, weight_g=3000 + index * 1000, fishing_method="spinning", reported_at=now - timedelta(hours=index), caught_at=now - timedelta(hours=index), player_name=f"Player {index}", source_type=SourceType.manual_import, source_confidence=90, moderation_status=ModerationStatus.approved))
|
|
db.commit()
|
|
|
|
|
|
def test_activity_filters_and_explains_score() -> None:
|
|
response = client.get("/api/v1/activity?waterbody=test-lake&fish=pike&hours=24")
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert "items" in payload
|
|
assert payload["total"] == 1
|
|
assert payload["limit"] == 20
|
|
assert payload["offset"] == 0
|
|
assert len(payload["items"]) == 1
|
|
assert payload["items"][0]["catches"] == 3
|
|
assert payload["items"][0]["unique_players"] == 3
|
|
assert "3 свежих улова" in payload["items"][0]["explanation"]
|
|
assert payload["items"][0]["sources"] == ["manual-import"]
|
|
|
|
|
|
def test_invalid_period_is_rejected() -> None:
|
|
assert client.get("/api/v1/activity?hours=13").status_code == 422
|
|
assert client.get("/api/v1/activity?sort=unknown").status_code == 422
|
|
|
|
|
|
def test_review_queue_filters_before_pagination() -> None:
|
|
with Session(engine) as db:
|
|
stage_observations(db, [{
|
|
"source_system": "rf4map", "source_external_id": f"queue-{i}",
|
|
"source_url": f"https://rf4map.ru/points/queue-{i}",
|
|
"fish": "Queue fish", "waterbody": "Queue water",
|
|
} for i in range(3)])
|
|
rows = list(db.scalars(select(ExternalObservation).where(ExternalObservation.source_system == "rf4map")))
|
|
for i, row in enumerate(rows):
|
|
row.status = "published" if i == 0 else "ready"
|
|
row.last_seen_at = datetime.now(timezone.utc) - timedelta(minutes=i)
|
|
db.commit()
|
|
try:
|
|
headers = {"Authorization": "Bearer change-me-in-production"}
|
|
url = "/api/v1/admin/external-observations?status=review&source_system=rf4map&limit=1"
|
|
first = client.get(url, headers=headers).json()
|
|
second = client.get(url + "&offset=1", headers=headers).json()
|
|
assert len(first) == len(second) == 1
|
|
assert first[0]["status"] == second[0]["status"] == "ready"
|
|
assert first[0]["id"] != second[0]["id"]
|
|
finally:
|
|
for row in rows:
|
|
db.delete(row)
|
|
db.commit()
|
|
|
|
|
|
def test_timeline_includes_more_than_catch_page_and_sitemap_includes_old_spots() -> None:
|
|
with Session(engine) as db:
|
|
fish = db.scalar(select(Fish).where(Fish.slug == "pike"))
|
|
water = db.scalar(select(Waterbody).where(Waterbody.slug == "test-lake"))
|
|
spot = Spot(waterbody=water, x=901, y=902)
|
|
db.add(spot)
|
|
db.flush()
|
|
spot_id = spot.id
|
|
reports = [CatchReport(fish=fish, waterbody=water, spot=spot, weight_g=1000,
|
|
reported_at=datetime.now(timezone.utc) - timedelta(hours=1 if i < 60 else 100),
|
|
source_type=SourceType.manual_import, source_confidence=70,
|
|
moderation_status=ModerationStatus.approved) for i in range(61)]
|
|
db.add_all(reports)
|
|
db.commit()
|
|
try:
|
|
result = client.get(f"/api/v1/spots/{spot_id}/timeline")
|
|
assert result.status_code == 200
|
|
assert sum(row["count"] for row in result.json()) == 60
|
|
assert len(client.get(f"/api/v1/spots/{spot_id}/catches").json()) == 50
|
|
for report in reports:
|
|
report.reported_at = datetime.now(timezone.utc) - timedelta(days=10)
|
|
db.commit()
|
|
assert "/spots/test-lake-901x902" in client.get("/api/v1/public-spot-pages").json()
|
|
finally:
|
|
for report in reports:
|
|
db.delete(report)
|
|
db.flush()
|
|
db.delete(spot)
|
|
db.commit()
|
|
|
|
|
|
def test_list_pagination_and_filter_validation() -> None:
|
|
assert client.get("/api/v1/fishes?limit=0").status_code == 422
|
|
assert client.get("/api/v1/fishes?limit=1&offset=0").status_code == 200
|
|
headers = {"Authorization": "Bearer change-me-in-production"}
|
|
assert client.get("/api/v1/admin/external-observations?status=unknown", headers=headers).status_code == 422
|
|
assert client.get("/api/v1/admin/catch-reports?offset=-1", headers=headers).status_code == 422
|
|
assert client.get("/api/v1/admin/catch-reports?limit=101", headers=headers).status_code == 422
|
|
assert client.get("/api/v1/admin/external-observations/00000000-0000-0000-0000-000000000000/alias-suggestions").status_code == 401
|
|
|
|
|
|
def test_public_source_status_hides_internal_details() -> None:
|
|
with Session(engine) as db:
|
|
if db.get(DataSource, "rf4db") is None:
|
|
db.add(DataSource(key="rf4db", name="RF4DB", base_url="https://rf4db.com", default_confidence=70, enabled=True))
|
|
db.commit()
|
|
response = client.get("/api/v1/source-status")
|
|
assert response.status_code == 200
|
|
assert response.json()
|
|
assert all("error_summary" not in item and "source_url" not in item for item in response.json())
|
|
|
|
|
|
def test_liveness_does_not_probe_dependencies() -> None:
|
|
response = client.get("/health?token=must-not-be-logged")
|
|
assert response.json() == {"status": "ok"}
|
|
assert len(response.headers["X-Request-ID"]) == 32
|
|
assert response.headers["X-Frame-Options"] == "DENY"
|
|
assert response.headers["Cross-Origin-Opener-Policy"] == "same-origin"
|
|
|
|
|
|
def test_admin_diagnostics_exposes_build_identity_only_to_admin() -> None:
|
|
assert client.get("/api/v1/admin/diagnostics").status_code == 401
|
|
response = client.get("/api/v1/admin/diagnostics", headers={"Authorization": "Bearer change-me-in-production"})
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["build"] == {"version": "0.1.0", "revision": "dev", "environment": "development"}
|
|
assert set(payload) == {"generated_at", "build", "counts"}
|
|
assert response.headers["Content-Disposition"] == "attachment; filename=rf4spotter-diagnostics.json"
|
|
serialized = response.text.lower()
|
|
for forbidden in ("player_name", "source_url", "error_summary", "raw_payload", "admin_token", "s3_"):
|
|
assert forbidden not in serialized
|
|
assert client.get("/api/v1/admin/moderation-history").status_code == 401
|
|
history = client.get("/api/v1/admin/moderation-history", headers={"Authorization": "Bearer change-me-in-production"})
|
|
assert history.status_code == 200
|
|
for forbidden in ("player_name", "source_url", "raw_payload", "screenshot"):
|
|
assert forbidden not in history.text.lower()
|
|
export = client.get("/api/v1/admin/moderation-history-export", headers={"Authorization": "Bearer change-me-in-production"})
|
|
assert export.status_code == 200
|
|
assert export.headers["Content-Disposition"] == "attachment; filename=rf4spotter-moderation-history.json"
|
|
assert set(export.json()) == {"generated_at", "count", "events"}
|
|
for forbidden in ("entity_id", "moderator", "reason", "player_name", "source_url", "raw_payload"):
|
|
assert forbidden not in export.text.lower()
|
|
|
|
|
|
def test_spot_detail_and_catches() -> None:
|
|
spot_id = client.get("/api/v1/activity").json()["items"][0]["spot_id"]
|
|
detail = client.get(f"/api/v1/spots/{spot_id}")
|
|
catches = client.get(f"/api/v1/spots/{spot_id}/catches")
|
|
assert detail.status_code == 200
|
|
assert detail.json()["catches_24h"] == 3
|
|
assert catches.status_code == 200
|
|
assert len(catches.json()) == 3
|
|
assert catches.json()[0]["source_system"] == "manual-import"
|
|
resolved = client.get("/api/v1/spots/resolve?waterbody=test-lake&x=10&y=20")
|
|
assert resolved.status_code == 200
|
|
assert resolved.json()["id"] == spot_id
|
|
|
|
|
|
def test_records_list_is_empty_before_import() -> None:
|
|
response = client.get("/api/v1/records")
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert "items" in payload
|
|
assert payload["total"] == 0
|
|
assert payload["limit"] == 50
|
|
assert payload["offset"] == 0
|
|
assert payload["items"] == []
|
|
|
|
|
|
def test_record_category_filter_is_applied_before_pagination() -> None:
|
|
with Session(engine) as db:
|
|
fish = db.scalar(select(Fish).where(Fish.slug == "pike"))
|
|
waterbody = db.scalar(select(Waterbody).where(Waterbody.slug == "test-lake"))
|
|
now = datetime.now(timezone.utc) - timedelta(days=30)
|
|
db.add_all([
|
|
CatchReport(fish=fish, waterbody=waterbody, weight_g=9000, caught_at=now, reported_at=now, source_type=SourceType.official_record, source_confidence=100, moderation_status=ModerationStatus.approved, raw_payload={"category": "other"}),
|
|
CatchReport(fish=fish, waterbody=waterbody, weight_g=8000, caught_at=now - timedelta(days=1), reported_at=now, source_type=SourceType.official_record, source_confidence=100, moderation_status=ModerationStatus.approved, raw_payload={"category": "wanted"}),
|
|
])
|
|
db.commit()
|
|
try:
|
|
response = client.get("/api/v1/records?category=wanted&limit=1")
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["total"] == 1 # only "wanted" matches
|
|
assert payload["limit"] == 1
|
|
assert payload["offset"] == 0
|
|
assert len(payload["items"]) == 1
|
|
assert payload["items"][0]["category"] == "wanted"
|
|
finally:
|
|
# Cleanup added records
|
|
db.execute(delete(CatchReport).where(
|
|
CatchReport.source_type == SourceType.official_record,
|
|
CatchReport.raw_payload["category"].as_string().in_(["other", "wanted"]),
|
|
))
|
|
db.commit()
|
|
|
|
|
|
def test_records_pagination_returns_correct_total_and_offset() -> None:
|
|
with Session(engine) as db:
|
|
fish = db.scalar(select(Fish).where(Fish.slug == "pike"))
|
|
waterbody = db.scalar(select(Waterbody).where(Waterbody.slug == "test-lake"))
|
|
now = datetime.now(timezone.utc)
|
|
# Add exactly 5 official records with unique weights
|
|
for index in range(5):
|
|
db.add(CatchReport(fish=fish, waterbody=waterbody, weight_g=70000 + index * 100, caught_at=now - timedelta(days=index), reported_at=now, source_type=SourceType.official_record, source_confidence=100, moderation_status=ModerationStatus.approved))
|
|
db.commit()
|
|
try:
|
|
# Page 1: limit=2, offset=0
|
|
response1 = client.get("/api/v1/records?limit=2&offset=0")
|
|
assert response1.status_code == 200
|
|
p1 = response1.json()
|
|
assert p1["total"] >= 5
|
|
assert p1["limit"] == 2
|
|
assert p1["offset"] == 0
|
|
assert len(p1["items"]) == 2
|
|
# Verify first item has our newest caught_at (index=0, weight=70000)
|
|
assert p1["items"][0]["weight_g"] == 70000
|
|
# Page 2: limit=2, offset=2
|
|
response2 = client.get("/api/v1/records?limit=2&offset=2")
|
|
assert response2.status_code == 200
|
|
p2 = response2.json()
|
|
assert p2["total"] == p1["total"] # total must be consistent
|
|
assert p2["limit"] == 2
|
|
assert p2["offset"] == 2
|
|
assert len(p2["items"]) == 2
|
|
# Page 3: limit=2, offset=4
|
|
response3 = client.get("/api/v1/records?limit=2&offset=4")
|
|
assert response3.status_code == 200
|
|
p3 = response3.json()
|
|
assert p3["total"] == p1["total"]
|
|
assert p3["offset"] == 4
|
|
# Last page should have remaining items
|
|
assert len(p3["items"]) <= 2
|
|
# Page 4: offset=total — past total, empty
|
|
response4 = client.get(f"/api/v1/records?limit=2&offset={p1['total']}")
|
|
assert response4.status_code == 200
|
|
p4 = response4.json()
|
|
assert p4["total"] == p1["total"]
|
|
assert p4["items"] == []
|
|
finally:
|
|
# Cleanup added records
|
|
db.execute(delete(CatchReport).where(
|
|
CatchReport.source_type == SourceType.official_record,
|
|
CatchReport.weight_g >= 70000,
|
|
))
|
|
db.commit()
|
|
|
|
|
|
def test_user_report_requires_moderation_before_activity() -> None:
|
|
created = client.post("/api/v1/catch-reports", json={"fish_slug": "pike", "waterbody_slug": "test-lake", "x": 77, "y": 88, "weight_g": 5500, "bait_name": "Новая приманка", "player_name": "Reporter"})
|
|
assert created.status_code == 201
|
|
assert created.headers["Cache-Control"] == "no-store"
|
|
assert created.json()["moderation_status"] == "pending"
|
|
report_id = created.json()["id"]
|
|
headers = {"Authorization": "Bearer change-me-in-production"}
|
|
pending = client.get("/api/v1/admin/catch-reports", headers=headers)
|
|
assert pending.status_code == 200
|
|
assert pending.headers["Cache-Control"] == "no-store"
|
|
assert any(item["id"] == report_id for item in pending.json())
|
|
approved = client.patch(f"/api/v1/admin/catch-reports/{report_id}", headers=headers, json={"status": "approved", "reason": "fixture verified", "expected_version": 0})
|
|
assert approved.status_code == 200
|
|
stale = client.patch(f"/api/v1/admin/catch-reports/{report_id}", headers=headers, json={"status": "rejected", "reason": "stale tab", "expected_version": 0})
|
|
assert stale.status_code == 409
|
|
assert "reload" in stale.json()["detail"]
|
|
activity = client.get("/api/v1/activity?waterbody=test-lake&fish=pike&hours=24").json()
|
|
assert any(item["x"] == 77 and item["catches"] == 1 for item in activity["items"])
|
|
|
|
|
|
def test_admin_requires_token() -> None:
|
|
assert client.get("/api/v1/admin/catch-reports").status_code == 401
|
|
assert client.get("/api/v1/admin/imports").status_code == 401
|
|
assert client.post("/api/v1/admin/imports/official-records").status_code == 401
|
|
assert client.get("/api/v1/admin/external-observations").status_code == 401
|
|
|
|
|
|
def test_external_observation_requires_mapping_and_complete_data_before_publication() -> None:
|
|
with Session(engine) as db:
|
|
stage_observations(db, [{
|
|
"source_system": "rf4db", "source_external_id": "review-complete",
|
|
"source_url": "https://rf4db.com/catches/review-complete", "fish": "Pike external",
|
|
"fish_external_id": "fish-1", "waterbody": "Lake external",
|
|
"waterbody_external_id": "lake-1", "x": 31, "y": 41, "weight_g": 6200,
|
|
"bait": "Тестовая приманка", "player_name": "External Player",
|
|
}])
|
|
observation_id = db.scalar(select(ExternalObservation.id).where(
|
|
ExternalObservation.source_external_id == "review-complete"
|
|
))
|
|
headers = {"Authorization": "Bearer change-me-in-production"}
|
|
premature = client.post(f"/api/v1/admin/external-observations/{observation_id}/publish", headers=headers, json={"expected_version": 0})
|
|
assert premature.status_code == 409
|
|
mapped = client.patch(
|
|
f"/api/v1/admin/external-observations/{observation_id}/mapping", headers=headers,
|
|
json={"fish_slug": "pike", "waterbody_slug": "test-lake", "note": "verified fixture", "expected_version": 0},
|
|
)
|
|
assert mapped.status_code == 200
|
|
assert mapped.json()["status"] == "ready"
|
|
published = client.post(f"/api/v1/admin/external-observations/{observation_id}/publish", headers=headers, json={"expected_version": 1})
|
|
assert published.status_code == 200
|
|
assert published.json()["status"] == "published"
|
|
repeated = client.post(f"/api/v1/admin/external-observations/{observation_id}/publish", headers=headers, json={"expected_version": 1})
|
|
assert repeated.status_code == 409
|
|
with Session(engine) as db:
|
|
observation = db.get(ExternalObservation, observation_id)
|
|
report = db.get(CatchReport, observation.catch_report_id)
|
|
aliases = list(db.scalars(select(ExternalEntityAlias).where(ExternalEntityAlias.source_system == "rf4db")))
|
|
assert report.moderation_status == ModerationStatus.approved
|
|
assert report.raw_payload["provenance"]["source_external_id"] == "review-complete"
|
|
assert {alias.entity_type for alias in aliases} == {"fish", "waterbody"}
|
|
|
|
|
|
def test_incomplete_external_observation_is_publicly_labelled_but_not_counted() -> None:
|
|
with Session(engine) as db:
|
|
stage_observations(db, [{
|
|
"source_system": "rf4db", "source_external_id": "review-incomplete",
|
|
"source_url": "https://rf4db.com/catches/review-incomplete", "fish": "Pike external",
|
|
"waterbody": "Lake external", "x": 32, "y": 42,
|
|
}])
|
|
observation_id = db.scalar(select(ExternalObservation.id).where(
|
|
ExternalObservation.source_external_id == "review-incomplete"
|
|
))
|
|
public = client.get("/api/v1/community-observations")
|
|
assert public.status_code == 200
|
|
signal = next(item for item in public.json() if item["id"] == str(observation_id))
|
|
assert signal["source_system"] == "rf4db"
|
|
assert signal["quality"] == "incomplete"
|
|
assert signal["missing_fields"] == ["вес"]
|
|
headers = {"Authorization": "Bearer change-me-in-production"}
|
|
incomplete = client.get("/api/v1/admin/external-observations?status=review&completeness=incomplete&q=Pike", headers=headers)
|
|
assert any(item["id"] == str(observation_id) for item in incomplete.json())
|
|
provenance = next(item for item in incomplete.json() if item["id"] == str(observation_id))
|
|
assert provenance["missing_fields"] == ["weight_g"]
|
|
assert provenance["first_seen_at"] and provenance["last_seen_at"]
|
|
assert set(provenance["source_payload"]) <= {"bait", "fishing_method", "rig_type", "retrieve_method", "retrieve_speed", "player_name", "published_at", "region", "category"}
|
|
complete = client.get("/api/v1/admin/external-observations?status=review&completeness=complete&q=Pike", headers=headers)
|
|
assert all(item["id"] != str(observation_id) for item in complete.json())
|
|
prioritized = client.get("/api/v1/admin/external-observations?status=review&order=risk", headers=headers)
|
|
assert prioritized.status_code == 200
|
|
assert prioritized.json()[0]["weight_g"] is None
|
|
assert all(item["x"] != 32 or item["y"] != 42 for item in client.get("/api/v1/activity").json()["items"])
|
|
mapped = client.patch(
|
|
f"/api/v1/admin/external-observations/{observation_id}/mapping", headers=headers,
|
|
json={"fish_slug": "pike", "waterbody_slug": "test-lake", "expected_version": 0},
|
|
)
|
|
assert mapped.json()["status"] == "mapped"
|
|
assert client.post(f"/api/v1/admin/external-observations/{observation_id}/publish", headers=headers, json={"expected_version": 1}).status_code == 409
|
|
rejected = client.patch(
|
|
f"/api/v1/admin/external-observations/{observation_id}/reject", headers=headers,
|
|
json={"reason": "weight is absent", "expected_version": 1},
|
|
)
|
|
assert rejected.json()["status"] == "rejected"
|
|
assert all(item["id"] != str(observation_id) for item in client.get("/api/v1/community-observations").json())
|
|
|
|
|
|
def test_admin_can_start_and_list_official_import(monkeypatch) -> None:
|
|
def fake_import(db: Session, **_: str) -> OfficialRecordImport:
|
|
run = OfficialRecordImport(
|
|
started_at=datetime.now(timezone.utc),
|
|
finished_at=datetime.now(timezone.utc),
|
|
status=ImportStatus.success,
|
|
source_url="fixture://admin",
|
|
rows_seen=2,
|
|
rows_created=2,
|
|
rows_updated=0,
|
|
)
|
|
db.add(run)
|
|
db.commit()
|
|
db.refresh(run)
|
|
return run
|
|
|
|
monkeypatch.setattr("app.routers.admin.import_records", fake_import)
|
|
headers = {"Authorization": "Bearer change-me-in-production"}
|
|
started = client.post("/api/v1/admin/imports/official-records", headers=headers)
|
|
assert started.status_code == 201
|
|
assert started.json()["source_url"] == "fixture://admin"
|
|
listed = client.get("/api/v1/admin/imports?limit=1&offset=0", headers=headers)
|
|
assert listed.status_code == 200
|
|
assert listed.json()[0]["id"] == started.json()["id"]
|
|
def busy_import(*args, **kwargs):
|
|
raise ImportAlreadyRunning("official import is already running")
|
|
|
|
monkeypatch.setattr("app.routers.admin.import_records", busy_import)
|
|
conflict = client.post("/api/v1/admin/imports/official-records", headers=headers)
|
|
assert conflict.status_code == 409
|
|
|
|
|
|
def test_pending_report_accepts_one_validated_screenshot(monkeypatch) -> None:
|
|
created = client.post("/api/v1/catch-reports", json={"fish_slug": "pike", "waterbody_slug": "test-lake", "x": 91, "y": 92, "weight_g": 4200}).json()
|
|
monkeypatch.setattr("app.routers.submissions.upload_screenshot", lambda raw, **metadata: "reports/test.jpg" if raw == b"image-bytes" and metadata == {"filename": "catch.jpg", "content_type": "image/jpeg"} else "unexpected")
|
|
upload_url = f"/api/v1/catch-reports/{created['id']}/screenshot"
|
|
assert client.post(upload_url, files={"screenshot": ("catch.jpg", b"image-bytes", "image/jpeg")}).status_code == 401
|
|
assert client.post(upload_url, headers={"X-Upload-Token": "wrong"}, files={"screenshot": ("catch.jpg", b"image-bytes", "image/jpeg")}).status_code == 401
|
|
response = client.post(upload_url, headers={"X-Upload-Token": created["screenshot_upload_token"]}, files={"screenshot": ("catch.jpg", b"image-bytes", "image/jpeg")})
|
|
assert response.status_code == 204
|
|
reused = client.post(upload_url, headers={"X-Upload-Token": created["screenshot_upload_token"]}, files={"screenshot": ("catch.jpg", b"image-bytes", "image/jpeg")})
|
|
assert reused.status_code == 401
|
|
|
|
|
|
def test_admin_delete_anonymizes_report_removes_screenshot_and_keeps_audit(monkeypatch) -> None:
|
|
created = client.post("/api/v1/catch-reports", json={"fish_slug": "pike", "waterbody_slug": "test-lake", "x": 93, "y": 94, "weight_g": 4300, "player_name": "Private Player", "source_url": "https://example.test/private", "comment": "private comment"}).json()
|
|
with Session(engine) as db:
|
|
report = db.get(CatchReport, UUID(created["id"]))
|
|
report.screenshot_key = "reports/private.jpg"
|
|
db.commit()
|
|
deleted_keys: list[str] = []
|
|
monkeypatch.setattr("app.routers.admin.delete_screenshot", deleted_keys.append)
|
|
headers = {"Authorization": "Bearer change-me-in-production"}
|
|
response = client.delete(f"/api/v1/admin/catch-reports/{created['id']}?expected_version=0", headers=headers)
|
|
assert response.status_code == 204
|
|
assert deleted_keys == ["reports/private.jpg"]
|
|
with Session(engine) as db:
|
|
report = db.get(CatchReport, UUID(created["id"]))
|
|
assert report.deleted_at is not None
|
|
assert report.moderation_status == ModerationStatus.rejected
|
|
assert report.player_name is None and report.source_url is None
|
|
assert report.screenshot_key is None and report.raw_payload is None
|
|
event = db.query(ModerationEvent).filter_by(catch_report_id=report.id).order_by(ModerationEvent.created_at.desc()).first()
|
|
assert event is not None
|
|
assert event.reason == "user report deleted and anonymized"
|
|
assert client.delete(f"/api/v1/admin/catch-reports/{created['id']}?expected_version=0", headers=headers).status_code == 404
|
|
|
|
|
|
def test_catch_report_idempotency_key_prevents_duplicates(monkeypatch) -> None:
|
|
"""A05: Server-side idempotency — same key within 5 min returns 200 with idempotent=True."""
|
|
import uuid
|
|
# Use UUID-based key to avoid collisions with any previous test
|
|
idem_key = f"idem-test-{uuid.uuid4().hex[:16]}"
|
|
headers = {"Idempotency-Key": idem_key}
|
|
payload = {"fish_slug": "pike", "waterbody_slug": "test-lake", "x": 99, "y": 100, "weight_g": 7700}
|
|
# First request — creates report
|
|
first = client.post("/api/v1/catch-reports", json=payload, headers=headers)
|
|
assert first.status_code == 201
|
|
assert first.json()["idempotent"] is False
|
|
report_id = first.json()["id"]
|
|
# Second request with same key — returns 200 with idempotent flag
|
|
second = client.post("/api/v1/catch-reports", json=payload, headers=headers)
|
|
assert second.status_code == 200, f"Expected 200, got {second.status_code}. Response: {second.json()}"
|
|
assert second.json()["idempotent"] is True
|
|
assert second.json()["id"] == report_id
|
|
assert second.json()["screenshot_upload_token"] == first.json()["screenshot_upload_token"]
|
|
changed = dict(payload, weight_g=7800)
|
|
conflict = client.post("/api/v1/catch-reports", json=changed, headers=headers)
|
|
assert conflict.status_code == 409
|