gendesign/tradein-mvp/backend/tests/services/test_location_index.py

"""Unit tests for app.services.location_index (replaces test_location_coef.py).

No live Postgres needed — DB is a minimal fake returning queued results (mirrors the
convention in tests/tasks/test_cadastral_geo_match.py / the deleted test_location_coef.py).
Covers:
  - pure functions: _category_weight, _in_ekb_bbox, _pct_deviation (incl. monotonicity)
  - _fetch_nearby_poi: qualitative POI ranking (unchanged behaviour from the old module)
  - compute_location_index: out-of-coverage degradation, insufficient-sample degradation
    (citywide AND local), radius-ladder expansion, explicit radius_m override, happy path
  - SQL discipline: psycopg v3 CAST, percentile_cont (not naive AVG/MIN/MAX) for outlier
    robustness
"""

from __future__ import annotations

import os
from typing import Any

# psycopg v3 driver required; stub DATABASE_URL before any app import (settings needs a DSN).
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

from app.services import location_index as lc

# A point well inside the EKB coverage bbox (city centre, Ploshchad 1905 goda area).
_LAT_IN_EKB = 56.838
_LON_IN_EKB = 60.605

# ── Pure functions ──────────────────────────────────────────────────────────


def test_category_weight_known_categories() -> None:
    assert lc._category_weight("metro_stop") == 6.0
    assert lc._category_weight("school") == 5.0
    assert lc._category_weight("kindergarten") == 4.5
    assert lc._category_weight("hospital") == 4.0
    assert lc._category_weight("shop_mall") == 4.0
    assert lc._category_weight("shop_supermarket") == 3.5
    assert lc._category_weight("bus_stop") == 4.5
    assert lc._category_weight("park") == 3.5
    assert lc._category_weight("pharmacy") == 2.5
    assert lc._category_weight("tram_stop") == 2.0
    assert lc._category_weight("shop_small") == 2.0


def test_category_weight_unknown_and_none_fall_back_to_default() -> None:
    assert lc._category_weight("unknown_category") == 1.0
    assert lc._category_weight(None) == 1.0


def test_in_ekb_bbox_center_is_inside() -> None:
    assert lc._in_ekb_bbox(_LAT_IN_EKB, _LON_IN_EKB) is True


def test_in_ekb_bbox_bounds_are_inclusive() -> None:
    assert lc._in_ekb_bbox(56.70, 60.50) is True
    assert lc._in_ekb_bbox(56.95, 60.75) is True


def test_in_ekb_bbox_outside_is_rejected() -> None:
    # Nizhny Tagil — same oblast (region_code=66), well outside the EKB product bbox.
    assert lc._in_ekb_bbox(57.910, 59.970) is False
    # Just past each edge of the bbox.
    assert lc._in_ekb_bbox(56.69, 60.60) is False
    assert lc._in_ekb_bbox(56.96, 60.60) is False
    assert lc._in_ekb_bbox(56.80, 60.49) is False
    assert lc._in_ekb_bbox(56.80, 60.76) is False


def test_pct_deviation_above_and_below_city_median() -> None:
    assert lc._pct_deviation(165_000.0, 150_000.0) == 10.0
    assert lc._pct_deviation(135_000.0, 150_000.0) == -10.0
    assert lc._pct_deviation(150_000.0, 150_000.0) == 0.0


def test_pct_deviation_not_artificially_clamped() -> None:
    """Owner requirement: a genuinely +40% district must read as +40%, not clamped."""
    assert lc._pct_deviation(210_000.0, 150_000.0) == 40.0


def test_pct_deviation_guards_zero_division() -> None:
    assert lc._pct_deviation(100_000.0, 0.0) == 0.0


