gendesign/backend/tests/test_velocity.py

"""Tests for velocity-score service (#34 D2).

Mock-based — не требуют живой БД.
Источник данных — objective_corpus_room_month (мигрировано с domrf_kn_sale_graph).
Mock shape совместим: sales query возвращает aliases (obj_id, total_sqm,
months_with_data, period_start, period_end, has_mapping) через LEFT JOIN
all_competitors + mapped (OBJ-2: ALL competitors included, unmapped has_mapping=False).
Третий вызов execute — bucket_rows (obj_id, room_bucket, units_sold, sqm_sold).
"""

from __future__ import annotations

import datetime
from unittest.mock import MagicMock, patch

import pytest

from app.services.site_finder.velocity import (
    _EKB_MEDIAN_FALLBACK_SQM_PER_MONTH,
    VelocityResult,
    compute_velocity,
)

# Тестовый WKT — небольшой квадрат в центре ЕКБ.
_PARCEL_WKT = "POINT(60.605 56.838)"


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


def _comp_row(obj_id: int, distance_m: float = 500.0) -> MagicMock:
    r = MagicMock()
    r.__getitem__ = lambda self, k: {
        "obj_id": obj_id,
        "comm_name": f"ЖК-{obj_id}",
        "dev_name": "TestDev",
        "obj_class": "комфорт",
        "district_name": "Ленинский",
        "distance_m": distance_m,
    }[k]
    return r


def _sales_row(
    obj_id: int,
    total_sqm: float,
    months: int,
    start: str = "2024-11-01",
    end: str = "2025-04-01",
    has_mapping: bool = True,
) -> MagicMock:
    r = MagicMock()
    start_d = datetime.date.fromisoformat(start)
    end_d = datetime.date.fromisoformat(end)
    r.__getitem__ = lambda self, k: {
        "obj_id": obj_id,
        "total_sqm": total_sqm,
        "months_with_data": months,
        "period_start": start_d,
        "period_end": end_d,
        # OBJ-2: LEFT JOIN all_competitors — все конкуренты включены,
        # has_mapping=True если есть маппинг в objective_complex_mapping.
        "has_mapping": has_mapping,
    }[k]
    return r


def _bucket_row(obj_id: int, room_bucket: str, units_sold: int, sqm_sold: float) -> MagicMock:
    r = MagicMock()
    r.__getitem__ = lambda self, k: {
        "obj_id": obj_id,
        "room_bucket": room_bucket,
        "units_sold": units_sold,
        "sqm_sold": sqm_sold,
    }[k]
    return r


def _make_db(
    comp_rows: list,
    sales_rows: list,
    bucket_rows: list | None = None,
) -> MagicMock:
    """Сконструировать mock Session с тремя последовательными вызовами execute.

    Порядок: comp_rows → sales_rows → bucket_rows.
    bucket_rows=None → пустой список (bucket query gracefully degraded).
    """
    db = MagicMock()
    execute_results = []
    for rows in [comp_rows, sales_rows, bucket_rows if bucket_rows is not None else []]:
        result = MagicMock()
        result.mappings.return_value.all.return_value = rows
        execute_results.append(result)
    db.execute.side_effect = execute_results
    return db


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


def test_no_competitors_returns_none():
    """Нет ЖК в радиусе → None."""
    db = _make_db(comp_rows=[], sales_rows=[])
    result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)
    assert result is None


def test_no_sales_data_returns_none():
    """ЖК есть, но нет данных objective_corpus_room_month → None."""
    comp_rows = [_comp_row(1), _comp_row(2)]
    db = _make_db(comp_rows=comp_rows, sales_rows=[])
    result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)
    assert result is None


