gendesign/backend/tests/services/site_finder/test_best_layouts.py

"""Unit-тесты для get_best_layouts (Fix SF-01: honest time_window velocity).

Проверяет, что разные time_window → разные deals_window → разный velocity_per_month.

Mock-стратегия: патчим db.execute с side_effect, повторяя порядок вызовов
в get_best_layouts:
  1. _PARCEL_CENTROID_SQL      → .mappings().first()
  2. _COMPETITORS_IN_RADIUS_SQL → .mappings().all()
  3. _INLINE_VELOCITY_SQL       → .mappings().all()
  4. _SUPPLY_BATCH_SQL          → .mappings().all()

Fix #1956: supply берёт per-object последний snapshot внутри _SUPPLY_BATCH_SQL
(flats_latest CTE), отдельного db.scalar(MAX(snapshot_date)) больше нет.

Ключевые asserts:
- last_month (1 мес) → velocity = deals_window / 1.0
- last_quarter (3 мес) → velocity = deals_window / 3.0
- last_year (12 мес) → velocity = deals_window / 12.0
- Разный deals_window при разных time_window → разный mix.
"""

from __future__ import annotations

import datetime as dt
from unittest.mock import MagicMock

import pytest

from app.schemas.parcel import BestLayoutsRequest
from app.services.site_finder.best_layouts import (
    _TIME_WINDOW_PARAMS,
    MAX_BUCKET_SHARE_PCT,
    NAMELESS_CLUSTER_M,
    _cap_and_redistribute,
    get_best_layouts,
    group_radius_objects,
    normalize_complex_name,
)

_TODAY = dt.date.today()
CAD_NUM = "66:41:0303161:123"


# ── Фабрики mock-строк ────────────────────────────────────────────────────────


def _coord_row(lon: float = 60.6, lat: float = 56.85) -> MagicMock:
    r = MagicMock()
    r.__getitem__ = lambda self, k: {"center_lon": lon, "center_lat": lat}[k]
    return r


def _obj_id_row(
    obj_id: int,
    comm_name: str | None = None,
    dev_name: str | None = None,
    latitude: float | None = None,
    longitude: float | None = None,
) -> dict[str, object]:
    """Строка из _COMPETITORS_IN_RADIUS_SQL.

    Issue #2177: SQL теперь возвращает comm_name/dev_name/гео для группировки в
    комплексы; get_best_layouts вызывает dict(r), поэтому возвращаем реальный dict
    (SQLAlchemy RowMapping тоже dict-конвертируем). По умолчанию каждый obj_id
    получает уникальное имя (`obj-<id>`) → каждый = своя группа, как раньше по
    obj_id (существующие coverage-инварианты остаются в силе).
    """
    return {
        "obj_id": obj_id,
        "comm_name": comm_name if comm_name is not None else f"obj-{obj_id}",
        "dev_name": dev_name if dev_name is not None else f"dev-{obj_id}",
        "latitude": latitude,
        "longitude": longitude,
    }


def _vel_row(
    room_bucket: str = "2",
    deals_window: float = 48.0,
    avg_area: float = 55.0,
    avg_price_rub: float | None = 120000.0,
    obj_ids: list[int] | None = None,
    window_start: dt.date | None = None,
    window_end: dt.date | None = None,
) -> MagicMock:
    """Строка из _INLINE_VELOCITY_SQL.

    deals_window — реальные сделки за честное окно (не 24 мес).
    """
    oids = obj_ids if obj_ids is not None else [1]
    ws = window_start or _TODAY - dt.timedelta(days=90)
    we = window_end or _TODAY

    r = MagicMock()
    r.__getitem__ = lambda self, k: {
        "room_bucket": room_bucket,
        "deals_window": deals_window,
        "avg_area_m2": avg_area,
        "avg_price_per_m2_rub": avg_price_rub,
        "competitor_obj_ids": oids,
        "competitor_count": len(oids),
        "window_start": ws,
        "window_end": we,
    }[k]
    return r


def _supply_row(rb: str, ab: str, units: int) -> MagicMock:
    r = MagicMock()
    r.__getitem__ = lambda self, k: {"rb": rb, "ab": ab, "units": units}[k]
    return r


def _make_db(
    coord: MagicMock | None = None,
    id_rows: list[MagicMock] | None = None,
    vel_rows: list[MagicMock] | None = None,
    supply_rows: list[MagicMock] | None = None,
) -> MagicMock:
    """Сконструировать mock Session.

    Порядок db.execute() (Fix #1956: db.scalar(MAX(snapshot_date)) удалён —
    per-object последний снимок считается внутри _SUPPLY_BATCH_SQL):
      1. centroid     → .mappings().first()
      2. competitors  → .mappings().all()
      3. velocity     → .mappings().all()
      4. supply       → .mappings().all()
    """
    db = MagicMock()

    r0 = MagicMock()
    r0.mappings.return_value.first.return_value = coord

    r1 = MagicMock()
    r1.mappings.return_value.all.return_value = id_rows or []

    r2 = MagicMock()
    r2.mappings.return_value.all.return_value = vel_rows or []

    r3 = MagicMock()
    r3.mappings.return_value.all.return_value = supply_rows or []

    db.execute.side_effect = [r0, r1, r2, r3]
    return db


