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

"""#1871 P2 — radius-tier (source, source_id) dedup в _fetch_analogs.

Radius-путь кэпит только per-address (rn_addr <= MAX_ANALOGS_PER_ADDRESS), но
(source, source_id)-дубли делят один address и выживают на разных rn_addr рангах →
раздувают n_analogs. Anchor-путь дедупит по (source, source_id); radius — нет.

Эти тесты проверяют SQL-фрагмент (rn_dup-окно всегда; `AND rn_dup = 1` фильтр за
флагом) во ВСЕХ четырёх тирах (S-canonical, S-fallback, H, W) + дедупную семантику
окна на уровне Python-симуляции (freshest scraped_at, namespaced ключ, NULL-guard).
Полный radius-путь требует PostGIS+БД, поэтому SQL проверяется на сгенерированном
тексте (как test_estimator_price_trend_dedup.py).
"""

from __future__ import annotations

import os
from unittest.mock import MagicMock, patch

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

import pytest

import app.services.estimator as est


def _capture_tier_sql(*, dedup_enabled: bool) -> list[str]:
    """Прогоняет _fetch_analogs с mock db, возвращает SQL-текст каждого тира.

    target_house_id + short_addr + year/floors заданы так, чтобы SQL ВСЕХ четырёх
    тиров отрендерился (каждый тир возвращает [] → fallthrough к следующему).
    """
    captured: list[str] = []
    db = MagicMock()

    def side_effect(*args, **kwargs):  # type: ignore[no-untyped-def]
        captured.append(str(args[0].text))
        result = MagicMock()
        result.mappings.return_value.all.return_value = []
        return result

    db.execute.side_effect = side_effect

    with patch.object(est.settings, "estimate_radius_dedup_enabled", dedup_enabled):
        est._fetch_analogs(
            db,
            lat=56.83,
            lon=60.6,
            rooms=2,
            area=50.0,
            radius_m=2000,
            full_address="г Екатеринбург, ул Малышева, д 30",
            year_built=2010,
            house_type="монолит",
            total_floors=20,
            target_house_id=123,
        )
    return captured


# ---------------------------------------------------------------------------
# Флаг по умолчанию
# ---------------------------------------------------------------------------


def test_radius_dedup_flag_default_true() -> None:
    from app.core.config import settings

    assert settings.estimate_radius_dedup_enabled is True


# ---------------------------------------------------------------------------
# SQL-фрагмент во всех тирах
# ---------------------------------------------------------------------------


def test_all_four_tiers_render() -> None:
    sqls = _capture_tier_sql(dedup_enabled=True)
    assert len(sqls) == 4, "ожидаем S-canonical, S-fallback, H, W"


def test_rn_dup_window_present_in_every_tier_when_enabled() -> None:
    sqls = _capture_tier_sql(dedup_enabled=True)
    for i, sql in enumerate(sqls):
        assert "AS rn_dup" in sql, f"tier#{i} без rn_dup-окна"
        assert "rn_dup = 1" in sql, f"tier#{i} без `AND rn_dup = 1` фильтра"


def test_dup_filter_dropped_when_disabled_but_window_stays() -> None:
    """Флаг OFF: фильтр `AND rn_dup = 1` убран, окно остаётся (безвредно)."""
    sqls = _capture_tier_sql(dedup_enabled=False)
    assert len(sqls) == 4
    for i, sql in enumerate(sqls):
        assert "AS rn_dup" in sql, f"tier#{i}: окно должно оставаться даже при OFF"
        assert "rn_dup = 1" not in sql, f"tier#{i}: фильтр должен исчезнуть при OFF"


def test_rn_dup_key_namespaced_and_freshest() -> None:
    """Ключ окна namespaced (id:/url:/ctid:) + ORDER BY scraped_at DESC."""
    sql = _capture_tier_sql(dedup_enabled=True)[0]
    assert "PARTITION BY source" in sql
    assert "'id:' || source_id::text" in sql
    assert "'url:' || source_url" in sql
    assert "'ctid:' || ctid::text" in sql
    # freshest: окно сортируется по scraped_at DESC → rn_dup=1 это свежайшая строка
    win_start = sql.find("AS rn_dup")
    win_region = sql[max(0, win_start - 400) : win_start]
    assert "ORDER BY scraped_at DESC" in win_region


# ---------------------------------------------------------------------------
# psycopg3-инвариант: column::type OK, :bind::type запрещён
# ---------------------------------------------------------------------------


