gendesign/backend/tests/api/v1/test_analyze_competitors_status.py

"""Тесты: /analyze endpoint возвращает site_status + ready_dt в competitors[].

Mock-based — не требуют живой БД.
Проверяет:
  - поля site_status и ready_dt присутствуют в каждом элементе competitors
  - первые позиции занимают строящиеся ЖК (site_status='Строящиеся')
  - сданные ЖК идут после строящихся
  - obj_pricing CTE ограничен ближними obj_id (pushdown perf-guard, #70/#307)
"""

from __future__ import annotations

import datetime
import inspect
import re
from unittest.mock import MagicMock

from app.api.v1 import parcels as parcels_module

# ── Вспомогательные фабрики ───────────────────────────────────────────────────


def _competitor_mapping(
    obj_id: int,
    comm_name: str,
    site_status: str,
    ready_dt: datetime.date | None,
    flat_count: int,
    distance_m: float = 500.0,
) -> MagicMock:
    """Имитирует sqlalchemy RowMapping для строки конкурента."""
    data: dict = {
        "obj_id": obj_id,
        "comm_name": comm_name,
        "dev_name": "TestDev",
        "obj_class": "комфорт",
        "flat_count": flat_count,
        "district_name": "Ленинский",
        "site_status": site_status,
        "ready_dt": ready_dt,
        "distance_m": distance_m,
    }
    m = MagicMock()
    m.__getitem__ = lambda self, k: data[k]
    m.keys = lambda: data.keys()
    m.__iter__ = lambda self: iter(data)
    m.items = lambda: data.items()
    return m


# ── Тестовые данные ───────────────────────────────────────────────────────────

# Два сданных ЖК с большим flat_count и один строящийся с маленьким.
# До фикса ORDER BY flat_count DESC → сданные шли первыми.
_ROWS_MIXED = [
    _competitor_mapping(
        1, "ПИК Космонавтов 11 корп.1", "Сданные", datetime.date(2022, 6, 1), 800, 300.0
    ),
    _competitor_mapping(
        2, "ПИК Космонавтов 11 корп.2", "Сданные", datetime.date(2023, 3, 1), 750, 310.0
    ),
    _competitor_mapping(
        3, "Новый ЖК Строящийся", "Строящиеся", datetime.date(2026, 9, 1), 200, 400.0
    ),
]


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


class TestCompetitorsHaveStatusFields:
    """site_status и ready_dt должны присутствовать в competitors[]."""

    def test_fields_present(self) -> None:
        """Каждый конкурент содержит site_status и ready_dt."""
        competitors = [dict(r.items()) for r in _ROWS_MIXED]
        for c in competitors:
            assert "site_status" in c, f"site_status отсутствует в {c}"
            assert "ready_dt" in c, f"ready_dt отсутствует в {c}"

    def test_site_status_values(self) -> None:
        """site_status принимает ожидаемые значения."""
        competitors = [dict(r.items()) for r in _ROWS_MIXED]
        statuses = {c["site_status"] for c in competitors}
        assert "Строящиеся" in statuses
        assert "Сданные" in statuses

    def test_ready_dt_is_date_or_none(self) -> None:
        """ready_dt — datetime.date или None."""
        competitors = [dict(r.items()) for r in _ROWS_MIXED]
        for c in competitors:
            val = c["ready_dt"]
            assert val is None or isinstance(
                val, datetime.date
            ), f"ready_dt имеет неожиданный тип {type(val)}: {val}"