def _request(**kwargs) -> BestLayoutsRequest:
    defaults: dict = {
        "radius_km": 1.0,
        "time_window": "last_quarter",
        "min_velocity_per_month": 0.0,
    }
    defaults.update(kwargs)
    return BestLayoutsRequest(**defaults)


# ── Тесты TIME_WINDOW_PARAMS ──────────────────────────────────────────────────


def test_time_window_params_keys() -> None:
    """Все три time_window определены, months_in_window > 0."""
    for key in ("last_month", "last_quarter", "last_year"):
        assert key in _TIME_WINDOW_PARAMS
        interval_str, months = _TIME_WINDOW_PARAMS[key]
        assert isinstance(interval_str, str) and len(interval_str) > 0
        assert months > 0


# ── Тест SF-01: разный deals_window → разный velocity ────────────────────────


def test_last_month_velocity_divisor_1() -> None:
    """time_window=last_month: velocity = deals_window / 1.0."""
    deals = 30.0
    db = _make_db(
        coord=_coord_row(),
        id_rows=[_obj_id_row(1)],
        vel_rows=[_vel_row("1", deals_window=deals, obj_ids=[1])],
    )
    req = _request(time_window="last_month")
    resp = get_best_layouts(db, CAD_NUM, req)

    assert len(resp.top_layouts) == 1
    assert resp.top_layouts[0].velocity_per_month == pytest.approx(30.0, rel=1e-3)


def test_last_quarter_velocity_divisor_3() -> None:
    """time_window=last_quarter: velocity = deals_window / 3.0."""
    deals = 30.0
    db = _make_db(
        coord=_coord_row(),
        id_rows=[_obj_id_row(1)],
        vel_rows=[_vel_row("1", deals_window=deals, obj_ids=[1])],
    )
    req = _request(time_window="last_quarter")
    resp = get_best_layouts(db, CAD_NUM, req)

    assert len(resp.top_layouts) == 1
    assert resp.top_layouts[0].velocity_per_month == pytest.approx(10.0, rel=1e-3)


def test_last_year_velocity_divisor_12() -> None:
    """time_window=last_year: velocity = deals_window / 12.0."""
    deals = 60.0
    db = _make_db(
        coord=_coord_row(),
        id_rows=[_obj_id_row(1)],
        vel_rows=[_vel_row("1", deals_window=deals, obj_ids=[1])],
    )
    req = _request(time_window="last_year")
    resp = get_best_layouts(db, CAD_NUM, req)

    assert len(resp.top_layouts) == 1
    assert resp.top_layouts[0].velocity_per_month == pytest.approx(5.0, rel=1e-3)


def test_different_time_windows_produce_different_velocity() -> None:
    """Одни и те же deals_window → разная velocity_per_month для разных time_window.

    Главный acceptance-тест SF-01: time_window влияет на velocity, не только на масштаб.
    При одном и том же deals_window=30:
      last_month  → 30.0
      last_quarter → 10.0
      last_year   → 2.5
    """
    deals = 30.0

    velocities: dict[str, float] = {}
    for tw in ("last_month", "last_quarter", "last_year"):
        db = _make_db(
            coord=_coord_row(),
            id_rows=[_obj_id_row(1)],
            vel_rows=[_vel_row("2", deals_window=deals, obj_ids=[1])],
        )
        req = _request(time_window=tw)
        resp = get_best_layouts(db, CAD_NUM, req)
        assert len(resp.top_layouts) == 1, f"No layouts for {tw}"
        velocities[tw] = resp.top_layouts[0].velocity_per_month

    # Все три значения различаются
    vals = list(velocities.values())
    assert vals[0] != vals[1] != vals[2], f"Velocities must differ: {velocities}"
    # last_month > last_quarter > last_year (одинаковые deals, разный знаменатель)
    assert velocities["last_month"] > velocities["last_quarter"] > velocities["last_year"]


# ── Тест: ranking по velocity и sum pct = 100 ────────────────────────────────


