gendesign/backend/tests/sql/test_2982_land_reservation_reparse.py

"""Починка разбора номера не удваивает land_reservation (#2982).

act_number входит в ключ `UNIQUE NULLS NOT DISTINCT (cad_num, act_number)`. На
проде 27 строк записаны с `act_number IS NULL`. Когда номер начинает извлекаться,
строка (участок, «1413») с (участок, NULL) не конфликтует — без очистки прогон
кладёт вторую строку рядом, и reservation_lookup показывает изъятие дважды.

Проверяется `_upsert_records` загрузчика целиком на временной таблице той же
формы: боевые данные не читаются и не меняются.
"""

from __future__ import annotations

import os

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

import pytest
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker

from app.workers.tasks.izyatie_ocr_ingest import _upsert_records


def _dsn() -> str:
    raw = os.environ.get("TEST_DATABASE_URL") or os.environ.get(
        "DATABASE_URL", "postgresql+psycopg://gendesign@localhost:15432/gendesign"
    )
    return (
        raw
        if raw.startswith("postgresql+")
        else raw.replace("postgresql://", "postgresql+psycopg://")
    )


def _db_reachable() -> tuple[bool, str]:
    try:
        eng = create_engine(_dsn(), connect_args={"connect_timeout": 3})
        with eng.connect() as c:
            c.execute(text("SELECT 1"))
        return True, ""
    except Exception as exc:
        return False, str(exc)


_DB_OK, _DB_ERR = _db_reachable()
pytestmark = pytest.mark.skipif(not _DB_OK, reason=f"Postgres недоступен: {_DB_ERR}")

_TABLE = """
CREATE TEMP TABLE land_reservation (
    id bigserial PRIMARY KEY,
    cad_num text NOT NULL,
    reservation_kind text NOT NULL,
    basis_act text NOT NULL,
    act_number text,
    act_date date,
    purpose text,
    doc_url text,
    source text NOT NULL DEFAULT 'page_pdf',
    is_active boolean NOT NULL DEFAULT true,
    raw_excerpt text,
    fetched_at timestamptz NOT NULL DEFAULT now(),
    CONSTRAINT uq_t UNIQUE NULLS NOT DISTINCT (cad_num, act_number)
) ON COMMIT DROP;
"""

КАД = "66:41:0303004:22"
ДОК = "https://xn--80acgfbsl1azdqr.xn--p1ai/file/89adb28a3677e7df933e2d9ce0f205c8"


def _row(cad: str = КАД, act: str | None = "1413", url: str = ДОК) -> dict:
    return {
        "cad_num": cad,
        "reservation_kind": "изъятие",
        "basis_act": "Сообщение о планируемом изъятии",
        "act_number": act,
        "act_date": "2022-05-27",
        "purpose": None,
        "doc_url": url,
        "source": "izyatie_ekb_ocr",
        "raw_excerpt": None,
    }


@pytest.fixture
def db():
    engine = create_engine(_dsn())
    session = sessionmaker(bind=engine)()
    try:
        session.execute(text(_TABLE))
        n = session.execute(text("SELECT count(*) FROM land_reservation")).scalar()
        assert n == 0, f"запрос попал НЕ во временную таблицу ({n} строк)"
        yield session
    finally:
        session.rollback()
        session.close()
        engine.dispose()


def _rows(db) -> list[tuple[str, str | None, str]]:
    return [
        (r[0], r[1], r[2])
        for r in db.execute(
            text("SELECT cad_num, act_number, doc_url FROM land_reservation ORDER BY 1, 2, 3")
        ).all()
    ]


def test_numbered_parse_replaces_numberless_row_of_same_document(db) -> None:
    """Головной: прод-состояние (номер NULL) + прогон с номером → одна строка с номером."""
    _upsert_records(db, [_row(act=None)])
    assert _rows(db) == [(КАД, None, ДОК)]

    assert _upsert_records(db, [_row(act="1413")]) == 1
    assert _rows(db) == [(КАД, "1413", ДОК)], "строка без номера осталась рядом с новой"

    _upsert_records(db, [_row(act="1413")])
    assert _rows(db) == [(КАД, "1413", ДОК)], "повторный прогон не идемпотентен"


def test_same_parcel_in_another_document_is_kept(db) -> None:
    """Контроль ширины: участок в ДРУГОМ документе — другое основание, его не трогаем."""
    другой = "https://xn--80acgfbsl1azdqr.xn--p1ai/file/9b9d9a998f578db56315bb816fc2ebf5"
    _upsert_records(db, [_row(act="259", url=другой)])
    _upsert_records(db, [_row(act=None)])
    _upsert_records(db, [_row(act="1413")])
    assert _rows(db) == [(КАД, "1413", ДОК), (КАД, "259", другой)]


def test_other_parcels_of_same_document_are_kept(db) -> None:
    """Контроль: чистка идёт по участку, а не по всему документу."""
    _upsert_records(db, [_row(cad="66:41:0303004:23", act=None)])
    _upsert_records(db, [_row(act="1413")])
    assert _rows(db) == [("66:41:0303004:22", "1413", ДОК), ("66:41:0303004:23", None, ДОК)]