from __future__ import annotations from pathlib import Path import pytest from sqlalchemy import create_engine, func, select from sqlalchemy.orm import Session from app.database import Base from app.importer import FetchResult, ImportAlreadyRunning, ImportSourceError, _lock_key, _official_import_lock, import_records, parse_html from app.models import CatchReport, ImportStatus, OfficialRecordImport, SourceType FIXTURE = Path(__file__).parents[3] / "tests" / "fixtures" / "records_ru_sample.html" WEEKLY_FIXTURE = Path(__file__).parents[3] / "tests" / "fixtures" / "weekly_records_sample.html" def test_import_lock_is_stable_and_fails_closed_when_busy() -> None: class Connection: def scalar(self, statement, parameters): assert "pg_try_advisory_lock" in str(statement) assert parameters == {"key": _lock_key("https://example.test", "RU", "records")} return False def close(self): self.closed = True connection = Connection() bind = type("Bind", (), {"dialect": type("Dialect", (), {"name": "postgresql"})(), "connect": lambda self: connection})() session = type("Session", (), {"get_bind": lambda self: bind})() assert _lock_key("https://example.test", "ru", "records") == _lock_key("https://example.test", "RU", "records") with pytest.raises(ImportAlreadyRunning, match="already running"): with _official_import_lock(session, url="https://example.test", region="RU", category="records"): raise AssertionError("busy lock must not enter import") assert connection.closed is True def test_parser_and_import_are_idempotent() -> None: html = FIXTURE.read_text(encoding="utf-8") parsed = parse_html(html, region="RU", category="records") assert len(parsed) == 2 assert parsed[1].weight_g == 2_519_264 engine = create_engine("sqlite://") Base.metadata.create_all(engine) with Session(engine) as db: first = import_records(db, url="fixture://records", region="RU", category="records", html=html) second = import_records(db, url="fixture://records", region="RU", category="records", html=html) assert (first.rows_created, first.rows_updated) == (2, 0) # A12: Second import of identical data creates no events (no fields changed) assert (second.rows_created, second.rows_updated) == (0, 0) assert db.scalar(select(func.count()).select_from(CatchReport).where(CatchReport.source_type == SourceType.official_record)) == 2 assert db.scalar(select(func.count()).select_from(OfficialRecordImport)) == 2 def test_manual_weekly_category_import_uses_its_own_identity() -> None: html = WEEKLY_FIXTURE.read_text(encoding="utf-8") engine = create_engine("sqlite://") Base.metadata.create_all(engine) with Session(engine) as db: run = import_records( db, url="fixture://weekly-records", region="RU", category="weekly-ultralight", html=html, ) report = db.scalar(select(CatchReport).where(CatchReport.source_type == SourceType.official_record)) assert (run.rows_seen, run.rows_created, run.rows_updated) == (1, 1, 0) assert report.weight_g == 8_023 assert report.raw_payload["category"] == "weekly-ultralight" assert report.raw_payload["fish"] == "Wochenfisch" def test_failed_import_preserves_previous_records_and_is_logged() -> None: html = FIXTURE.read_text(encoding="utf-8") engine = create_engine("sqlite://") Base.metadata.create_all(engine) with Session(engine) as db: import_records(db, url="fixture://records", region="RU", category="records", html=html) before = db.scalar(select(func.count()).select_from(CatchReport)) with pytest.raises(ImportSourceError, match="records table not found"): import_records(db, url="fixture://broken", region="RU", category="records", html="") assert db.scalar(select(func.count()).select_from(CatchReport)) == before failed = db.scalar(select(OfficialRecordImport).where(OfficialRecordImport.status == ImportStatus.failed)) assert failed is not None assert failed.source_url == "fixture://broken" assert "records table not found" in (failed.error_summary or "") def test_import_rejects_changed_column_contract() -> None: html = FIXTURE.read_text(encoding="utf-8").replace( 'class="col data"', 'class="col changed"', 1 ) with pytest.raises(ImportSourceError, match="records columns changed"): parse_html(html, region="RU", category="records") def test_import_reuses_http_validators_and_handles_not_modified(monkeypatch) -> None: html = FIXTURE.read_text(encoding="utf-8") parsed = parse_html(html, region="RU", category="records") calls: list[tuple[str | None, str | None]] = [] def fake_fetch(url: str, *, region: str, category: str, etag: str | None, last_modified: str | None) -> FetchResult: calls.append((etag, last_modified)) if len(calls) == 1: return FetchResult(parsed, 200, '"fixture-v1"', "Wed, 02 Sep 2026 00:00:00 GMT", "text/html", len(html)) return FetchResult(None, 304, None, None, None, 0) monkeypatch.setattr("app.importer.fetch_records", fake_fetch) engine = create_engine("sqlite://") Base.metadata.create_all(engine) with Session(engine) as db: first = import_records(db, url="https://example.test/records", region="RU", category="records") second = import_records(db, url="https://example.test/records", region="RU", category="records") assert calls == [(None, None), ('"fixture-v1"', "Wed, 02 Sep 2026 00:00:00 GMT")] assert first.response_status == 200 assert second.response_status == 304 assert second.not_modified is True assert second.response_etag == '"fixture-v1"' assert second.rows_seen == 0