def test_ranking_and_pct_sum_100() -> None:
    """3 room_buckets → ranking по velocity, sum pct = 100."""
    id_rows = [_obj_id_row(1), _obj_id_row(2), _obj_id_row(3)]
    vel_rows = [
        _vel_row("studio", deals_window=9.0, avg_area=26.0, obj_ids=[1]),  # 9/3=3.0
        _vel_row("1", deals_window=24.0, avg_area=40.0, obj_ids=[2]),  # 24/3=8.0
        _vel_row("2", deals_window=48.0, avg_area=55.0, obj_ids=[3]),  # 48/3=16.0
    ]
    supply_rows = [
        _supply_row("studio", "<25", 20),
        _supply_row("1", "40-60", 60),
        _supply_row("2", "40-60", 80),
    ]
    db = _make_db(coord=_coord_row(), id_rows=id_rows, vel_rows=vel_rows, supply_rows=supply_rows)
    req = _request(time_window="last_quarter")
    resp = get_best_layouts(db, CAD_NUM, req)

    top = resp.top_layouts
    assert len(top) == 3
    # rank 1 = "2" (наибольший velocity 16.0)
    assert top[0].room_bucket == "2"
    assert top[0].rank == 1
    assert top[0].velocity_per_month == pytest.approx(16.0, rel=1e-3)
    # rank 2 = "1" (8.0)
    assert top[1].room_bucket == "1"
    assert top[1].velocity_per_month == pytest.approx(8.0, rel=1e-3)
    # ранги уникальны
    assert sorted(t.rank for t in top) == [1, 2, 3]
    # sum pct = 100
    mix = resp.recommendation_for_tz.mix
    assert sum(m.pct for m in mix) == 100


# ── Тест: пустые конкуренты ───────────────────────────────────────────────────


def test_no_competitors_returns_empty_response() -> None:
    """Нет конкурентов в радиусе → пустые top_layouts + confidence=low."""
    db = _make_db(coord=_coord_row(), id_rows=[], vel_rows=[])
    req = _request()
    resp = get_best_layouts(db, CAD_NUM, req)

    assert resp.top_layouts == []
    assert resp.data_quality.confidence == "low"
    assert resp.recommendation_for_tz.based_on_obj_count == 0


# ── Тест: centroid не найден ──────────────────────────────────────────────────


def test_centroid_not_found_raises_value_error() -> None:
    """Геометрия участка не найдена → ValueError."""
    db = _make_db(coord=None)
    req = _request()

    with pytest.raises(ValueError, match="не найдена"):
        get_best_layouts(db, "99:99:9999999:999", req)


# ── Тест: min_velocity фильтрует строки ──────────────────────────────────────


def test_min_velocity_filters_low_rows() -> None:
    """min_velocity_per_month=5 → строки с velocity<5 не попадают в top_layouts.

    last_quarter (3 мес):
      studio: 9 / 3 = 3.0  < 5.0 → отфильтрован
      1:      24 / 3 = 8.0  > 5.0 → остаётся
    """
    id_rows = [_obj_id_row(1), _obj_id_row(2)]
    vel_rows = [
        _vel_row("studio", deals_window=9.0, obj_ids=[1]),
        _vel_row("1", deals_window=24.0, obj_ids=[2]),
    ]
    db = _make_db(coord=_coord_row(), id_rows=id_rows, vel_rows=vel_rows)
    req = _request(time_window="last_quarter", min_velocity_per_month=5.0)
    resp = get_best_layouts(db, CAD_NUM, req)

    top = resp.top_layouts
    assert len(top) == 1
    assert top[0].room_bucket == "1"
    assert top[0].velocity_per_month == pytest.approx(8.0, rel=1e-3)


# ── Тест: exclude_competitor_obj_ids ─────────────────────────────────────────


def test_exclude_competitor_obj_ids() -> None:
    """exclude_competitor_obj_ids=[20] при единственном конкуренте → пустой ответ."""
    id_rows = [_obj_id_row(20)]
    db = _make_db(coord=_coord_row(), id_rows=id_rows, vel_rows=[])
    req = _request(exclude_competitor_obj_ids=[20])
    resp = get_best_layouts(db, CAD_NUM, req)

    assert resp.top_layouts == []
    assert resp.data_quality.objects_total_in_radius == 1


# ── Тест: total_sold_in_window совпадает с deals_window ──────────────────────


def test_total_sold_in_window_matches_deals_window() -> None:
    """total_sold_in_window в TopLayoutRow = deals_window (целое)."""
    deals = 37.0
    db = _make_db(
        coord=_coord_row(),
        id_rows=[_obj_id_row(5)],
        vel_rows=[_vel_row("3", deals_window=deals, obj_ids=[5])],
    )
    req = _request(time_window="last_quarter")
    resp = get_best_layouts(db, CAD_NUM, req)

    assert len(resp.top_layouts) == 1
    assert resp.top_layouts[0].total_sold_in_window == int(deals)


# ── Тесты Fix #1229: supply / velocity bucket vocabulary match ────────────────


