gendesign/backend/tests/sql/test_2986_permits_source_key.py

"""Бизнес-ключ gisogd_permits — key документа на портале, а не (группа, номер) (#2986).

`UNIQUE (doc_group, doc_num)` вводился, чтобы склеивать ОДИН документ, пришедший из
двух схем портала. Замер 20.08.2026 показал, что эта задача почти отсутствует, а
побочный эффект огромен — `docNum` у ГИСОГД не уникален: разрешение и изменения к
нему носят один номер.

    группа   документов   различных key   различных docNum   схлопывается
    DocRS         6098          6096            4305             1793
    DocRV         5419          5415            4969              450
    DocIZ          548           547             393              155 (+не грузилась)

    общих docNum между схемами (DocRS): 2      ← ради этого и вводился ключ
    общих key    между схемами (DocRS): 2      ← те же два, key их тоже склеит

Пример 66-06-06-2026: портал отдаёт два документа (key …719586 — разрешение,
key …752293 — изменения к нему), а на проде осталось только изменение. Так вытеснено
598 из 4320 строк РНС (13.8 %).

Тесты герметичные: временная копия таблицы, боевые данные не читаются и не меняются.
"""

from __future__ import annotations

import os

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

import re
from pathlib import Path

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

_MIGRATION = (
    Path(__file__).resolve().parents[3]
    / "data"
    / "sql"
    / "192_gisogd_permits_key_by_source_key.sql"
)


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()

# Прод-форма таблицы ДО миграции 191 (важные для ключа поля).
_TEMP_OLD = """
CREATE TEMP TABLE gisogd_permits (
    id            bigserial PRIMARY KEY,
    doc_group     text NOT NULL,
    doc_num       text NOT NULL,
    doc_name      text,
    date_reg      date,
    source_schema text NOT NULL,
    source_key    text NOT NULL,
    CONSTRAINT chk_gisogd_permits_doc_group CHECK (doc_group IN ('RS', 'RV')),
    CONSTRAINT gisogd_permits_doc_group_doc_num_key UNIQUE (doc_group, doc_num)
) ON COMMIT DROP
"""

# Два РАЗНЫХ документа с одним номером — дословно с портала.
_РАЗРЕШЕНИЕ = {
    "doc_group": "RS",
    "doc_num": "66-06-06-2026",
    "doc_name": "Разрешение на строительство № 66-06-06-2026 от 26.02.2026",
    "date_reg": "2026-02-26",
    "source_schema": "agate_sverdregion",
    "source_key": "1000130002719586",
}
_ИЗМЕНЕНИЕ = {
    "doc_group": "RS",
    "doc_num": "66-06-06-2026",
    "doc_name": "Изменения в разрешение на строительство №66-06-06-2026 от 26.02.2026",
    "date_reg": "2026-08-04",
    "source_schema": "agate_sverdregion",
    "source_key": "1000130002752293",
}

_INSERT_COLS = (
    "(doc_group, doc_num, doc_name, date_reg, source_schema, source_key)"
    " VALUES (:doc_group,:doc_num,:doc_name,CAST(:date_reg AS date),"
    ":source_schema,:source_key)"
)
# Старый UPSERT — цитата прежнего кода, нужна как контроль механизма.
_OLD_UPSERT = (
    f"INSERT INTO gisogd_permits {_INSERT_COLS}"
    " ON CONFLICT (doc_group, doc_num) DO UPDATE SET doc_name = EXCLUDED.doc_name,"
    " date_reg = EXCLUDED.date_reg, source_key = EXCLUDED.source_key"
    " WHERE EXCLUDED.date_reg > gisogd_permits.date_reg"
)
_NEW_UPSERT = (
    f"INSERT INTO gisogd_permits {_INSERT_COLS}"
    " ON CONFLICT (source_key) DO UPDATE SET doc_name = EXCLUDED.doc_name,"
    " date_reg = EXCLUDED.date_reg"
)


def _migration_chunks() -> list[str]:
    raw = _MIGRATION.read_text(encoding="utf-8")
    без_комментов = "\n".join(ln for ln in raw.splitlines() if not ln.lstrip().startswith("--"))
    return [
        ch.strip()
        for ch in re.split(r";\s*(?:\n|$)", без_комментов)
        if ch.strip() and ch.strip().upper() not in {"BEGIN", "COMMIT"}
    ]


def _session():
    return sessionmaker(bind=create_engine(_dsn()))()


# ── Гейт на лоадер: работает БЕЗ базы, двусторонний ───────────────────────────


def test_loader_knows_the_amendments_group() -> None:
    """DocIZ обязан быть в GROUP_CODE — иначе 548 документов не грузятся вовсе.

    На origin/main словарь содержит только DocRS/DocRV — конкретное неверное
    значение, а не отсутствующий символ.
    """
    from app.services.scrapers.gisogd66 import GROUP_CODE

    assert (
        GROUP_CODE.get("DocIZ") == "IZ"
    ), f"группа изменений не грузится; GROUP_CODE = {GROUP_CODE}"