class TestCompetitorsSortOrder:
    """Строящиеся ЖК должны идти первыми независимо от flat_count."""

    def test_stroyashchiesya_first(self) -> None:
        """Строящийся ЖК с flat_count=200 должен быть раньше сданных с flat_count=800."""

        # Симулируем SQL ORDER BY:
        # CASE site_status WHEN 'Строящиеся' THEN 0 ELSE 1 END, distance_m ASC
        def _sort_key(r: MagicMock) -> tuple:
            data = dict(r.items())
            status_order = 0 if data["site_status"] == "Строящиеся" else 1
            return (status_order, data["distance_m"])

        sorted_rows = sorted(_ROWS_MIXED, key=_sort_key)
        first = dict(sorted_rows[0].items())
        assert first["site_status"] == "Строящиеся", (
            f"Первый конкурент должен быть 'Строящиеся', " f"но получили '{first['site_status']}'"
        )

    def test_flat_count_desc_would_break_order(self) -> None:
        """Демонстрирует, что старый ORDER BY flat_count DESC ставил сданные первыми."""
        sorted_by_flat = sorted(
            _ROWS_MIXED,
            key=lambda r: dict(r.items())["flat_count"],
            reverse=True,
        )
        first_old = dict(sorted_by_flat[0].items())
        # Старая логика: первым шёл ЖК с flat_count=800 (Сданные)
        assert first_old["flat_count"] == 800
        assert first_old["site_status"] == "Сданные"


class TestObjPricingPushdown:
    """obj_pricing CTE считается ТОЛЬКО для ближних obj_id (perf-guard, #70/#307).

    Старая версия (#1332 OBJ-3) агрегировала objective_lots (2.1 ГБ) по ВСЕМ
    ~286 маппингам и выбрасывала далёкие медианы в LEFT JOIN. Pushdown ограничивает
    scope множеством ближних конкурентов → seq scan заменяется index scan
    (на проде ~700мс → ~290мс, buffer reads ×10).

    Корректность: AVG/COUNT на domrf_obj_id независимы между группами, поэтому
    scoping байт-идентичен для тех obj_id, что реально джойнятся (подтверждено
    на проде: 43 ближних группы, 0 расхождений со старой агрегацией).

    Тест защищает от тихого реверта на полнотабличную агрегацию.
    """

    @staticmethod
    def _competitor_sql() -> str:
        """Нормализованный (схлопнутые пробелы) текст competitor-SQL из analyze_parcel."""
        src = inspect.getsource(parcels_module.analyze_parcel)
        marker = "WITH latest_obj AS ("
        # В analyze_parcel два CTE-блока с этим маркером (competitors + pipeline);
        # competitor-блок — первый и единственный с obj_pricing.
        start = src.index(marker)
        block = src[start : src.index('"""', start)]
        return re.sub(r"\s+", " ", block)

    def test_obj_pricing_scoped_to_nearby(self) -> None:
        """obj_pricing фильтрует domrf_obj_id по ближним (pushdown), не агрегирует всё."""
        sql = self._competitor_sql()
        assert "obj_pricing AS (" in sql, "obj_pricing CTE исчез — проверь analyze_parcel"
        # Pushdown-предикат: scope агрегации на ближний набор.
        assert "cm.domrf_obj_id IN (SELECT obj_id FROM nearby_obj)" in sql, (
            "obj_pricing должен ограничивать domrf_obj_id ближними obj_id (pushdown), "
            "иначе objective_lots (2.1ГБ) сканируется целиком — perf-регрессия #70/#307"
        )

    def test_final_select_reuses_nearby_cte(self) -> None:
        """Финальный SELECT и obj_pricing берут один и тот же nearby_obj — scope == consumed."""
        sql = self._competitor_sql()
        assert "nearby_obj AS (" in sql, "nearby_obj CTE отсутствует"
        # Финальная выборка идёт из nearby_obj (а не из полного latest_obj),
        # гарантируя что scope obj_pricing совпадает с потребляемым множеством.
        assert "FROM nearby_obj o" in sql, (
            "финальный SELECT должен читать из nearby_obj, чтобы scope цен "
            "совпадал с выводимыми конкурентами"
        )

    def test_aggregation_per_domrf_obj_id(self) -> None:
        """Агрегаты (AVG/COUNT FILTER) считаются per domrf_obj_id (логика цен #1332).

        #1964: источник агрегатов сменился с сырого objective_lots (alias ol) на
        physflat-дедуп CTE obj_lots_latest (alias oll) — см. test_obj_pricing_*_physflat
        ниже. Сами агрегатные выражения и группировка per-obj_id неизменны.
        """
        sql = self._competitor_sql()
        assert "ROUND(AVG(oll.price_per_m2_rub)::numeric, 0) AS avg_price_per_m2_rub" in sql
        assert "lots_with_price" in sql
        assert "COUNT(*) FILTER (WHERE oll.is_sold) AS units_sold" in sql
        assert "COUNT(*) FILTER (WHERE NOT oll.is_sold) AS units_available" in sql
        assert "GROUP BY np.domrf_obj_id" in sql

    def test_obj_pricing_dedups_physflat_inline(self) -> None:
        """#1964: obj_pricing агрегирует physflat-дедуп набор (DISTINCT ON), НЕ сырой.

        objective_lots раздут ~2.91× (мульти objective_lot_id на физлот через
        пере-листинги) → units_sold/units_available карточки конкурента были завышены.
        Дедупим INLINE (DISTINCT ON physflat-ключ, последний снапшот) со scope по
        project_name ближних ЖК. НЕ через v_objective_lots_latest: view материализует
        ВСЮ таблицу 1.76M (join не проходит ниже DISTINCT ON → seq-scan+sort ~5.7 s на
        request-path analyze_parcel; inline scoped → ~0.4 s, прод-EXPLAIN #1964).
        """
        sql = self._competitor_sql()
        assert "obj_lots_latest AS (" in sql, "должен быть physflat-дедуп CTE (#1964)"
        assert "DISTINCT ON ( ol.project_name, ol.corpus_name, ol.section," in sql.replace(
            "\n", " "
        ), "obj_lots_latest должен дедупить по physflat-ключу"
        assert "snapshot_date DESC, ol.id DESC" in sql, "берём последний снапшот физлота"
        assert (
            "v_objective_lots_latest" not in sql
        ), "request-path: view материализует всю таблицу — нужен inline DISTINCT ON (#1964)"