def test_supply_velocity_buckets_match_for_2_rooms() -> None:
    """Fix #1229: supply bucket для 2-комн матчит velocity '2' независимо от площади.

    Раньше supply отдельно вычислял euro-1/euro-2 для rooms=2 + area<50 → эти
    строки выпадали из знаменателя bucket '2' → sold_pct/is_oversold двушек
    были завышены. После фикса вся supply rooms=2 идёт в '2' и sold_pct
    рассчитывается от полного supply.

    Здесь моделируем что весь supply для '2' уже агрегирован SQL-стороной в
    один (rb='2', ab='40-60') ряд: 20 deals_window против 50 supply → 40% sold.
    """
    deals = 20.0
    id_rows = [_obj_id_row(7)]
    vel_rows = [_vel_row("2", deals_window=deals, avg_area=45.0, obj_ids=[7])]
    supply_rows = [_supply_row("2", "40-60", 50)]
    db = _make_db(
        coord=_coord_row(),
        id_rows=id_rows,
        vel_rows=vel_rows,
        supply_rows=supply_rows,
    )
    req = _request(time_window="last_quarter")
    resp = get_best_layouts(db, CAD_NUM, req)

    assert len(resp.top_layouts) == 1
    row = resp.top_layouts[0]
    assert row.room_bucket == "2"
    assert row.supply_units_in_radius == 50
    # 20 / 50 * 100 = 40.0 (раньше при euro-* разделении знаменатель был меньше
    # → sold_pct > 40 или is_oversold=True).
    assert row.sold_pct_of_supply == pytest.approx(40.0, rel=1e-3)
    assert row.is_oversold is False


def test_supply_does_not_emit_euro_buckets() -> None:
    """Fix #1229: supply SQL больше НЕ содержит литералов 'euro-1' / 'euro-2'.

    Regression guard: если кто-то восстанавливает SF-08 euro-биннинг в supply
    без согласования с velocity-стороной (objective_corpus_room_month отдаёт
    {studio,1,2,3,4+}, не euro-*), sold_pct двушек снова поедет.
    """
    from app.services.site_finder.best_layouts import _SUPPLY_BATCH_SQL

    sql_text = str(_SUPPLY_BATCH_SQL.text)
    assert "'euro-1'" not in sql_text, "_SUPPLY_BATCH_SQL вернул euro-1 — see #1229"
    assert "'euro-2'" not in sql_text, "_SUPPLY_BATCH_SQL вернул euro-2 — see #1229"


def test_supply_dedups_objects_to_latest_snapshot() -> None:
    """Regression guard против object-snapshot fan-out (supply units ~8.5x inflation).

    domrf_kn_objects — snapshot-dimension (UNIQUE (obj_id, snapshot_date), ~8
    снимков/obj_id). Если supply снова заджойнит flats НАПРЯМУЮ к objects без
    дедупа до одного снимка на obj_id, каждый flat посчитается ~8.5x →
    supply_units_in_radius завышен, sold_pct_of_supply занижен, is_oversold
    недо-фаерит (все три — user-facing). Фикс: objects-first MATERIALIZED CTE
    с DISTINCT ON (obj_id) по последнему снимку, flats джойнятся к нему.
    """
    from app.services.site_finder.best_layouts import _SUPPLY_BATCH_SQL

    sql_text = str(_SUPPLY_BATCH_SQL.text)
    # objects дедуплицируются до одного (последнего) снимка на obj_id
    assert "DISTINCT ON (o.obj_id)" in sql_text, "supply не дедупит objects → fan-out вернётся"
    assert "snapshot_date DESC" in sql_text, "DISTINCT ON должен брать ПОСЛЕДНИЙ снимок"
    # objects-first CTE материализован — иначе планнер инлайнит → flats-first seq scan
    assert "MATERIALIZED" in sql_text, "nearby CTE должен быть MATERIALIZED (план objects-first)"
    # flats джойнятся к дедуплицированному nearby, НЕ напрямую к raw objects
    assert "JOIN domrf_kn_objects" not in sql_text, "прямой flats→objects join = fan-out по снимкам"


def test_supply_joins_flats_per_object_latest_snapshot() -> None:
    """Fix #1956 regression guard: supply джойнит flats по ПЕР-ОБЪЕКТНОМУ снимку.

    domrf_kn_flats — ПО-ОБЪЕКТНЫЙ time-series (каждый ЖК скрейпится в свой день).
    Регрессия #1944 джойнила flats по ОДНОЙ глобальной дате
    `f.snapshot_date = MAX(snapshot_date по всей таблице)` → на ней присутствует
    обычно 1 объект → supply_units_in_radius=0 для всех строк → frontend
    «Срок продажи 0 мес» / «% продано —». Фикс: flats_latest CTE с
    DISTINCT ON (obj_id) ... ORDER BY obj_id, snapshot_date DESC — у каждого
    объекта свой последний снимок. Этот guard ловит возврат глобал-max джойна.
    """
    from app.services.site_finder.best_layouts import _SUPPLY_BATCH_SQL

    sql_text = str(_SUPPLY_BATCH_SQL.text)
    # per-object последний снимок flats
    assert "flats_latest" in sql_text, "нет flats_latest CTE → supply снова на глобал-max снимке"
    assert (
        "DISTINCT ON (f.obj_id)" in sql_text
    ), "flats не дедупятся per-object → нет per-obj снимка"
    # НЕ должно быть джойна по единой внешней дате-параметру (регрессия #1944)
    assert ":latest_snap" not in sql_text, "глобальный :latest_snap вернулся → supply=0 регрессия"
    assert (
        "MAX(snapshot_date)" not in sql_text
    ), "глобальный MAX(snapshot_date) → supply=0 регрессия"


