import json from dataclasses import asdict from datetime import datetime, timezone from pathlib import Path import pytest from sqlalchemy import create_engine, func, select from sqlalchemy.orm import Session from app.community_importer import CommunityImportError, stage_observations from app.database import Base from app.models import DataSource, ExternalObservation from rf4_research.community_sources import parse_rf4db_catches FIXTURE = Path(__file__).parents[3] / "tests" / "fixtures" / "rf4db_catches_sample.html" def record(source: str = "rf4db", external_id: str = "catch-1") -> dict[str, object]: return { "source_system": source, "source_external_id": external_id, "source_url": f"https://rf4db.com/ru/catches/{external_id}" if source == "rf4db" else f"https://rf4-stat.ru/fishing/{external_id}", "fish": "Щука", "fish_external_id": "pike", "waterbody": "Тестовое озеро", "waterbody_external_id": "test-lake", "x": 71, "y": 92, "weight_g": None, "published_at": None, "bait": "Приманка", } @pytest.fixture def db() -> Session: engine = create_engine("sqlite://") Base.metadata.create_all(engine) with Session(engine) as session: yield session engine.dispose() def test_staging_is_idempotent_and_preserves_first_seen(db: Session) -> None: first = datetime(2026, 9, 3, 10, tzinfo=timezone.utc) second = datetime(2026, 9, 3, 11, tzinfo=timezone.utc) assert stage_observations(db, [record()], fetched_at=first) == (1, 0) changed = record() | {"x": 73, "weight_g": 5_000} assert stage_observations(db, [changed], fetched_at=second) == (0, 1) item = db.scalar(select(ExternalObservation)) assert item is not None assert (item.x, item.weight_g, item.status) == (73, 5_000, "staged") assert item.first_seen_at.replace(tzinfo=timezone.utc) == first assert item.last_seen_at.replace(tzinfo=timezone.utc) == second assert db.scalar(select(func.count()).select_from(ExternalObservation)) == 1 source = db.get(DataSource, "rf4db") assert source is not None and source.enabled is False def test_external_ids_are_isolated_by_source(db: Session) -> None: created, updated = stage_observations(db, [record("rf4db"), record("rf4stat-fishing")]) assert (created, updated) == (2, 0) def test_parser_json_can_be_staged_without_losing_provenance(db: Session) -> None: parsed = parse_rf4db_catches(FIXTURE.read_text(encoding="utf-8")) payload = json.loads(json.dumps([asdict(item) for item in parsed], default=str)) assert stage_observations(db, payload) == (1, 0) item = db.scalar(select(ExternalObservation)) assert item is not None assert (item.source_system, item.fish_external_id, item.x, item.y) == ("rf4db", "pike", 71, 92) def test_invalid_source_rolls_back_caller_transaction(db: Session) -> None: with pytest.raises(CommunityImportError, match="unsupported source_system"): stage_observations(db, [record("unknown")]) db.rollback() assert db.scalar(select(func.count()).select_from(ExternalObservation)) == 0 @pytest.mark.parametrize("change", [ {"source_url": "https://attacker.example/catch-1"}, {"x": 10_001}, {"weight_g": 3_000_001}, ]) def test_invalid_provenance_and_ranges_are_rejected(db: Session, change: dict[str, object]) -> None: with pytest.raises(CommunityImportError): stage_observations(db, [record() | change]) db.rollback()