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

"""#3051 — полосы цен по округам Москвы + региональный потолок ppm².

Что проверяется:
1. Ключ города сделки (deal_city_key / deal_city_key_sql) — один на derivation
   полос и на все читающие места estimator'а.
2. Регион 66 не двигается: src_city у его сделок пуст → ключ равен city.
3. Новый потолок не срезает дорогой московский округ, старый глобальный —
   срезал бы.
4. Полосы региона 66 по новой формуле потолка совпадают со старым литералом.

Живого Postgres нет — SQL-инварианты проверяются по тексту запроса, поведение
фильтра — на подменённой Session (паттерн из tests/test_dkp_corridor_as_of_2846.py).
Авторитетная проверка инварианта региона 66 сделана на проде 2026-09-10 прогоном
новой derivation в режиме SELECT: 383 строки против 383 текущих, все совпали по
(ppm2_min, ppm2_max, n_deals, tier); сделок региона 66, где ключ != city, — 0.
"""

from __future__ import annotations

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

import pytest

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

from app.services import estimator as est
from app.services.deal_city_key import (
    DEAL_CITY_KEY_COLUMN,
    deal_city_band_bounds_sql,
    deal_city_band_join_sql,
    deal_city_key,
    deal_city_key_sql,
    resolve_city_band,
)
from app.tasks.deal_city_price_bands_refresh import (
    _REDERIVE_SQL,
    REGION_CEILING_CAP_Q,
    REGION_CEILING_FLOOR,
    REGION_CEILING_MEDIAN_MULT,
    REGION_CEILING_MEDIAN_Q,
    region_ppm2_max,
)

# Замеры прода 2026-09-10, на которых калибровался потолок.
_R66_P50 = 52_706
_R66_P9999 = 615_312
_R77_P50 = 294_457
_R77_P9999 = 1_944_535
_R77_CEILING = 1_766_742  # ПИН прода: множитель * p50 региона 77 при текущих константах
_MAX_OKRUG_P99 = 1_405_882  # самый дорогой округ Москвы


def _sql_code(sql: str) -> str:
    """Текст запроса без строк-комментариев — числа из комментариев не считаем."""
    return "\n".join(ln for ln in sql.splitlines() if not ln.strip().startswith("--"))


# ── 1. Ключ города ───────────────────────────────────────────────────────────


def test_key_region66_equals_city() -> None:
    """Свердловская область: src_city пуст у всех сделок → ключ равен city."""
    assert deal_city_key({"city": "Асбест", "raw_payload": {}}) == "Асбест"
    assert deal_city_key({"city": "Асбест", "raw_payload": None}) == "Асбест"
    assert deal_city_key({"city": "Асбест"}) == "Асбест"
    # NULLIF(x, '') в SQL: пустая строка — то же, что отсутствие значения.
    assert deal_city_key({"city": "Асбест", "raw_payload": {"src_city": ""}}) == "Асбест"


def test_key_moscow_with_src_city_is_okrug() -> None:
    row = {"city": "Москва", "raw_payload": {"src_city": "муниципальный округ Хамовники"}}
    assert deal_city_key(row) == "муниципальный округ Хамовники"


def test_key_moscow_without_src_city_is_moscow() -> None:
    """6.73% московских сделок без src_city → собственная строка-фолбэк 'Москва'."""
    assert deal_city_key({"city": "Москва", "raw_payload": {"src_city": None}}) == "Москва"
    assert deal_city_key({"city": "Москва", "raw_payload": {}}) == "Москва"


def test_key_column_wins_over_raw_payload() -> None:
    """Готовую колонку city_key (её считает SQL) не переопределяем питоном."""
    row = {
        DEAL_CITY_KEY_COLUMN: "муниципальный округ Метрогородок",
        "city": "Москва",
        "raw_payload": {"src_city": "муниципальный округ Хамовники"},
    }
    assert deal_city_key(row) == "муниципальный округ Метрогородок"


def test_key_sql_shape() -> None:
    assert deal_city_key_sql("d") == "COALESCE(NULLIF(d.raw_payload->>'src_city', ''), d.city)"
    assert deal_city_key_sql("") == "COALESCE(NULLIF(raw_payload->>'src_city', ''), city)"