# ── Тесты _cap_and_redistribute (Fix SF-09 review) ───────────────────────────


@pytest.mark.parametrize(
    "pct_map, expect_pathological",
    [
        # 1. normal: одиночный bucket > 35, free достаточно capacity
        ({"1k": 50, "studio": 30, "2k": 20}, False),
        # 2. heavy skew (3-bucket): surplus=40, capacity=20+25=45 — помещается
        ({"1k": 75, "studio": 15, "2k": 10}, False),
        # 3. multiple buckets > 35
        ({"1k": 50, "studio": 40, "2k": 10}, False),
        # 4. all > 35 — pathological
        ({"1k": 50, "studio": 50}, True),
        # 5. граничный: один bucket ровно на cap — не clamp
        ({"1k": 35, "studio": 35, "2k": 30}, False),
        # 6. single bucket 100% — pathological (нет free)
        ({"1k": 100}, True),
        # 7. 2-bucket heavy: surplus=55, capacity=25 — pathological (не помещается)
        ({"1k": 90, "studio": 10}, True),
        # 8. все ≤ cap — fast-path без изменений
        ({"1k": 30, "studio": 35, "2k": 35}, False),
        # 9. 2-bucket: 70/30 → surplus=35, capacity=5 → pathological
        ({"1k": 70, "studio": 30}, True),
        # 10. 2-bucket: 99/1 → surplus=64, capacity=34 → pathological
        ({"1k": 99, "studio": 1}, True),
    ],
)
def test_cap_and_redistribute_invariants(
    pct_map: dict[str, int],
    expect_pathological: bool,
) -> None:
    """Invariant: max(pct) ≤ cap И sum(pct) == 100 (или cap_skipped=True в pathological).

    Pathological — `cap_skipped=True`, max МОЖЕТ быть > cap (геометрически surplus
    не вмещается в free capacity).
    """
    result, cap_skipped = _cap_and_redistribute(pct_map)

    assert (
        cap_skipped == expect_pathological
    ), f"cap_skipped={cap_skipped} но ожидали {expect_pathological} для {pct_map}"
    assert (
        sum(result.values()) == 100
    ), f"sum={sum(result.values())} != 100 для {pct_map} → {result}"
    if not expect_pathological:
        assert (
            max(result.values()) <= MAX_BUCKET_SHARE_PCT
        ), f"max={max(result.values())} > cap={MAX_BUCKET_SHARE_PCT} для {pct_map} → {result}"


@pytest.mark.parametrize(
    "deals, expect_pathological, label",
    [
        # 3-bucket с достаточной capacity — surplus помещается, cap соблюдён
        ({"1k": 75, "studio": 15, "2k": 10}, False, "{1k:75, studio:15, 2k:10}"),
        ({"1k": 80, "studio": 12, "2k": 8}, False, "{1k:80, studio:12, 2k:8}"),
        ({"1k": 60, "studio": 30, "2k": 10}, False, "{1k:60, studio:30, 2k:10}"),
        ({"a": 50, "b": 30, "c": 20}, False, "{50, 30, 20}"),
        # 2-bucket — surplus геометрически не помещается, cap_skipped=True
        ({"1k": 90, "studio": 10}, True, "{1k:90, studio:10}"),
        ({"1k": 70, "studio": 30}, True, "{1k:70, studio:30}"),
        ({"1k": 99, "studio": 1}, True, "{1k:99, studio:1}"),
    ],
)
def test_cap_reproduced_failing_cases(
    deals: dict[str, int], expect_pathological: bool, label: str
) -> None:
    """Review round-2 reproduced cases: 2-bucket — pathological, 3-bucket — fit cap."""
    result, cap_skipped = _cap_and_redistribute(deals)
    assert (
        cap_skipped == expect_pathological
    ), f"cap_skipped={cap_skipped} ожидали {expect_pathological} для {label}"
    assert sum(result.values()) == 100, f"sum != 100 для {label} → {result}"
    if not expect_pathological:
        assert (
            max(result.values()) <= MAX_BUCKET_SHARE_PCT
        ), f"max={max(result.values())} > {MAX_BUCKET_SHARE_PCT} для {label} → {result}"


def test_cap_iteration_count_bounded() -> None:
    """Round 2 regression: алгоритм завершается за ≤ len(pct_map)+1 итераций.

    Round 1 bag: на 2-bucket {1k:70, studio:30} цикл осциллировал бесконечно.
    Round 2 fix: capacity-aware redistribute + hard `for _ in range(N+1)` guard.
    Этот тест гарантирует что вызов не зависает (pytest-timeout не нужен).
    """
    import time

    pathological_cases = [
        {"1k": 70, "studio": 30},
        {"1k": 99, "studio": 1},
        {"1k": 90, "studio": 10},
        {"1k": 50, "studio": 50},
    ]
    for case in pathological_cases:
        start = time.perf_counter()
        result, cap_skipped = _cap_and_redistribute(case)
        elapsed_ms = (time.perf_counter() - start) * 1000
        assert elapsed_ms < 100, f"Завис ({elapsed_ms:.0f}ms) на {case}"
        assert sum(result.values()) == 100, f"sum != 100 для {case}"
        # 2-bucket с одним > cap всегда pathological (surplus > free capacity)
        if case != {"1k": 50, "studio": 50}:
            assert cap_skipped, f"Ожидали cap_skipped=True для {case}"


