Add administrative official import API

2026-09-03 07:59:15 +07:00
parent 44c3e797aa
commit ddd8b84055
7 changed files with 104 additions and 13 deletions
@@ -2,12 +2,13 @@ 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 import_records, parse_html
-from app.models import CatchReport, OfficialRecordImport, SourceType
+from app.importer import ImportSourceError, import_records, parse_html
+from app.models import CatchReport, ImportStatus, OfficialRecordImport, SourceType


 FIXTURE = Path(__file__).parents[3] / "tests" / "fixtures" / "records_ru_sample.html"
@@ -28,3 +29,29 @@ def test_parser_and_import_are_idempotent() -> None:
        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_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="record columns changed"):
+        parse_html(html, region="RU", category="records")