# ── 2. SQL-инварианты derivation и читающих запросов ─────────────────────────


def test_rederive_sql_groups_by_key_and_has_regional_ceiling() -> None:
    sql = str(_REDERIVE_SQL)
    key = deal_city_key_sql("")
    assert f"GROUP BY region_code, {key}" in sql
    assert f"{key} AS city" in sql
    # Потолок стал региональным: литерала-константы 800000 в ветках tiered больше
    # нет, он остался только якорем внутри GREATEST в region_stats.
    assert "region_ppm2_max" in sql
    assert f"percentile_cont({REGION_CEILING_CAP_Q})" in sql
    code = _sql_code(sql)
    # Якорь GREATEST в region_stats — единственное место с этим числом.
    assert code.count(str(REGION_CEILING_FLOOR)) == 1
    assert "LEAST(c.ppm2_p99, r.region_ppm2_max)" in sql
    # p99 города больше не клампится литералом на этапе city_stats.
    assert "LEAST(round(percentile_cont(0.99)" not in sql


def test_estimator_reads_bands_by_same_key() -> None:
    """Все три читающих места используют ТО ЖЕ выражение ключа, что derivation."""
    src = open(est.__file__, encoding="utf-8").read()
    # Оба SQL-джойна ДКП-коридора (street и city-wide widen) подставляют ОДНУ
    # константу, а не свою копию выражения; сама константа — из deal_city_key.
    assert src.count("{_DEAL_CITY_BAND_JOIN_D}") == 2
    assert src.count("{_DEAL_CITY_BAND_BOUNDS}") == 2
    assert est._DEAL_CITY_BAND_JOIN_D == deal_city_band_join_sql("d", indent=" " * 20)
    assert est._DEAL_CITY_BAND_BOUNDS == deal_city_band_bounds_sql(indent=" " * 22)
    assert est._DEAL_CITY_KEY_SQL_PLAIN == deal_city_key_sql("")
    # Питоновский путь: ключ считает SQL в SELECT сделок.
    assert "{_DEAL_CITY_KEY_SQL_PLAIN} AS {DEAL_CITY_KEY_COLUMN}" in src
    # Одноступенчатых джойна и границ (только b) не осталось.
    assert "b.city = d.city" not in src
    assert "COALESCE(b.ppm2_min, CAST(" not in src


# ── 3. Потолок ───────────────────────────────────────────────────────────────


def test_ceiling_formula_is_single_source() -> None:
    """СТОРОЖ: SQL-выражение потолка собрано из ТЕХ ЖЕ констант, что region_ppm2_max().

    Раньше формула жила двумя копиями (SQL + локальная копия в этом файле), и
    подмена множителя 6 на 3 оставляла все тесты зелёными, роняя потолок Москвы
    с 1766742 до 883371. Теперь каждая константа обязана встретиться в собранном
    SQL ровно столько раз, сколько её кладёт сборка, а питоновская сторона — та
    же функция region_ppm2_max из модуля derivation, а не копия.
    """
    code = _sql_code(str(_REDERIVE_SQL))
    assert code.count(str(REGION_CEILING_FLOOR)) == 1  # якорь GREATEST
    assert code.count(f"percentile_cont({REGION_CEILING_CAP_Q})") == 1  # шапка p99.99
    median_sql = f"{REGION_CEILING_MEDIAN_MULT} * round(percentile_cont({REGION_CEILING_MEDIAN_Q})"
    assert code.count(median_sql) == 1  # множитель медианы
    # Обе ветви питоновской функции наблюдаемы и построены из тех же чисел.
    assert region_ppm2_max(10**9, 10**6) == REGION_CEILING_MEDIAN_MULT * 10**6
    assert region_ppm2_max(0, 0) == REGION_CEILING_FLOOR