def test_pct_deviation_is_monotonic_in_local_median() -> None:
    """Индекс строго монотонен по локальной медиане при фиксированной городской — в отличие
    от старого coef (немонотонные бакеты на реальных данных, см. модуль docstring).

    Значения ниже — медианы ₽/м² по дистанционным бакетам от центра ЕКБ, измеренные на 31
    тыс. лотов (аудит владельца продукта), отсортированные по возрастанию. Индекс,
    построенный на этих же локальных медианах, обязан сохранить порядок.
    """
    city_median = 155_000.0
    local_medians_ascending = [
        93_677.0,
        136_729.0,
        150_063.0,
        159_382.0,
        159_486.0,
        191_682.0,
        249_686.0,
    ]
    pct_values = [lc._pct_deviation(m, city_median) for m in local_medians_ascending]
    assert pct_values == sorted(pct_values)


# ── SQL discipline ────────────────────────────────────────────────────────────


def test_no_psycopg_v3_colon_colon_cast() -> None:
    """psycopg v3: never :param::type — must use CAST(:param AS type)."""
    import re

    for sql in (
        lc._MEDIAN_PPM2_LOCAL_SQL,
        lc._MEDIAN_PPM2_CITYWIDE_SQL,
        lc._NEAREST_POI_SQL,
    ):
        assert not re.search(r":\w+::", str(sql.text))


def test_median_queries_use_percentile_not_naive_minmax() -> None:
    """Outlier robustness requirement: percentile_cont(0.5) (median), not AVG/MIN/MAX."""
    for sql in (lc._MEDIAN_PPM2_LOCAL_SQL, lc._MEDIAN_PPM2_CITYWIDE_SQL):
        sql_text = str(sql.text).lower()
        assert "percentile_cont(0.5)" in sql_text
        assert "avg(" not in sql_text
        assert "min(" not in sql_text
        assert "max(" not in sql_text


def test_median_queries_exclude_city_centroid_and_bound_bbox() -> None:
    """Comparable-selection quality control (owner requirement #1): city-centroid geocodes
    excluded (mirrors estimator.py #769 Part E), sample bounded to the EKB bbox."""
    for sql in (lc._MEDIAN_PPM2_LOCAL_SQL, lc._MEDIAN_PPM2_CITYWIDE_SQL):
        sql_text = str(sql.text)
        assert "geo_precision IS DISTINCT FROM 'city'" in sql_text
        assert "bbox_south" in sql_text and "bbox_north" in sql_text
        assert "bbox_west" in sql_text and "bbox_east" in sql_text


# ── _fetch_nearby_poi (qualitative "что рядом" list) ─────────────────────────


class _FakeResult:
    def __init__(
        self,
        *,
        scalar_value: Any = None,
        mapping_rows: list[dict] | None = None,
        mapping_one: dict | None = None,
    ):
        self._scalar_value = scalar_value
        self._mapping_rows = mapping_rows or []
        self._mapping_one = mapping_one

    def scalar(self) -> Any:
        return self._scalar_value

    def mappings(self) -> Any:
        outer = self

        class _Mappings:
            def all(self) -> list[dict]:
                return outer._mapping_rows

            def first(self) -> dict | None:
                return outer._mapping_one

        return _Mappings()


class _FakeDB:
    """Minimal Session stand-in: execute() returns queued results in order."""

    def __init__(self, results: list[_FakeResult]) -> None:
        self._results = list(results)
        self.executed: list[Any] = []

    def execute(self, clause: Any, params: dict | None = None) -> _FakeResult:
        self.executed.append((clause, params))
        return self._results.pop(0)


def test_fetch_nearby_poi_empty_mirror_returns_unavailable() -> None:
    db = _FakeDB([_FakeResult(scalar_value=0)])
    poi, status = lc._fetch_nearby_poi(db, _LAT_IN_EKB, _LON_IN_EKB, lc.DEFAULT_POI_RADIUS_M, 7)
    assert poi == []
    assert status == "unavailable"
    assert len(db.executed) == 1  # only the count probe ran