def test_cap_and_redistribute_no_dominant_unchanged() -> None:
    """Если все bucket'ы ≤ cap — результат идентичен входу (fast-path)."""
    pct_map = {"studio": 20, "1": 35, "2": 30, "3": 15}
    result, cap_skipped = _cap_and_redistribute(pct_map)
    assert not cap_skipped
    assert result == pct_map


def test_cap_and_redistribute_empty() -> None:
    """Пустой dict → возвращается как есть."""
    result, cap_skipped = _cap_and_redistribute({})
    assert result == {}
    assert not cap_skipped


def test_cap_skipped_flag_propagates_to_recommendation() -> None:
    """Pathological case → cap_skipped=True в recommendation_for_tz ответа."""
    # 2 bucket'а по 50% — pathological
    id_rows = [_obj_id_row(1), _obj_id_row(2)]
    vel_rows = [
        _vel_row("studio", deals_window=50.0, obj_ids=[1]),
        _vel_row("1", deals_window=50.0, obj_ids=[2]),
    ]
    db = _make_db(coord=_coord_row(), id_rows=id_rows, vel_rows=vel_rows)
    req = _request(time_window="last_quarter")
    resp = get_best_layouts(db, CAD_NUM, req)

    # С deals 50/50 → normalize_pct даёт {studio:50, 1:50} — оба выше cap
    assert resp.recommendation_for_tz.cap_skipped is True


def test_cap_skipped_false_for_normal_case() -> None:
    """Normal case с capping → cap_skipped=False в recommendation_for_tz."""
    id_rows = [_obj_id_row(1), _obj_id_row(2), _obj_id_row(3)]
    vel_rows = [
        _vel_row("1k", deals_window=75.0, obj_ids=[1]),
        _vel_row("studio", deals_window=15.0, obj_ids=[2]),
        _vel_row("2k", deals_window=10.0, obj_ids=[3]),
    ]
    db = _make_db(coord=_coord_row(), id_rows=id_rows, vel_rows=vel_rows)
    req = _request(time_window="last_quarter")
    resp = get_best_layouts(db, CAD_NUM, req)

    assert resp.recommendation_for_tz.cap_skipped is False
    mix = resp.recommendation_for_tz.mix
    assert all(row.pct <= MAX_BUCKET_SHARE_PCT for row in mix)
    assert sum(row.pct for row in mix) == 100


# ── Issue #2177: группировка obj_id в комплексы ──────────────────────────────


# Реальный прод-кейс: радиус 1 км от 66:41:0204016:10 = 12 obj_id, 5 комплексов.
# Координаты сконструированы под дистанционные ограничения из issue:
#   • Симфония — 3 точки в ~200 м (name-группа, дистанция не важна);
#   • Эфес безымянные — 3 точки в ~6 м друг от друга (< 300 м → один кластер),
#     ~600 м от именованного «7 ключей» (> 300 м → отдельная группа).
_SYM1 = (56.850449, 60.600821)
_SYM2 = (56.850449, 60.604107)
_SYM3 = (56.852246, 60.600821)
_KLYUCHI = (56.85, 60.593429)
_EFES1 = (56.85, 60.583572)
_EFES2 = (56.85, 60.583671)
_EFES3 = (56.850045, 60.583621)


def _prod_radius_rows() -> list[dict[str, object]]:
    """12 obj_id прод-кейса (радиус 1 км от 66:41:0204016:10)."""
    return [
        # Симфония (Желдорипотека) — 3 obj_id, одинаковое comm_name+dev_name
        _obj_id_row(3346, "ЖК «Симфония»", "Желдорипотека", *_SYM1),
        _obj_id_row(43063, "ЖК «Симфония»", "Желдорипотека", *_SYM2),
        _obj_id_row(43064, "ЖК «Симфония»", "Желдорипотека", *_SYM3),
        # Квартет (Эталон) — 2 obj_id, повтор «Квартет» ×3 в строке
        _obj_id_row(33955, "Квартет Квартет Квартет", "Эталон", 56.851, 60.602),
        _obj_id_row(21801, "Квартет Квартет Квартет", "Эталон", 56.8512, 60.6022),
        # Квартал Депо (Брусника) — 2 obj_id, «квартал» без кавычек = часть имени
        _obj_id_row(63122, "Квартал Депо", "Брусника", 56.8505, 60.5985),
        _obj_id_row(63282, "Квартал Депо", "Брусника", 56.8506, 60.5986),
        # Траектория (Брусника) — 1 obj_id
        _obj_id_row(68027, "ЖК «Траектория»", "Брусника", 56.8495, 60.6015),
        # 7 ключей (Эфес) — именованная 2 очередь
        _obj_id_row(15731, "2 очередь жилого комплекса «7 ключей»", "Эфес", *_KLYUCHI),
        # 3 безымянных Эфеса — гео-кучка ~6 м, ~600 м от «7 ключей»
        _obj_id_row(55320, "", "Эфес", *_EFES1),
        _obj_id_row(55321, "", "Эфес", *_EFES2),
        _obj_id_row(55322, "", "Эфес", *_EFES3),
    ]