def test_ceiling_region66_unchanged() -> None:
    """Регион 66: формула вырождается ровно в прежний литерал 800000."""
    assert region_ppm2_max(_R66_P9999, _R66_P50) == 800_000
    # Запас: чтобы потолок сдвинулся, нужны ОДНОВРЕМЕННО медиана > 133 333 и
    # p99.99 > 800 000 (сейчас 52 706 и 615 312).
    assert REGION_CEILING_MEDIAN_MULT * _R66_P50 < REGION_CEILING_FLOOR
    assert _R66_P9999 < REGION_CEILING_FLOOR


def test_ceiling_region77_lifts_and_is_guarded() -> None:
    assert region_ppm2_max(_R77_P9999, _R77_P50) == _R77_CEILING
    # Шапка p99.99 не даёт множителю медианы разогнать потолок бесконечно.
    assert region_ppm2_max(900_000, _R77_P50) == 900_000
    # Якорь 800000 не даёт потолку упасть ниже исторического даже при обвале цен.
    assert region_ppm2_max(100_000, 10_000) == REGION_CEILING_FLOOR


def test_new_ceiling_does_not_cut_expensive_okrug() -> None:
    """Самый дорогой округ (p99=1 405 882) целиком помещается под потолок 77."""
    assert _MAX_OKRUG_P99 < region_ppm2_max(_R77_P9999, _R77_P50)
    assert _MAX_OKRUG_P99 > 800_000  # старый литерал резал бы его


def test_plausible_deal_uses_okrug_band() -> None:
    bands = {
        (77, "муниципальный округ Хамовники"): (100_000, _MAX_OKRUG_P99),
        (77, "Москва"): (22_475, 772_165),
        (66, "Асбест"): (8_000, 254_831),
    }
    # Дорогой округ: 1.3 М ₽/м² — легитимный рынок, проходит.
    assert est._is_plausible_deal(
        1_300_000,
        5,
        12,
        city="муниципальный округ Хамовники",
        city_fallback="Москва",
        bands=bands,
        region_code=77,
    )
    # Тот же ppm² под глобальным потолком (DEAL_MAX_PPM2=800000) был бы отброшен.
    assert not est._is_plausible_deal(
        1_300_000, 5, 12, city="Москва", city_fallback="Москва", bands={}, region_code=77
    )
    # Строка-фолбэк 'Москва' — осмысленная полоса, а не ЕКБ-константы:
    # 30 000 ₽/м² ниже DEAL_MIN_PPM2=50000, но внутри московского фолбэка.
    assert est._is_plausible_deal(
        30_000, 5, 12, city="Москва", city_fallback="Москва", bands=bands, region_code=77
    )
    assert not est._is_plausible_deal(
        30_000, 5, 12, city="Москва", city_fallback="Москва", bands={}, region_code=77
    )
    # Регион 66 — прежнее поведение.
    assert est._is_plausible_deal(
        41_000, 3, 5, city="Асбест", city_fallback="Асбест", bands=bands, region_code=66
    )
    assert not est._is_plausible_deal(
        300_000, 3, 5, city="Асбест", city_fallback="Асбест", bands=bands, region_code=66
    )
    # Екатеринбург намеренно не в таблице → глобальные константы.
    assert est._is_plausible_deal(
        120_000,
        3,
        5,
        city="Екатеринбург",
        city_fallback="Екатеринбург",
        bands=bands,
        region_code=66,
    )


# ── 4. _fetch_deals: ключ доезжает до фильтра и не течёт в ответ ─────────────


def _deal_row(**over: Any) -> dict[str, Any]:
    row = {
        "source": "rosreestr",
        "address": "Москва, ул. Остоженка, 1",
        "lat": 55.74,
        "lon": 37.6,
        "rooms": 2,
        "area_m2": 80.0,
        "floor": 5,
        "total_floors": 12,
        "price_rub": 104_000_000.0,
        "price_per_m2": 1_300_000.0,
        "city": "Москва",
        "region_code": 77,
        DEAL_CITY_KEY_COLUMN: "муниципальный округ Хамовники",
        "deal_date": None,
        "days_on_market": None,
        "cadastral_number": None,
        "distance_m": 100.0,
    }
    row.update(over)
    return row


