100 lines
4.7 KiB
Python
100 lines
4.7 KiB
Python
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, ImportSourceError, 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_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)
|
|
assert (second.rows_created, second.rows_updated) == (0, 2)
|
|
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="<html></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
|