class TestCompetitorAvgAreaPd:
    """#2111: средняя площадь квартиры конкурента (avg_area_pd) доходит до попапа.

    obj_pricing CTE считает ROUND(AVG(oll.area_pd),1) AS avg_area_pd, финальный
    SELECT выбирает p.avg_area_pd, а _competitor_with_coords должен протолкнуть
    ключ в loose-dict competitors[] (parcel.py:651). Mock-based, без БД.
    """

    def test_avg_area_pd_passes_through(self) -> None:
        """_competitor_with_coords сохраняет avg_area_pd из входной строки."""
        row = {
            "obj_id": 1,
            "comm_name": "ЖК Тест",
            "dev_name": "TestDev",
            "avg_price_per_m2_rub": 150000,
            "avg_area_pd": 47.3,
            "units_sold": 10,
            "units_available": 5,
            "lat": 56.838011,
            "lon": 60.597474,
        }
        out = parcels_module._competitor_with_coords(row)
        assert out["avg_area_pd"] == 47.3

    def test_avg_area_pd_none_when_absent(self) -> None:
        """Отсутствие avg_area_pd → ключ есть, значение None (graceful, ЖК без маппинга)."""
        row = {"obj_id": 2, "comm_name": "ЖК Без Цен", "lat": None, "lon": None}
        out = parcels_module._competitor_with_coords(row)
        assert out["avg_area_pd"] is None

    def test_sql_computes_and_selects_avg_area_pd(self) -> None:
        """Competitor-SQL вычисляет AVG(area_pd) и выбирает avg_area_pd в финале."""
        src = inspect.getsource(parcels_module.analyze_parcel)
        marker = "WITH latest_obj AS ("
        start = src.index(marker)
        block = re.sub(r"\s+", " ", src[start : src.index('"""', start)])
        assert "ROUND(AVG(oll.area_pd)::numeric, 1) AS avg_area_pd" in block
        assert "p.avg_area_pd" in block