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

"""#1186 — query-side guard: novostroyki строки не попадают в comp-пул вторички.

Тесты статические (SQL-string inspection) + поведенческие (_fetch_analogs mock).

Каноничный предикат: (listing_segment IS NULL OR listing_segment = 'vtorichka')
NULL = legacy вторичка до миграции 011 — отбрасывать нельзя.
"""

from __future__ import annotations

import inspect
import os
import re
from typing import Any
from unittest.mock import MagicMock

# Settings требует DATABASE_URL при инициализации (fail-fast, C-3).
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")

from app.services import estimator as est_mod
from app.tasks import asking_to_sold_ratio as ratio_mod

# ── Canonical guard predicate ─────────────────────────────────────────────────

_GUARD = "(listing_segment IS NULL OR listing_segment = 'vtorichka')"
_GUARD_RE = re.compile(
    r"\((?:\w+\.)?listing_segment\s+IS\s+NULL\s+OR\s+(?:\w+\.)?listing_segment\s*=\s*'vtorichka'\)",
    re.IGNORECASE,
)


def _norm(s: str) -> str:
    """Drop -- comments и склей пробелы для устойчивого поиска."""
    no_comments = re.sub(r"--[^\n]*", "", s)
    return re.sub(r"\s+", " ", no_comments).strip()


# ── estimator.py static checks ────────────────────────────────────────────────


def test_common_where_has_canonical_guard() -> None:
    """_COMMON_WHERE используется Tier S и Tier H — должен содержать канон-предикат."""
    assert _GUARD_RE.search(
        est_mod._COMMON_WHERE
    ), "_COMMON_WHERE lacks novostroyki guard — Tier S/H comp set contaminated"


def test_common_where_no_old_neq_form() -> None:
    """Старый вариант listing_segment <> 'novostroyki' запрещён — не охватывает иные сегменты."""
    assert "listing_segment <> 'novostroyki'" not in est_mod._COMMON_WHERE
    assert "listing_segment != 'novostroyki'" not in est_mod._COMMON_WHERE


def test_tier_w_inline_has_canonical_guard() -> None:
    """Tier W использует inline SQL, а не _COMMON_WHERE — нужен собственный guard."""
    # Получаем исходник функции _fetch_analogs.
    src = inspect.getsource(est_mod._fetch_analogs)
    # Tier W — inline блок (не через _COMMON_WHERE).
    assert _GUARD_RE.search(src), "_fetch_analogs Tier W inline SQL lacks novostroyki guard"
    # Старой формы быть не должно.
    assert "listing_segment <> 'novostroyki'" not in src
    assert "listing_segment != 'novostroyki'" not in src


def test_tier_a_anchor_gated_python_side_not_sql_guard() -> None:
    """#1774: Tier A больше НЕ хардкодит SQL-guard — он делает gated Python-side
    relaxation (впускает novostroyki только если в доме есть вторичка/NULL).

    Assert: (a) Tier C всё ещё несёт канон-guard в SQL, (b) Tier A делает
    Python-side gating — источник содержит и флаг настройки, и hardened
    secondary_count/primary_count gate (#1774, после code-review).
    """
    src = inspect.getsource(est_mod._fetch_anchor_comps)
    # (a) Tier C SQL guard сохранён (поиск находит единственное вхождение — Tier C).
    assert _GUARD_RE.search(src), "Tier C anchor SQL lost the canonical novostroyki guard"
    # (b) Tier A теперь gated Python-side (#1774).
    assert (
        "estimate_sb_tier_a_allow_primary_if_secondary_present" in src
    ), "Tier A no longer references the #1774 gating flag — gated relaxation missing"
    # Hardened gate: secondary_count ≥ primary_count (заменил has_secondary).
    assert "secondary_count" in src, "Tier A lacks secondary_count Python-side gate (#1774)"
    assert "primary_count" in src, "Tier A lacks primary_count comparison in hardened gate (#1774)"


