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_has_canonical_guard() -> None:
    """Tier A anchor (same building) — ранее не имел segment-guard (#1186 fix)."""
    src = inspect.getsource(est_mod._fetch_anchor_comps)
    assert _GUARD_RE.search(src), "_fetch_anchor_comps Tier A SQL lacks novostroyki guard"


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.
    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)
_REDERIVE_BOUNDS_TEXT = str(ratio_mod._REDERIVE_BOUNDS_SQL.text)
_REDERIVE_TIERED_TEXT = str(ratio_mod._REDERIVE_TIERED_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}"


def test_bounds_sql_has_guard() -> None:
    """_REDERIVE_BOUNDS_SQL (tier bounds percentiles) — guard обязателен."""
    assert _GUARD_RE.search(_REDERIVE_BOUNDS_TEXT), "_REDERIVE_BOUNDS_SQL lacks novostroyki guard"


def test_tiered_ask_ctes_have_guard() -> None:
    """_REDERIVE_TIERED_SQL: ask_tiered + ask_side_all + ask_global — минимум 3 вхождения."""
    matches = len(_GUARD_RE.findall(_REDERIVE_TIERED_TEXT))
    assert matches >= 3, (
        f"Expected ≥3 guard occurrences in _REDERIVE_TIERED_SQL "
        f"(ask_tiered + ask_side_all + 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"