def test_group_radius_objects_prod_case_groups() -> None:
    """Прод-кейс: 12 obj_id → 5 именованных комплексов + 1 безымянный кластер (#2177).

    Issue-заголовок «5 реальных комплексов» перечисляет ИМЕНОВАННЫЕ проекты
    (Симфония, Квартет, Квартал Депо, Траектория, 7 ключей). 3 безымянных Эфеса
    в ~600 м от «7 ключей» issue ЯВНО требует держать ОТДЕЛЬНОЙ группой («честно
    неизвестный проект», НЕ приклеивать дальше 300 м). Итог — 6 групп; ключевой
    инвариант: «7 ключей» и безымянные Эфесы НЕ слиты.
    """
    groups = group_radius_objects(_prod_radius_rows())

    # состав групп: множество frozenset(obj_ids) должно совпасть
    got = {frozenset(g.obj_ids) for g in groups}
    expected = {
        frozenset({3346, 43063, 43064}),  # Симфония (snapshot-дубли схлопнуты)
        frozenset({33955, 21801}),  # Квартет (повтор имени схлопнут)
        frozenset({63122, 63282}),  # Квартал Депо («квартал» сохранён в ядре)
        frozenset({68027}),  # Траектория
        frozenset({15731}),  # 7 ключей (именованная 2 очередь)
        frozenset({55320, 55321, 55322}),  # безымянные Эфесы — свой кластер
    }
    assert len(groups) == 6, f"ожидалось 6 групп, получено {len(groups)}: {got}"
    # load-bearing инвариант: 7 ключей и безымянные Эфесы разделены >300 м
    assert frozenset({15731}) in got, "7 ключей должна быть отдельной группой"
    assert (
        frozenset({55320, 55321, 55322}) in got
    ), "3 безымянных Эфеса — один кластер, отдельный от 7 ключей"
    assert got == expected


def test_group_radius_objects_symphony_collapses_snapshot_dupes() -> None:
    """Симфония: 3 obj_id с одинаковым comm_name+dev_name → одна группа."""
    rows = [
        _obj_id_row(3346, "ЖК «Симфония»", "Желдорипотека", *_SYM1),
        _obj_id_row(43063, "ЖК «Симфония»", "Желдорипотека", *_SYM2),
        _obj_id_row(43064, "ЖК «Симфония»", "Желдорипотека", *_SYM3),
    ]
    groups = group_radius_objects(rows)
    assert len(groups) == 1
    assert sorted(groups[0].obj_ids) == [3346, 43063, 43064]


def test_group_nameless_efes_two_clusters_do_not_merge() -> None:
    """Два разнесённых (>300 м) безымянных кластера одного застройщика НЕ сливаются."""
    # кластер A ~6 м; кластер B в ~600 м от A; оба Эфес, comm_name пустой
    rows = [
        _obj_id_row(55320, "", "Эфес", *_EFES1),
        _obj_id_row(55321, "", "Эфес", *_EFES2),
        _obj_id_row(55322, "", "Эфес", *_EFES3),
        # второй кластер — рядом с «7 ключей» (600 м от первого)
        _obj_id_row(90001, "", "Эфес", 56.85, 60.593429),
        _obj_id_row(90002, "", "Эфес", 56.850009, 60.5934),
    ]
    groups = group_radius_objects(rows)
    assert len(groups) == 2, f"ожидалось 2 кластера, получено {len(groups)}"
    clusters = {frozenset(g.obj_ids) for g in groups}
    assert frozenset({55320, 55321, 55322}) in clusters
    assert frozenset({90001, 90002}) in clusters


def test_group_nameless_different_dev_never_merge() -> None:
    """Безымянные точки разных застройщиков в 1 м — всё равно разные группы."""
    rows = [
        _obj_id_row(1, "", "Эфес", 56.85, 60.60),
        _obj_id_row(2, "", "Брусника", 56.850001, 60.60),
    ]
    groups = group_radius_objects(rows)
    assert len(groups) == 2


def test_group_nameless_without_coords_each_own_group() -> None:
    """Безымянный obj_id без координат — своя группа (не с чем кластеризовать)."""
    rows = [
        _obj_id_row(1, "", "Эфес", None, None),
        _obj_id_row(2, "", "Эфес", None, None),
    ]
    groups = group_radius_objects(rows)
    assert len(groups) == 2