def test_no_bind_param_double_colon_cast() -> None:
    """В сгенерированном SQL нет :name::type (только column::type, что легально)."""
    import re

    bad = re.compile(r":[a-z_]+::[a-z]")
    for i, sql in enumerate(_capture_tier_sql(dedup_enabled=True)):
        assert not bad.search(sql), f"tier#{i}: найден запрещённый :bind::type"


# ---------------------------------------------------------------------------
# Семантика дедупа окна (Python-симуляция row_number() OVER PARTITION/ORDER)
# ---------------------------------------------------------------------------


def _simulate_rn_dup(rows: list[dict]) -> list[dict]:
    """Эмулирует SQL-окно rn_dup=1: per (source, key) freshest scraped_at.

    Зеркалит SQL: ключ = source + namespaced(id:source_id | url:source_url |
    ctid:ctid). NULL source_id+source_url → разные ctid → разные группы (обе живут).
    """
    groups: dict[tuple, dict] = {}
    for r in rows:
        if r.get("source_id") is not None:
            key = (r["source"], "id", r["source_id"])
        elif r.get("source_url") is not None:
            key = (r["source"], "url", r["source_url"])
        else:
            key = (r["source"], "ctid", r["ctid"])
        cur = groups.get(key)
        if cur is None or r["scraped_at"] > cur["scraped_at"]:
            groups[key] = r
    return list(groups.values())


def test_same_source_source_id_collapses_to_freshest() -> None:
    rows = [
        {"source": "yandex", "source_id": "A1", "source_url": None, "ctid": "(0,1)",
         "scraped_at": 100, "price_per_m2": 150_000},
        {"source": "yandex", "source_id": "A1", "source_url": None, "ctid": "(0,2)",
         "scraped_at": 200, "price_per_m2": 151_000},
    ]
    out = _simulate_rn_dup(rows)
    assert len(out) == 1
    assert out[0]["scraped_at"] == 200, "должна выжить freshest (scraped_at=200)"


def test_different_source_id_both_survive() -> None:
    rows = [
        {"source": "cian", "source_id": "X", "source_url": None, "ctid": "(0,1)",
         "scraped_at": 100, "price_per_m2": 140_000},
        {"source": "cian", "source_id": "Y", "source_url": None, "ctid": "(0,2)",
         "scraped_at": 100, "price_per_m2": 145_000},
    ]
    out = _simulate_rn_dup(rows)
    assert len(out) == 2


def test_same_source_id_different_source_both_survive() -> None:
    """(source, source_id) — PARTITION включает source: одинаковый id у разных
    источников НЕ схлопывается."""
    rows = [
        {"source": "cian", "source_id": "Z", "source_url": None, "ctid": "(0,1)",
         "scraped_at": 100, "price_per_m2": 140_000},
        {"source": "yandex", "source_id": "Z", "source_url": None, "ctid": "(0,2)",
         "scraped_at": 100, "price_per_m2": 145_000},
    ]
    out = _simulate_rn_dup(rows)
    assert len(out) == 2


def test_null_source_id_null_url_both_survive_via_ctid() -> None:
    """CASE-guard: 2 строки с NULL source_id И NULL source_url → ctid-tiebreak →
    обе сохранены (не схлопываются по NULL)."""
    rows = [
        {"source": "rosreestr", "source_id": None, "source_url": None, "ctid": "(0,1)",
         "scraped_at": 100, "price_per_m2": 130_000},
        {"source": "rosreestr", "source_id": None, "source_url": None, "ctid": "(0,2)",
         "scraped_at": 100, "price_per_m2": 132_000},
    ]
    out = _simulate_rn_dup(rows)
    assert len(out) == 2


def test_null_source_id_falls_back_to_source_url() -> None:
    """source_id NULL, но одинаковый source_url → схлопываются (url-ключ)."""
    rows = [
        {"source": "avito", "source_id": None, "source_url": "http://a/1", "ctid": "(0,1)",
         "scraped_at": 100, "price_per_m2": 120_000},
        {"source": "avito", "source_id": None, "source_url": "http://a/1", "ctid": "(0,2)",
         "scraped_at": 300, "price_per_m2": 121_000},
    ]
    out = _simulate_rn_dup(rows)
    assert len(out) == 1
    assert out[0]["scraped_at"] == 300


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(pytest.main([__file__, "-q"]))