def test_tier_c_anchor_has_canonical_guard_not_segment_param() -> None:
    """Tier C anchor — канон-предикат, не CAST(:segment AS text) (тот вариант отбрасывал NULL)."""
    src = inspect.getsource(est_mod._fetch_anchor_comps)
    # Canonical guard present (Tier C; Tier A guard снят в #1774 → gated Python-side).
    assert _GUARD_RE.search(src)
    # Старый параметрический вариант убран.
    assert "listing_segment = CAST(:segment AS text)" not in src


# ── asking_to_sold_ratio.py static checks ─────────────────────────────────────

_REDERIVE_SQL_TEXT = str(ratio_mod._REDERIVE_SQL.text)


def test_ask_side_cte_has_guard() -> None:
    """ask_side CTE в _REDERIVE_SQL (per-rooms asking медиана) — guard обязателен."""
    assert _GUARD_RE.search(
        _REDERIVE_SQL_TEXT
    ), "ask_side CTE in _REDERIVE_SQL lacks novostroyki guard"


def test_ask_global_cte_has_guard() -> None:
    """ask_global CTE в _REDERIVE_SQL (global fallback asking медиана) — guard обязателен."""
    # _REDERIVE_SQL содержит два `ask_global`-блока; ищем оба через count.
    matches = len(_GUARD_RE.findall(_REDERIVE_SQL_TEXT))
    assert (
        matches >= 2
    ), f"Expected ≥2 guard occurrences in _REDERIVE_SQL (ask_side + ask_global), got {matches}"


# ── Поведенческие тесты: _fetch_analogs (mock DB) ────────────────────────────


def _make_listing(
    *,
    source: str = "avito",
    address: str = "ул. Ленина, 1",
    distance_m: float = 100.0,
    relevance_score: float = 0.1,
    price_rub: float = 5_000_000.0,
    area_m2: float = 40.0,
    rooms: int = 2,
    listing_segment: str | None = None,
) -> dict[str, Any]:
    """Минимальный listing-dict, имитирующий вывод DB mapping."""
    return {
        "source": source,
        "source_url": f"https://{source}.ru/offer/1",
        "address": address,
        "lat": 56.838,
        "lon": 60.595,
        "rooms": rooms,
        "area_m2": area_m2,
        "floor": 3,
        "total_floors": 9,
        "price_rub": price_rub,
        "price_per_m2": price_rub / area_m2,
        "listing_date": None,
        "days_on_market": 15,
        "photo_urls": [],
        "scraped_at": None,
        "distance_m": distance_m,
        "relevance_score": relevance_score,
        "building_cadastral_number": None,
        "listing_segment": listing_segment,
    }


def _make_db_mock(rows: list[dict[str, Any]]) -> MagicMock:
    """Session mock: db.execute().mappings().all() возвращает rows."""
    db = MagicMock()
    db.execute.return_value.mappings.return_value.all.return_value = rows
    return db


def test_fetch_analogs_sql_guard_present_novostroyki_excluded() -> None:
    """SQL guard корректно записан: listing_segment='novostroyki' не должен присутствовать
    в WHERE-предикате (guard зарыт в SQL, а не в Python post-filter).

    Тест проверяет СТАТИКУ SQL внутри _fetch_analogs (Tier W path), а не runtime фильтрацию —
    runtime-фильтрацию сделала бы только интеграционная БД. Убеждаемся, что кто-то в будущем
    не переписал SQL, убрав guard.
    """
    src = inspect.getsource(est_mod._fetch_analogs)
    # Guard должен присутствовать хотя бы один раз в теле функции.
    assert _GUARD_RE.search(
        src
    ), "_fetch_analogs SQL no longer contains novostroyki guard — guard was removed!"


def test_null_segment_listing_not_excluded_by_guard() -> None:
    """listing_segment=NULL (legacy вторичка до м.011) должны ПОПАСТЬ в comps.

    Поведенческий тест: mock возвращает строку с segment=None, _fetch_analogs
    должен включить её в результат (_stratify_candidates не отфильтровывает NULL).
    """
    from app.services.estimator import _fetch_analogs

    null_seg_listing = _make_listing(listing_segment=None, distance_m=50.0, relevance_score=0.05)
    db = _make_db_mock([null_seg_listing])

    result, _fallback, _tier = _fetch_analogs(
        db, lat=56.838, lon=60.595, rooms=2, area=40.0, radius_m=1000
    )

    # NULL-segment listing должен присутствовать в результате (не отброшен Python-стороной).
    assert any(
        r.get("source") == "avito" for r in result
    ), "NULL-segment listing was unexpectedly excluded from comp set"