def test_healthy_sales_returns_result():
    """12 конкурентов с нормальными продажами → score в (0,1), confidence='high'."""
    n = 12
    comp_rows = [_comp_row(i, distance_m=300.0 + i * 100) for i in range(1, n + 1)]
    # Каждый ЖК продаёт 4500 м² за 6 мес → 750 м²/мес. Суммарно: 4500*12 = 54000 за 6 мес.
    sales_rows = [_sales_row(i, total_sqm=4500.0, months=6) for i in range(1, n + 1)]

    db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows)

    with patch(
        "app.services.site_finder.velocity._get_ekb_median",
        return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH,
    ):
        result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)

    assert result is not None
    assert result.competitors_count == n
    assert 0.0 < result.velocity_score <= 1.0
    assert result.confidence == "high"
    assert result.months_observed == 6


def test_few_competitors_low_confidence():
    """2 конкурента → confidence='low'."""
    comp_rows = [_comp_row(1), _comp_row(2)]
    sales_rows = [_sales_row(1, total_sqm=3000.0, months=2)]

    db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows)

    with patch(
        "app.services.site_finder.velocity._get_ekb_median",
        return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH,
    ):
        result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)

    assert result is not None
    assert result.confidence == "low"


def test_medium_confidence():
    """7 конкурентов, 4 месяца → confidence='medium'."""
    n = 7
    comp_rows = [_comp_row(i) for i in range(1, n + 1)]
    sales_rows = [_sales_row(i, total_sqm=4000.0, months=4) for i in range(1, n + 1)]

    db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows)

    with patch(
        "app.services.site_finder.velocity._get_ekb_median",
        return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH,
    ):
        result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)

    assert result is not None
    assert result.confidence == "medium"


def test_ekb_median_fallback_used_when_none():
    """Если _get_ekb_median вернул None — используется fallback-константа."""
    comp_rows = [_comp_row(1)]
    sales_rows = [_sales_row(1, total_sqm=9000.0, months=6)]

    db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows)

    with patch("app.services.site_finder.velocity._get_ekb_median", return_value=None):
        result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)

    assert result is not None
    assert result.ekb_median_sqm == _EKB_MEDIAN_FALLBACK_SQM_PER_MONTH


def test_score_capped_at_1():
    """Огромный объём → score не превышает 1.0."""
    comp_rows = [_comp_row(1)]
    # 1 000 000 м² за месяц — абсурдно много
    sales_rows = [_sales_row(1, total_sqm=6_000_000.0, months=6)]

    db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows)

    with patch(
        "app.services.site_finder.velocity._get_ekb_median",
        return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH,
    ):
        result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)

    assert result is not None
    assert result.velocity_score == pytest.approx(1.0)


def test_score_zero_when_no_sales_sqm():
    """total_sqm=0 → VelocityResult с velocity_data_available=False, score=0.

    OBJ-2: функция больше не возвращает None при нулевых продажах —
    возвращает «пустое» состояние (velocity_data_available=False, source='none').
    Rosreestr-fallback пропускается т.к. cad_quarter не передан.
    """
    comp_rows = [_comp_row(1)]
    # total_sqm=0 — нет продаж → velocity_data_available=False
    sales_rows = [_sales_row(1, total_sqm=0.0, months=5)]

    db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows)

    with patch(
        "app.services.site_finder.velocity._get_ekb_median",
        return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH,
    ):
        result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)

    assert result is not None
    assert result.velocity_data_available is False
    assert result.velocity_score == pytest.approx(0.0)
    assert result.velocity_source == "none"


def test_as_dict_structure():
    """as_dict() содержит все ожидаемые ключи, включая by_room_bucket."""
    vr = VelocityResult(
        competitors_count=5,
        monthly_velocity_sqm=3000.0,
        ekb_median_sqm=4500.0,
        velocity_score=0.333,
        confidence="medium",
        months_observed=4,
        period_start="2024-11",
        period_end="2025-02",
        sample_competitors=[],
        by_room_bucket={"1": {"units": 10, "sqm": 450.0, "complexes_count": 2}},
        velocity_data_available=True,
        velocity_source="objective",
    )
    d = vr.as_dict()
    assert "competitors_count" in d
    assert "velocity_score" in d
    assert "confidence" in d
    assert "period" in d
    assert d["period"]["start"] == "2024-11"
    assert d["period"]["end"] == "2025-02"
    assert d["velocity_score"] == pytest.approx(0.333, abs=1e-3)
    assert "by_room_bucket" in d
    assert d["by_room_bucket"]["1"]["units"] == 10
    # OBJ-2: новые поля velocity_data_available и velocity_source
    assert d["velocity_data_available"] is True
    assert d["velocity_source"] == "objective"