def _db(deal_rows: list[dict[str, Any]], band_rows: list[dict[str, Any]]) -> Any:
    def _execute(query: Any, params: dict[str, Any] | None = None) -> MagicMock:
        result = MagicMock()
        sql = str(query)
        if "FROM deal_city_price_bands" in sql:
            result.mappings.return_value.all.return_value = band_rows
        else:
            result.mappings.return_value.all.return_value = deal_rows
        return result

    db = MagicMock()
    db.execute.side_effect = _execute
    return db


_BANDS_ROWS = [
    {
        "region_code": 77,
        "city": "муниципальный округ Хамовники",
        "ppm2_min": 100_000,
        "ppm2_max": _MAX_OKRUG_P99,
    },
    {"region_code": 77, "city": "Москва", "ppm2_min": 22_475, "ppm2_max": 772_165},
    {"region_code": 66, "city": "Асбест", "ppm2_min": 8_000, "ppm2_max": 254_831},
]


def _fetch(
    rows: list[dict[str, Any]], band_rows: list[dict[str, Any]] | None = None
) -> list[dict[str, Any]]:
    est._city_price_bands_cache = None
    try:
        # `rooms` у `_fetch_deals` больше нет: #3256 убрал параметр, а не просто
        # перестал класть его в SQL — сделки Росреестра комнатность не несут.
        # Тест писался на ветке, отведённой до того тикета, и разъехался с main
        # молча: текстового конфликта в мерже не было, красным стало только на
        # post-merge прогоне (#3440).
        return est._fetch_deals(
            _db(rows, _BANDS_ROWS if band_rows is None else band_rows),
            lat=55.74,
            lon=37.6,
            area=80.0,
            radius_m=1000,
        )
    finally:
        est._city_price_bands_cache = None


def test_fetch_deals_keeps_expensive_okrug_and_hides_key_column() -> None:
    out = _fetch([_deal_row()])
    assert len(out) == 1
    # Служебная колонка не должна утечь в actual_deals ответа API.
    assert DEAL_CITY_KEY_COLUMN not in out[0]
    assert out[0]["city"] == "Москва"


def test_fetch_deals_drops_same_price_under_moscow_fallback_band() -> None:
    """Та же сделка без src_city ключуется как 'Москва' → 1.3 М ₽/м² вне полосы."""
    assert _fetch([_deal_row(**{DEAL_CITY_KEY_COLUMN: "Москва"})]) == []


def test_fetch_deals_region66_unaffected() -> None:
    row = _deal_row(
        city="Асбест",
        region_code=66,
        price_per_m2=41_000.0,
        price_rub=3_280_000.0,
        **{DEAL_CITY_KEY_COLUMN: "Асбест"},
    )
    out = _fetch([row])
    assert len(out) == 1
    assert DEAL_CITY_KEY_COLUMN not in out[0]


# ── 5. Двухступенчатый поиск полосы (#3051, окно между деплоем и рефрешем) ────


def test_band_lookup_order_okrug_then_city_then_globals() -> None:
    """Ступени: полоса округа → полоса города сделки → глобальные ЕКБ-константы."""
    default = (est.DEAL_MIN_PPM2, est.DEAL_MAX_PPM2)
    bands = {
        (77, "муниципальный округ Хамовники"): (100_000, _MAX_OKRUG_P99),
        (77, "Москва"): (22_475, 772_165),
    }
    okrug = "муниципальный округ Хамовники"
    assert resolve_city_band(bands, 77, okrug, "Москва", default) == (100_000, _MAX_OKRUG_P99)
    # Строки округа ещё нет (ночной рефреш не прогонялся) → городская полоса.
    assert resolve_city_band(bands, 77, "муниципальный округ Некрасовка", "Москва", default) == (
        22_475,
        772_165,
    )
    # Нет ни округа, ни города — только тогда глобальные константы.
    assert resolve_city_band(bands, 77, "округ Некрасовка", "Тверь", default) is default
    assert resolve_city_band({}, 77, okrug, "Москва", default) is default
    assert resolve_city_band(None, 77, okrug, "Москва", default) is default
    # Регион 66: ключ ступени 1 равен ключу ступени 2 → один и тот же результат.
    r66 = {(66, "Асбест"): (8_000, 254_831)}
    assert resolve_city_band(r66, 66, "Асбест", "Асбест", default) == (8_000, 254_831)
    assert resolve_city_band(r66, 66, "Екатеринбург", "Екатеринбург", default) is default