def test_vtorichka_segment_listing_included() -> None:
    """listing_segment='vtorichka' — явный вторичный сегмент — должен попасть в comps."""
    from app.services.estimator import _fetch_analogs

    vtor_listing = _make_listing(listing_segment="vtorichka", distance_m=80.0)
    db = _make_db_mock([vtor_listing])

    result, _, _ = _fetch_analogs(db, lat=56.838, lon=60.595, rooms=2, area=40.0, radius_m=1000)
    assert len(result) >= 1, "vtorichka-segment listing was unexpectedly excluded"


# ── Негативные assert'ы: запрещённые формы guard'а во всём estimator'е ────────


def _estimator_full_src() -> str:
    """Полный исходник модуля estimator (для статической проверки всех SQL-блоков)."""
    return inspect.getsource(est_mod)


def test_no_bare_vtorichka_eq_without_null_guard() -> None:
    """Запрет голого `listing_segment = 'vtorichka'` без IS NULL OR части.

    Такой вариант отбросит NULL (legacy вторичку до миграции 011) — баг охвата.
    """
    src = _estimator_full_src()
    # Ищем вхождения = 'vtorichka' которым НЕ предшествует IS NULL OR на той же строке.
    bad_matches = [
        line
        for line in src.splitlines()
        if "= 'vtorichka'" in line and "IS NULL OR" not in line and "--" not in line.lstrip()[:3]
    ]
    assert not bad_matches, (
        "Found bare `= 'vtorichka'` without IS NULL OR guard in estimator.py:\n"
        + "\n".join(bad_matches)
    )


def test_no_neq_novostroyki_form_in_estimator() -> None:
    """Запрет формы `listing_segment <> 'novostroyki'` / `!= 'novostroyki'` в estimator.

    Эта форма пропускает прочие сегменты (коммерция и т.п.) и не охватывает NULL.
    Канон: IS NULL OR = 'vtorichka'.
    """
    src = _estimator_full_src()
    assert (
        "<> 'novostroyki'" not in src
    ), "estimator.py contains deprecated `<> 'novostroyki'` form — use canonical guard"
    assert (
        "!= 'novostroyki'" not in src
    ), "estimator.py contains deprecated `!= 'novostroyki'` form — use canonical guard"


def test_anchor_comps_no_dead_listing_segment_param() -> None:
    """_fetch_anchor_comps не должна принимать listing_segment как параметр.

    После #1186 guard хардкожен в SQL — параметр убран, чтобы caller не мог
    случайно передать 'novostroyki' или другой сегмент и обойти guard.
    """
    import inspect as _inspect

    sig = _inspect.signature(est_mod._fetch_anchor_comps)
    assert "listing_segment" not in sig.parameters, (
        "_fetch_anchor_comps still has `listing_segment` parameter — "
        "it was removed in #1186 review fixup; guard is hardcoded in SQL"
    )


def test_fetch_anchor_comps_tier_c_canonical_guard_not_parametric() -> None:
    """Tier C в _fetch_anchor_comps использует канон-guard, а не параметрическое сравнение.

    Параметрическая форма `listing_segment = CAST(:segment AS text)` исключала NULL
    и давала caller'у возможность выбрать произвольный сегмент (включая novostroyki).
    """
    src = inspect.getsource(est_mod._fetch_anchor_comps)
    assert "CAST(:segment AS text)" not in src, (
        "_fetch_anchor_comps Tier C still uses parametric segment filter — "
        "must be hardcoded canonical guard"
    )
    # И канон-guard на месте (Tier C-блок).
    assert _GUARD_RE.search(
        src
    ), "_fetch_anchor_comps lacks canonical guard after removing parametric form"