def test_sample_competitors_top5():
    """sample_competitors содержит не более 5 элементов, отсортированных по убыванию."""
    n = 8
    comp_rows = [_comp_row(i) for i in range(1, n + 1)]
    sales_rows = [_sales_row(i, total_sqm=float(i * 1000), months=5) for i in range(1, n + 1)]
    db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows)

    with patch(
        "app.services.site_finder.velocity._get_ekb_median",
        return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH,
    ):
        result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)

    assert result is not None
    assert len(result.sample_competitors) <= 5
    sqms = [c["total_sqm_period"] for c in result.sample_competitors]
    assert sqms == sorted(sqms, reverse=True)


def test_by_room_bucket_aggregation():
    """by_room_bucket агрегирует units/sqm поверх всех конкурентов корректно."""
    comp_rows = [_comp_row(1), _comp_row(2)]
    sales_rows = [
        _sales_row(1, total_sqm=3000.0, months=3),
        _sales_row(2, total_sqm=2000.0, months=3),
    ]
    bucket_rows = [
        _bucket_row(1, "студия", units_sold=38, sqm_sold=1520.0),
        _bucket_row(1, "1", units_sold=22, sqm_sold=990.0),
        _bucket_row(2, "студия", units_sold=18, sqm_sold=720.0),
        _bucket_row(2, "1", units_sold=13, sqm_sold=585.0),
    ]
    db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows, bucket_rows=bucket_rows)

    with patch(
        "app.services.site_finder.velocity._get_ekb_median",
        return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH,
    ):
        result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)

    assert result is not None
    assert "студия" in result.by_room_bucket
    assert "1" in result.by_room_bucket
    # студия: 38+18=56 units, complexes from obj 1 and 2
    assert result.by_room_bucket["студия"]["units"] == 56
    assert result.by_room_bucket["студия"]["complexes_count"] == 2
    # 1-к: 22+13=35 units
    assert result.by_room_bucket["1"]["units"] == 35
    # sqm rounded
    assert result.by_room_bucket["студия"]["sqm"] == pytest.approx(2240.0)


def test_by_room_bucket_empty_when_no_bucket_data():
    """Если bucket query вернул пустой список — by_room_bucket пустой dict."""
    comp_rows = [_comp_row(1)]
    sales_rows = [_sales_row(1, total_sqm=5000.0, months=5)]
    db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows, bucket_rows=[])

    with patch(
        "app.services.site_finder.velocity._get_ekb_median",
        return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH,
    ):
        result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)

    assert result is not None
    assert result.by_room_bucket == {}


def test_sample_competitors_include_by_room_bucket():
    """sample_competitors каждого элемента содержит by_room_bucket."""
    comp_rows = [_comp_row(1), _comp_row(2)]
    sales_rows = [
        _sales_row(1, total_sqm=6000.0, months=4),
        _sales_row(2, total_sqm=4000.0, months=4),
    ]
    bucket_rows = [
        _bucket_row(1, "2", units_sold=30, sqm_sold=1800.0),
        _bucket_row(2, "2", units_sold=20, sqm_sold=1200.0),
    ]
    db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows, bucket_rows=bucket_rows)

    with patch(
        "app.services.site_finder.velocity._get_ekb_median",
        return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH,
    ):
        result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)

    assert result is not None
    for comp in result.sample_competitors:
        assert "by_room_bucket" in comp
    # obj_id=1 had bucket data
    top = result.sample_competitors[0]
    assert top["obj_id"] == 1
    assert top["by_room_bucket"].get("2") == 30