def test_sql_band_join_and_bounds_are_two_step() -> None:
    """В SQL двухступенчатость — второй LEFT JOIN и трёхаргументный COALESCE."""
    join = est._DEAL_CITY_BAND_JOIN_D
    assert join.count("LEFT JOIN deal_city_price_bands") == 2
    assert "AND b.city = " + deal_city_key_sql("d") in join  # ступень 1 — округ
    assert "AND bc.city = d.city" in join  # ступень 2 — город сделки
    bounds = est._DEAL_CITY_BAND_BOUNDS
    assert "COALESCE(b.ppm2_min, bc.ppm2_min, CAST(:ppm_min AS int))" in bounds
    assert "COALESCE(b.ppm2_max, bc.ppm2_max, CAST(:ppm_max AS int))" in bounds


def test_plausible_deal_degrades_to_city_band_not_ekb_constants() -> None:
    """Окно деплоя: по региону 77 в таблице ровно одна строка 'Москва'."""
    bands = {(77, "Москва"): (22_475, 772_165)}
    okrug = "муниципальный округ Некрасовка"
    # 30 000 ₽/м² ниже DEAL_MIN_PPM2=50000, но внутри московского фолбэка → keep.
    assert est._is_plausible_deal(
        30_000, 5, 12, city=okrug, city_fallback="Москва", bands=bands, region_code=77
    )
    # Явный отказ от второй ступени: та же сделка проваливается в ЕКБ-калибровку
    # и отбрасывается. Пропуск city_fallback здесь дал бы TypeError — см. тест ниже.
    assert not est._is_plausible_deal(
        30_000, 5, 12, city=okrug, city_fallback=None, bands=bands, region_code=77
    )
    # Городской потолок при этом продолжает работать.
    assert not est._is_plausible_deal(
        900_000, 5, 12, city=okrug, city_fallback="Москва", bands=bands, region_code=77
    )
    # Регион 66: обе ступени — один ключ, поведение прежнее.
    r66 = {(66, "Асбест"): (8_000, 254_831)}
    assert est._is_plausible_deal(
        41_000, 3, 5, city="Асбест", city_fallback="Асбест", bands=r66, region_code=66
    )
    assert not est._is_plausible_deal(
        300_000, 3, 5, city="Асбест", city_fallback="Асбест", bands=r66, region_code=66
    )


_MOSCOW_ONLY_BANDS = [
    {"region_code": 77, "city": "Москва", "ppm2_min": 22_475, "ppm2_max": 772_165}
]


def test_fetch_deals_before_first_refresh_uses_city_band() -> None:
    """_fetch_deals в том же окне: округ без своей строки судится полосой 'Москва'."""
    cheap = _deal_row(price_per_m2=30_000.0, price_rub=2_400_000.0)
    assert len(_fetch([cheap], _MOSCOW_ONLY_BANDS)) == 1  # ЕКБ-пол 50000 отбросил бы
    assert _fetch([cheap], []) == []  # без полос вообще — те самые ЕКБ-константы
    # Дорогая сделка того же округа пока отсекается городским потолком 772165 —
    # окружная полоса (до 1 405 882) появится после первого ночного рефреша.
    assert _fetch([_deal_row()], _MOSCOW_ONLY_BANDS) == []


def test_fetch_deals_region66_two_step_is_identical() -> None:
    """Регион 66: src_city пуст → обе ступени дают 'Асбест', результат прежний."""
    row = _deal_row(
        city="Асбест",
        region_code=66,
        price_per_m2=41_000.0,
        price_rub=3_280_000.0,
        **{DEAL_CITY_KEY_COLUMN: "Асбест"},
    )
    r66 = [{"region_code": 66, "city": "Асбест", "ppm2_min": 8_000, "ppm2_max": 254_831}]
    assert len(_fetch([row], r66)) == 1
    assert _fetch([row], []) == []  # 41 000 < DEAL_MIN_PPM2 — глобальная ступень жива