def test_loader_upserts_by_source_key() -> None:
    """UPSERT обязан конфликтовать по source_key, а не по (doc_group, doc_num).

    На origin/main в тексте запроса стоит старый ключ — он и схлопывает
    разрешение с изменениями к нему.
    """
    import inspect

    from app.services.scrapers.gisogd66 import _upsert_permit

    src = inspect.getsource(_upsert_permit)
    assert "ON CONFLICT (source_key)" in src, "UPSERT конфликтует не по source_key"
    assert (
        "ON CONFLICT (doc_group, doc_num)" not in src
    ), "старый ключ всё ещё в запросе — разрешение и изменения схлопнутся"


# ── Механизм и миграция: нужен живой Postgres ────────────────────────────────

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


@_pg
def test_old_key_collapses_permit_and_its_amendment() -> None:
    """Демонстрация механизма: со СТАРЫМ ключом остаётся одна строка из двух.

    Не двусторонний тест, а показ причины: и старая схема, и старый UPSERT здесь
    процитированы явно, поэтому он одинаково зелёный на обеих сторонах. Нужен,
    чтобы утверждение «ключ схлопывает разные документы» было проверяемым, а не
    декларативным.
    """
    s = _session()
    try:
        s.execute(text(_TEMP_OLD))
        s.execute(text(_OLD_UPSERT), _РАЗРЕШЕНИЕ)
        s.execute(text(_OLD_UPSERT), _ИЗМЕНЕНИЕ)
        rows = s.execute(text("SELECT doc_name, source_key FROM gisogd_permits")).all()
        assert len(rows) == 1, f"ожидали схлопывание, получили {len(rows)} строк"
        assert (
            "Изменения" in rows[0][0]
        ), f"вытеснено не то: осталось {rows[0][0]!r} — на проде остаётся именно изменение"
    finally:
        s.rollback()
        s.close()


@_pg
def test_after_migration_both_documents_survive() -> None:
    """Головной: после миграции разрешение и изменения к нему живут раздельно."""
    s = _session()
    try:
        s.execute(text(_TEMP_OLD))
        for chunk in _migration_chunks():
            s.execute(text(chunk))
        s.execute(text(_NEW_UPSERT), _РАЗРЕШЕНИЕ)
        s.execute(text(_NEW_UPSERT), _ИЗМЕНЕНИЕ)
        rows = s.execute(text("SELECT doc_name FROM gisogd_permits ORDER BY source_key")).all()
        assert len(rows) == 2, f"документы всё ещё схлопываются: {rows}"
        имена = " | ".join(r[0] for r in rows)
        assert "Разрешение на строительство" in имена and "Изменения" in имена, имена
    finally:
        s.rollback()
        s.close()


@_pg
def test_cross_schema_duplicate_still_merges() -> None:
    """Контроль: настоящий межсхемный дубль (ОБЩИЙ key) по-прежнему склеивается.

    Ради этого случая старый ключ и вводился — новый обязан его сохранить.
    """
    s = _session()
    try:
        s.execute(text(_TEMP_OLD))
        for chunk in _migration_chunks():
            s.execute(text(chunk))
        s.execute(text(_NEW_UPSERT), _РАЗРЕШЕНИЕ)
        из_другой_схемы = dict(_РАЗРЕШЕНИЕ, source_schema="agate_ekbgo")
        s.execute(text(_NEW_UPSERT), из_другой_схемы)
        n = s.execute(text("SELECT count(*) FROM gisogd_permits")).scalar()
        assert n == 1, f"межсхемный дубль размножился: {n} строк"
    finally:
        s.rollback()
        s.close()


@_pg
def test_migration_allows_the_izmeneniya_group() -> None:
    """Контроль CHECK: после миграции doc_group='IZ' принимается, мусор — нет."""
    s = _session()
    try:
        s.execute(text(_TEMP_OLD))
        for chunk in _migration_chunks():
            s.execute(text(chunk))
        s.execute(
            text(f"INSERT INTO gisogd_permits {_INSERT_COLS}"),
            dict(_ИЗМЕНЕНИЕ, doc_group="IZ", source_key="izm-1"),
        )
        assert s.execute(text("SELECT count(*) FROM gisogd_permits")).scalar() == 1
        s.execute(text("SAVEPOINT p"))
        try:
            s.execute(
                text(f"INSERT INTO gisogd_permits {_INSERT_COLS}"),
                dict(_ИЗМЕНЕНИЕ, doc_group="XX", source_key="mus-1"),
            )
            raise AssertionError("CHECK пропустил мусорную группу 'XX'")
        except Exception as exc:
            assert "chk_gisogd_permits_doc_group" in str(exc), str(exc)[:120]
        s.execute(text("ROLLBACK TO SAVEPOINT p"))
    finally:
        s.rollback()
        s.close()