gendesign/backend/tests/sql/test_2464_area_bin_unknown.py

"""Лоты без площади — отдельная корзина, а не «<25 м²» (#2464).

`_SUPPLY_ONLY_LOTS_SQL` раскладывал `area_pd IS NULL` в ту же корзину `'<25'`,
что и настоящие студии. Замер прода 20.08.2026 (последний снапшот на физлот,
premise_kind='квартира', не проданные):

    в продаже            181 353
    без area_pd           11 557   (6.4 %)
    реально < 25 м²        7 013

То есть корзина «<25» состояла из неизвестного на 62 % и завышала долю мелких
лотов в структуре остатков — это блок «По предложению (без темпа продаж)»
отчёта. Зеркала у такого отображения не было: `layout_signature.area_bin`
принимает float и NULL-ветки не имеет вовсе, а velocity-MV по площади не
группирует. Исключать лоты тоже нельзя — они реально в продаже.

Тест герметичный и прогоняет НАСТОЯЩИЙ SQL: временная таблица `objective_lots`
затеняет боевую в пределах сессии (temp-схема идёт раньше public в search_path),
поэтому запрос из модуля выполняется дословно, а прод-данные не читаются.
"""

from __future__ import annotations

import os

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

from decimal import Decimal

import pytest
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker


def _dsn() -> str:
    raw = os.environ.get("TEST_DATABASE_URL") or os.environ.get(
        "DATABASE_URL", "postgresql+psycopg://gendesign@localhost:15432/gendesign"
    )
    return (
        raw
        if raw.startswith("postgresql+")
        else raw.replace("postgresql://", "postgresql+psycopg://")
    )


def _db_reachable() -> tuple[bool, str]:
    try:
        eng = create_engine(_dsn(), connect_args={"connect_timeout": 3})
        with eng.connect() as c:
            c.execute(text("SELECT 1"))
        return True, ""
    except Exception as exc:
        return False, str(exc)


_DB_OK, _DB_ERR = _db_reachable()
pytestmark = pytest.mark.skipif(not _DB_OK, reason=f"Postgres недоступен: {_DB_ERR}")

_TEMP_TABLE = """
CREATE TEMP TABLE objective_lots (
    id               bigserial PRIMARY KEY,
    project_name     text,
    corpus_name      text,
    section          text,
    floor            integer,
    lot_number       text,
    rooms_int        smallint,
    area_pd          numeric,
    price_per_m2_rub numeric,
    is_sold          boolean,
    status           text,
    premise_kind     text,
    snapshot_date    date
) ON COMMIT DROP
"""

# Один проект, четыре лота: без площади, настоящая студия, 30 м², 50 м².
_ROWS = [
    ("ЖК Пример", "к1", "с1", 3, "101", 1, None, 200000, "квартира"),
    ("ЖК Пример", "к1", "с1", 3, "102", 1, Decimal("22.0"), 200000, "квартира"),
    ("ЖК Пример", "к1", "с1", 4, "201", 1, Decimal("30.0"), 210000, "квартира"),
    ("ЖК Пример", "к1", "с1", 5, "301", 2, Decimal("50.0"), 220000, "квартира"),
]


def _buckets() -> dict[str, int]:
    """Выполнить настоящий запрос модуля на временной таблице → {корзина: лотов}."""
    from app.services.site_finder.best_layouts import (
        _SUPPLY_ONLY_LOTS_SQL,
        _SUPPLY_ONLY_PREMISE_KIND,
        _SUPPLY_ONLY_PRICE_MAX,
        _SUPPLY_ONLY_PRICE_MIN,
    )

    session = sessionmaker(bind=create_engine(_dsn()))()
    try:
        session.execute(text(_TEMP_TABLE))
        for pn, corp, sec, fl, lot, rooms, area, price, kind in _ROWS:
            session.execute(
                text(
                    "INSERT INTO objective_lots (project_name, corpus_name, section, floor,"
                    " lot_number, rooms_int, area_pd, price_per_m2_rub, is_sold, status,"
                    " premise_kind, snapshot_date) VALUES (:pn,:corp,:sec,:fl,:lot,:rooms,"
                    ":area,:price,false,'в продаже',:kind,CURRENT_DATE)"
                ),
                {
                    "pn": pn,
                    "corp": corp,
                    "sec": sec,
                    "fl": fl,
                    "lot": lot,
                    "rooms": rooms,
                    "area": area,
                    "price": price,
                    "kind": kind,
                },
            )
        rows = (
            session.execute(
                _SUPPLY_ONLY_LOTS_SQL,
                {
                    "names": ["ЖК Пример"],
                    "premise_kind": _SUPPLY_ONLY_PREMISE_KIND,
                    "price_min": _SUPPLY_ONLY_PRICE_MIN,
                    "price_max": _SUPPLY_ONLY_PRICE_MAX,
                },
            )
            .mappings()
            .all()
        )
        return {str(r["area_bin"]): int(r["lots_count"]) for r in rows}
    finally:
        session.rollback()
        session.close()


def test_lot_without_area_is_not_counted_as_tiny() -> None:
    """Головной: лот без площади не должен попадать в корзину настоящих студий.

    На origin/main корзина «<25» содержит 2 лота вместо одного — второй это
    лот с area_pd IS NULL.
    """
    b = _buckets()
    assert b.get("<25") == 1, (
        f"в корзину «<25» попал лот без площади: {b} — на проде так туда "
        "уезжают 11 557 квартир при 7 013 настоящих"
    )


def test_unknown_area_gets_its_own_bucket() -> None:
    """Лот без площади обязан быть виден отдельно, а не исчезнуть.

    Исключение занизило бы предложение на 6.4 %: лоты реально в продаже.
    """
    b = _buckets()
    assert b.get("н/д") == 1, f"корзины «н/д» нет — лот без площади потерян: {b}"


def test_total_lots_preserved() -> None:
    """Контроль сохранности: сумма по корзинам равна числу лотов в продаже."""
    b = _buckets()
    assert sum(b.values()) == len(_ROWS), f"лоты потеряны или задвоены: {b}"


def test_normal_buckets_unchanged() -> None:
    """Контроль от переусердствования: обычные площади раскладываются как раньше."""
    b = _buckets()
    assert b.get("25-40") == 1, f"корзина 25-40: {b}"
    assert b.get("40-60") == 1, f"корзина 40-60: {b}"