gendesign/tradein-mvp/backend/tests/test_rosreestr_dedup_key.py

"""Unit-тесты для формата dedup_hash rosreestr ДКП-сделок (#576).

До #576 dedup_hash хранил md5('ros:dkp:' || id) — collision-prone, не human-readable,
дедуп не восстановить из значения. После — плоский натуральный ключ 'ros:dkp:N'
(инъективный, zero-collision, обратимый в исходный rosreestr id).

Чисто-юнит: ассертим SQL-текст, эмитируемый import_rosreestr_dkp (через inspect.getsource),
плюс содержимое backfill-миграции 077 — без живого FDW / DB-фикстуры.
"""

import inspect
import os
from pathlib import Path

# Импорт app.services.scheduler тянет app.core.config.Settings → требует DATABASE_URL.
# Ставим заглушку ДО импорта (как в test_scheduler.py) — тест чисто-статический, БД не трогает.
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

from app.services import scheduler

_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
_MIGRATION_077 = _SQL_DIR / "077_dedup_hash_plain_key_backfill.sql"

# Источник функции импорта — читаем один раз.
_IMPORT_SRC = inspect.getsource(scheduler.import_rosreestr_dkp)
# Тело без docstring: docstring упоминает старое md5-выражение в пояснительных целях,
# поэтому negative-ассерты («md5 больше нет») гоняем по коду, а не по докстрингу.
_IMPORT_BODY = _IMPORT_SRC.split('"""', 2)[-1]


def test_import_uses_plain_natural_key_not_md5() -> None:
    """SELECT-выражение dedup_hash — плоский ключ 'ros:dkp:' || id, без md5()."""
    # Плоский натуральный ключ присутствует как SELECT-выражение dedup_hash.
    assert "'ros:dkp:' || CAST(id AS text)           AS dedup_hash" in _IMPORT_SRC
    # md5-обёртки натурального ключа в коде больше нет (старое выражение #549).
    assert "md5('ros:dkp:'" not in _IMPORT_BODY
    assert "md5(" not in _IMPORT_BODY


def test_import_populates_source_id() -> None:
    """Новые строки сохраняют исходный rosreestr id в source_id (дедуп переустанавливаем)."""
    # source_id попадает в SELECT, в список колонок INSERT и в params.
    assert "AS source_id_src" in _IMPORT_SRC
    assert "source, dedup_hash, source_id," in _IMPORT_SRC
    assert "CAST(:source_id AS text)" in _IMPORT_SRC
    assert '"source_id": str(row["source_id_src"])' in _IMPORT_SRC


def test_import_keeps_on_conflict_dedup_hash() -> None:
    """Дедуп по-прежнему через ON CONFLICT (dedup_hash) DO NOTHING."""
    assert "ON CONFLICT (dedup_hash) DO NOTHING" in _IMPORT_SRC


def test_import_uses_psycopg_v3_casts_not_double_colon() -> None:
    """psycopg v3: параметры приводятся через CAST(:x AS type), никогда :x::type.

    Литеральные ::-касты на колонках FDW (round(deal_price)::bigint и т.п.) допустимы —
    запрет касается только :param::type, который psycopg v3 ломает.
    """
    import re

    # Ни один bind-параметр (:name) не должен использовать ::type.
    assert not re.search(r":\w+::", _IMPORT_SRC)
    # Параметры приводятся через CAST(:param AS type).
    assert "CAST(:since AS date)" in _IMPORT_SRC
    assert "CAST(:last_id AS bigint)" in _IMPORT_SRC
    assert "CAST(:source_id AS text)" in _IMPORT_SRC


# ── backfill-миграция 077 ────────────────────────────────────────────────────


def test_migration_077_exists() -> None:
    assert _MIGRATION_077.is_file(), f"missing migration: {_MIGRATION_077}"


def test_migration_077_converts_md5_to_plain_key() -> None:
    """077 матчит существующие строки по md5 и пишет плоский ключ + source_id."""
    sql = _MIGRATION_077.read_text("utf-8")
    # Восстанавливаем id из FDW по тому же md5, что писался раньше.
    assert "md5('ros:dkp:' || CAST(id AS text))" in sql
    assert "FROM gendesign_rosreestr_deals" in sql
    # Целевые значения — плоский ключ и source_id.
    assert "dedup_hash = 'ros:dkp:' || CAST(src.id AS text)" in sql
    assert "source_id  = CAST(src.id AS text)" in sql
    # Только rosreestr-строки, матч по старому md5-значению.
    assert "d.source = 'rosreestr'" in sql
    assert "d.dedup_hash = src.h" in sql
    # Транзакционно.
    assert "BEGIN;" in sql and "COMMIT;" in sql


def test_migration_077_shared_filter_matches_live_import() -> None:
    """Дедуп-релевантные фильтры src CTE миграции 077 совпадают с живым импортом.

    Исключение — city ILIKE (см. test_live_import_dropped_ekb_city_filter ниже):
    077 backfill'ил ЕКБ-строки под EKB-only scope того времени; живой импорт расширен
    на всю Свердловскую область (region_code=66, все города), поэтому city-фильтр из
    живого импорта СНЯТ намеренно. Остальные клозы обязаны совпадать байт-в-байт,
    иначе 077 конвертировал бы не тот набор строк.
    """
    sql = _MIGRATION_077.read_text("utf-8")
    for clause in (
        "region_code = 66",
        "realestate_type_code = '002001003000'",
        "area BETWEEN 18 AND 200",
        "deal_price BETWEEN 1000000 AND 100000000",
        "street IS NOT NULL AND trim(street) <> ''",
        "doc_type = 'ДКП'",
    ):
        assert clause in sql, f"missing filter clause in migration: {clause!r}"
        assert clause in _IMPORT_SRC, f"missing filter clause in import: {clause!r}"


def test_live_import_dropped_ekb_city_filter() -> None:
    """Живой импорт БОЛЬШЕ не фильтрует по городу — Mera расширена на всю обл. 66.

    Историческая миграция 077 (уже применена на прод, трогать нельзя) сохраняет свой
    EKB-only city-фильтр; живой импорт — нет. Guard против случайного возврата
    ЕКБ-скоупа, который снова отрезал бы +47k не-ЕКБ сделок.
    """
    sql = _MIGRATION_077.read_text("utf-8")
    # 077 — исторический EKB-scope, city-фильтр там остаётся неизменным.
    assert "city ILIKE '%катеринбург%'" in sql
    # Живой импорт city-фильтр снял намеренно (region_code=66 = вся область).
    assert "ILIKE '%катеринбург%'" not in _IMPORT_SRC