# ── 6. Предикат исключения ЕКБ и обязательность второй ступени ───────────────


def test_rederive_excludes_ekb_by_city_key_not_raw_column() -> None:
    """Исключение ЕКБ судится тем же выражением, по которому идёт GROUP BY.

    Раньше предикат смотрел на СЫРУЮ колонку city, а группировка — уже на ключ.
    Совпадение держалось на данных: прод 2026-09-11, регион 66 — 108 623 сделки,
    исключено по city 55 749, по ключу 55 749, расхождение 0. Появись источник с
    src_city='Екатеринбург' у сделки с другим city — сырой предикат пропустил бы
    её в derivation, завёл строку полосы 'Екатеринбург', и ступень 1 нашла бы её
    для настоящих ЕКБ-сделок, сломав намеренное исключение.
    """
    sql = str(_REDERIVE_SQL)
    key = deal_city_key_sql("")
    assert sql.count(f"NOT (region_code = 66 AND {key} = 'Екатеринбург')") == 2
    assert "region_code = 66 AND city = 'Екатеринбург'" not in sql


def test_rederive_requires_nonnull_city_key_not_raw_column() -> None:
    """Непустоту судит выражение КЛЮЧА, а не сырая колонка city.

    Раньше оба CTE фильтровали `AND city IS NOT NULL`, хотя GROUP BY и предикат
    ЕКБ уже жили на ключе. Сделка с непустым src_city и NULL в city даёт валидный
    ключ, но выпадала из derivation ЦЕЛИКОМ — и из своей городской строки, и из
    региональной статистики, молча занижая n_deals и перцентили региона, включая
    потолок. Прод 2026-09-11: в популяции 321 560 сделок, city IS NULL — 0 строк,
    поэтому сегодняшний результат не меняется ни по одному региону (66 — 52 874
    строки, p1=15345, p50=52706, p99.99=615312; 77 — 212 937 строк,
    34221 / 294457 / 1944535; обе формы дают одно и то же, расхождение 0).
    """
    code = _sql_code(str(_REDERIVE_SQL))
    key = deal_city_key_sql("")
    assert code.count(f"AND {key} IS NOT NULL") == 2
    assert "AND city IS NOT NULL" not in code


def test_plausible_deal_requires_city_fallback_when_bands_given() -> None:
    """Пропуск второй ступени при переданных полосах — TypeError, а не тихий дефект.

    До правки такой вызов молча возвращал False: московская сделка по 30 000 ₽/м²
    (внутри городской полосы 22475..772165) судилась ЕКБ-полом DEAL_MIN_PPM2=50000.
    """
    bands = {(77, "Москва"): (22_475, 772_165)}
    okrug = "муниципальный округ Некрасовка"
    with pytest.raises(TypeError, match="city_fallback"):
        est._is_plausible_deal(30_000, 5, 12, city=okrug, bands=bands, region_code=77)
    # Явный None разрешён — отказ от второй ступени виден в коде вызова.
    assert not est._is_plausible_deal(
        30_000, 5, 12, city=okrug, city_fallback=None, bands=bands, region_code=77
    )
    assert est._is_plausible_deal(
        30_000, 5, 12, city=okrug, city_fallback="Москва", bands=bands, region_code=77
    )


def test_plausible_deal_without_bands_keeps_positional_calls() -> None:
    """bands=None: судить нечем, обе ступени вырождаются в глобальные константы.

    Поэтому параметр обязателен УСЛОВНО: безусловный сломал бы 30 позиционных
    вызовов tests/test_deals_sanitize.py, не поймав ни одного реального дефекта.
    """
    assert est._is_plausible_deal(150_000, 5, 9)
    assert not est._is_plausible_deal(39_700, 5, 9)
    assert est._is_plausible_deal(est.DEAL_MIN_PPM2, 1, None)
    assert not est._is_plausible_deal(est.DEAL_MAX_PPM2 + 1, 5, 9)