def test_fetch_nearby_poi_no_poi_in_radius_is_legit_ok() -> None:
    db = _FakeDB([_FakeResult(scalar_value=500), _FakeResult(mapping_rows=[])])
    poi, status = lc._fetch_nearby_poi(db, _LAT_IN_EKB, _LON_IN_EKB, lc.DEFAULT_POI_RADIUS_M, 7)
    assert poi == []
    assert status == "ok"


def test_fetch_nearby_poi_ranks_by_weight_not_distance_only() -> None:
    rows = [
        {"name": "Школа №1", "category": "school", "distance_m": 300.0},
        {"name": "ТЦ Мега", "category": "shop_mall", "distance_m": 900.0},
        {"name": "Метро Ботаническая", "category": "metro_stop", "distance_m": 150.0},
        {"name": "Аптека", "category": "pharmacy", "distance_m": 50.0},
    ]
    db = _FakeDB([_FakeResult(scalar_value=1000), _FakeResult(mapping_rows=rows)])
    poi, status = lc._fetch_nearby_poi(db, _LAT_IN_EKB, _LON_IN_EKB, 1200, 7)
    assert status == "ok"
    assert len(poi) == 4
    # metro_stop (weight 6.0) at 150m outranks school (5.0) at 300m — weight-driven, not
    # distance-only ranking.
    assert poi[0].poi_type == "metro_stop"


def test_fetch_nearby_poi_limits_to_top_n() -> None:
    rows = [
        {"name": f"POI {i}", "category": "shop_small", "distance_m": float(100 + i * 10)}
        for i in range(20)
    ]
    db = _FakeDB([_FakeResult(scalar_value=20), _FakeResult(mapping_rows=rows)])
    poi, _status = lc._fetch_nearby_poi(db, _LAT_IN_EKB, _LON_IN_EKB, 1200, 7)
    assert len(poi) == 7


# ── compute_location_index: degradation + ladder logic ───────────────────────


def test_compute_location_index_out_of_coverage_skips_all_db_calls() -> None:
    """Owner requirement #2: point outside EKB → honest 'no data', never a fallback number.

    Also a perf/honesty check: no DB round-trip at all for an out-of-scope point.
    """
    db = _FakeDB([])
    result = lc.compute_location_index(db, lat=57.910, lon=59.970)  # Nizhny Tagil
    assert result.status == "out_of_coverage"
    assert result.location_index_pct is None
    assert result.local_median_price_per_m2 is None
    assert result.city_median_price_per_m2 is None
    assert result.sample_size == 0
    assert result.nearby_poi == []
    assert result.poi_status == "unavailable"
    assert db.executed == []


def test_compute_location_index_citywide_sample_too_small_short_circuits() -> None:
    """Degenerate citywide reference (e.g. empty dev DB) → insufficient_data without ever
    issuing a local-radius query (nothing to compare against anyway)."""
    db = _FakeDB(
        [
            _FakeResult(scalar_value=0),  # POI mirror empty
            _FakeResult(mapping_one={"median_ppm2": None, "n": 3}),  # citywide: n < MIN
        ]
    )
    result = lc.compute_location_index(db, lat=_LAT_IN_EKB, lon=_LON_IN_EKB)
    assert result.status == "insufficient_data"
    assert result.location_index_pct is None
    assert result.sample_size == 3
    assert len(db.executed) == 2  # poi-count + citywide only — no radius-ladder queries


def test_compute_location_index_first_radius_rung_sufficient() -> None:
    db = _FakeDB(
        [
            _FakeResult(scalar_value=0),  # poi mirror empty
            _FakeResult(mapping_one={"median_ppm2": 150_000.0, "n": 4000}),  # citywide
            _FakeResult(mapping_one={"median_ppm2": 165_000.0, "n": 25}),  # radius[0]=800
        ]
    )
    result = lc.compute_location_index(db, lat=_LAT_IN_EKB, lon=_LON_IN_EKB)
    assert result.status == "ok"
    assert result.radius_m == lc.RADIUS_LADDER_M[0]
    assert result.sample_size == 25
    assert result.local_median_price_per_m2 == 165_000
    assert result.city_median_price_per_m2 == 150_000
    assert result.location_index_pct == 10.0
    assert len(db.executed) == 3  # ladder stopped at rung 1 — no further radius queries