# ── Нормализация имён ─────────────────────────────────────────────────────────


@pytest.mark.parametrize(
    "raw, expected",
    [
        # ёлочки: ядро из «N очередь жилого комплекса «X»»
        ("2 очередь жилого комплекса «7 ключей»", "7 ключей"),
        ("1-я очередь ЖК «Символ»", "символ"),
        # повторы ядра
        ("Квартет Квартет Квартет", "квартет"),
        ("Квартет", "квартет"),
        # generic-префиксы
        ("ЖК «Траектория»", "траектория"),
        ("ЖК Траектория", "траектория"),
        ("Жилой комплекс Символ", "символ"),
        ("микрорайон Академический", "академический"),
        # «квартал» без кавычек — часть имени, НЕ префикс
        ("Квартал Депо", "квартал депо"),
        ("квартал депо", "квартал депо"),
        # пустое / None
        ("", ""),
        (None, ""),
        ("   ", ""),
        # регистр + пробелы
        ("  Симфония  ", "симфония"),
        ("ЖК   «Символ»", "символ"),
    ],
)
def test_normalize_complex_name(raw: str | None, expected: str) -> None:
    assert normalize_complex_name(raw) == expected


def test_normalize_kvartal_depo_stays_intact() -> None:
    """«Квартал Депо» не должен потерять слово «квартал» (regression guard)."""
    assert normalize_complex_name("Квартал Депо") == "квартал депо"
    # два obj_id «Квартал Депо» → одна группа (ядро совпадает)
    rows = [
        _obj_id_row(63122, "Квартал Депо", "Брусника", 56.8505, 60.5985),
        _obj_id_row(63282, "Квартал Депо", "Брусника", 56.8506, 60.5986),
    ]
    groups = group_radius_objects(rows)
    assert len(groups) == 1
    assert sorted(groups[0].obj_ids) == [63122, 63282]


# ── Coverage по группам ───────────────────────────────────────────────────────


def test_coverage_counts_groups_not_raw_obj_ids() -> None:
    """Прод-кейс coverage: velocity у 2 групп → «2 из 6 комплексов», raw=12.

    Из 6 групп velocity-данные (obj_ids_with_data) есть у Симфонии (obj 3346) и
    Квартета (obj 33955). Группа считается покрытой, если ХОТЯ БЫ ОДИН её obj_id
    в velocity-выборке. Знаменатель = группы (6), НЕ сырые obj_id (12).
    """
    id_rows = _prod_radius_rows()
    # velocity только у Симфонии (через obj 3346) и Квартета (через obj 33955)
    vel_rows = [
        _vel_row("2", deals_window=48.0, obj_ids=[3346]),
        _vel_row("1", deals_window=24.0, obj_ids=[33955]),
    ]
    db = _make_db(coord=_coord_row(), id_rows=id_rows, vel_rows=vel_rows)
    req = _request(time_window="last_quarter")
    resp = get_best_layouts(db, CAD_NUM, req)

    dq = resp.data_quality
    assert dq.raw_objects_total == 12, "сырой счёт obj_id должен быть 12"
    assert dq.objects_total_in_radius == 6, "знаменатель = комплексы (6), не obj_id"
    assert dq.objects_with_velocity_data == 2, "покрыты Симфония + Квартет"
    # 2/6 = 33.3% (не 2/12 = 16.7%)
    assert dq.velocity_coverage_pct == pytest.approx(33.3, rel=1e-2)


def test_coverage_group_covered_by_any_member_obj_id() -> None:
    """Группа «с данными», если velocity есть хотя бы у одного её obj_id.

    Симфония = {3346, 43063, 43064}; velocity только у 43064 (не у 3346) →
    группа всё равно покрыта.
    """
    id_rows = [
        _obj_id_row(3346, "ЖК «Симфония»", "Желдорипотека", *_SYM1),
        _obj_id_row(43063, "ЖК «Симфония»", "Желдорипотека", *_SYM2),
        _obj_id_row(43064, "ЖК «Симфония»", "Желдорипотека", *_SYM3),
    ]
    vel_rows = [_vel_row("2", deals_window=30.0, obj_ids=[43064])]
    db = _make_db(coord=_coord_row(), id_rows=id_rows, vel_rows=vel_rows)
    req = _request(time_window="last_quarter")
    resp = get_best_layouts(db, CAD_NUM, req)

    dq = resp.data_quality
    assert dq.objects_total_in_radius == 1
    assert dq.objects_with_velocity_data == 1
    assert dq.raw_objects_total == 3
    assert dq.velocity_coverage_pct == pytest.approx(100.0, rel=1e-3)


def test_nameless_cluster_constant_reasonable() -> None:
    """NAMELESS_CLUSTER_M экспортируется и в разумных пределах (guard)."""
    assert 100.0 <= NAMELESS_CLUSTER_M <= 500.0