def test_compute_location_index_expands_ladder_when_first_rung_insufficient() -> None:
    db = _FakeDB(
        [
            _FakeResult(scalar_value=0),
            _FakeResult(mapping_one={"median_ppm2": 150_000.0, "n": 4000}),
            _FakeResult(mapping_one={"median_ppm2": 200_000.0, "n": 10}),  # 800m: too few
            _FakeResult(mapping_one={"median_ppm2": 180_000.0, "n": 30}),  # 1500m: enough
        ]
    )
    result = lc.compute_location_index(db, lat=_LAT_IN_EKB, lon=_LON_IN_EKB)
    assert result.status == "ok"
    assert result.radius_m == lc.RADIUS_LADDER_M[1]
    assert result.sample_size == 30
    assert len(db.executed) == 4


def test_compute_location_index_insufficient_even_at_max_radius() -> None:
    db = _FakeDB(
        [
            _FakeResult(scalar_value=0),
            _FakeResult(mapping_one={"median_ppm2": 150_000.0, "n": 4000}),
            _FakeResult(mapping_one={"median_ppm2": 200_000.0, "n": 5}),  # 800m
            _FakeResult(mapping_one={"median_ppm2": 195_000.0, "n": 12}),  # 1500m
            _FakeResult(mapping_one={"median_ppm2": 190_000.0, "n": 15}),  # 2500m — still < 20
        ]
    )
    result = lc.compute_location_index(db, lat=_LAT_IN_EKB, lon=_LON_IN_EKB)
    assert result.status == "insufficient_data"
    assert result.location_index_pct is None
    assert result.local_median_price_per_m2 is None
    assert result.city_median_price_per_m2 == 150_000
    assert result.radius_m == lc.RADIUS_LADDER_M[-1]
    assert result.sample_size == 15  # honest: shows how close it got, not just "no data"
    assert len(db.executed) == 5  # exhausted the full ladder


def test_compute_location_index_explicit_radius_skips_ladder() -> None:
    """An explicit radius_m must be used AS-IS — no adaptive expansion (caller-controlled)."""
    db = _FakeDB(
        [
            _FakeResult(scalar_value=0),
            _FakeResult(mapping_one={"median_ppm2": 150_000.0, "n": 4000}),
            _FakeResult(mapping_one={"median_ppm2": 160_000.0, "n": 50}),  # single query only
        ]
    )
    result = lc.compute_location_index(db, lat=_LAT_IN_EKB, lon=_LON_IN_EKB, radius_m=1000)
    assert result.status == "ok"
    assert result.radius_m == 1000
    assert len(db.executed) == 3  # exactly one radius query, no ladder rungs tried
    nearest_call_params = db.executed[2][1]
    assert nearest_call_params["radius_m"] == 1000


def test_compute_location_index_poi_unavailable_does_not_block_index() -> None:
    """poi_status and status degrade INDEPENDENTLY — an empty POI mirror must not prevent a
    perfectly computable price-based index."""
    db = _FakeDB(
        [
            _FakeResult(scalar_value=0),  # poi mirror empty
            _FakeResult(mapping_one={"median_ppm2": 150_000.0, "n": 4000}),
            _FakeResult(mapping_one={"median_ppm2": 172_500.0, "n": 40}),
        ]
    )
    result = lc.compute_location_index(db, lat=_LAT_IN_EKB, lon=_LON_IN_EKB)
    assert result.status == "ok"
    assert result.poi_status == "unavailable"
    assert result.nearby_poi == []
    assert result.location_index_pct == 15.0