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

"""Unit tests for app.services.location_coef (#2045 BE-3, LocationDrawer).

No live Postgres needed — DB is a minimal fake returning queued results (mirrors the
convention in tests/tasks/test_cadastral_geo_match.py). Covers:
  - pure functions: _category_weight, _score_to_coef, normalization constants
  - compute_location_coef: weighted top-N scoring, empty-mirror graceful fallback,
    no-POI-in-radius (legit zero-score, NOT "unavailable")
"""

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_coef as lc

# ── 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_top7_weight_sum_matches_ptica() -> None:
    """Same category set as Site Finder → identical top-7 normalization constant (31.5)."""
    assert lc._TOP7_WEIGHT_SUM == 31.5
    assert abs(lc._MAX_STRAIGHT_SCORE - 0.315) < 1e-9


def test_score_to_coef_bounds() -> None:
    assert lc._score_to_coef(0.0) == 0.95
    assert lc._score_to_coef(100.0) == 1.05


def test_score_to_coef_midpoint() -> None:
    assert lc._score_to_coef(50.0) == 1.0


def test_score_to_coef_is_monotonic() -> None:
    scores = [0.0, 10.0, 25.0, 50.0, 75.0, 90.0, 100.0]
    coefs = [lc._score_to_coef(s) for s in scores]
    assert coefs == sorted(coefs)


# ── compute_location_coef with a fake DB ─────────────────────────────────────


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

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

    def mappings(self) -> Any:
        class _Mappings:
            def __init__(self, rows: list[dict]) -> None:
                self._rows = rows

            def all(self) -> list[dict]:
                return self._rows

        return _Mappings(self._mapping_rows)


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_compute_location_coef_empty_mirror_returns_unavailable() -> None:
    """osm_poi_ekb_local not yet refreshed (count=0) → unavailable, no fabricated factors."""
    db = _FakeDB([_FakeResult(scalar_value=0)])
    result = lc.compute_location_coef(db, lat=56.84, lon=60.6)
    assert result.coef == 1.0
    assert result.factors == []
    assert result.geo_source == "unavailable"
    # Only the count probe ran — no nearest-POI query issued against an empty mirror.
    assert len(db.executed) == 1


def test_compute_location_coef_no_poi_in_radius_is_legit_zero_score() -> None:
    """Mirror populated (count>0) but nothing within radius → coef floor, NOT unavailable."""
    db = _FakeDB(
        [
            _FakeResult(scalar_value=500),  # mirror has rows elsewhere
            _FakeResult(mapping_rows=[]),  # nothing near this point
        ]
    )
    result = lc.compute_location_coef(db, lat=56.84, lon=60.6)
    assert result.factors == []
    assert result.geo_source == "osm_poi_ekb"
    assert result.coef == lc._score_to_coef(0.0) == 0.95


def test_compute_location_coef_weights_and_ranks_top_n() -> None:
    """Nearer + higher-weight-category POI ranks above farther/lower-weight ones."""
    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)])
    result = lc.compute_location_coef(db, lat=56.84, lon=60.6, top_n=7)

    assert result.geo_source == "osm_poi_ekb"
    assert len(result.factors) == 4
    # metro_stop (weight 6.0) at 150m beats school (5.0) at 300m despite being closer only
    # marginally — sanity check the ranking is weight-driven, not distance-only.
    assert result.factors[0].poi_type == "metro_stop"
    # Weights strictly descending (sorted DESC by weight before slicing to top_n).
    weights = [f.weight for f in result.factors]
    assert weights == sorted(weights, reverse=True)
    # coef must land inside the documented [0.95, 1.05] MVP range.
    assert 0.95 <= result.coef <= 1.05


def test_compute_location_coef_limits_to_top_n() -> None:
    """More than top_n candidates → only top_n factors surface in the response."""
    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)])
    result = lc.compute_location_coef(db, lat=56.84, lon=60.6, top_n=7)
    assert len(result.factors) == 7


def test_compute_location_coef_unknown_category_uses_default_weight() -> None:
    rows = [{"name": "Неизвестный POI", "category": "some_new_osm_tag", "distance_m": 200.0}]
    db = _FakeDB([_FakeResult(scalar_value=1), _FakeResult(mapping_rows=rows)])
    result = lc.compute_location_coef(db, lat=56.84, lon=60.6)
    assert len(result.factors) == 1
    expected_weight = (1.0 / (200.0 + 100.0)) * lc.CATEGORY_WEIGHTS["default"]
    assert abs(result.factors[0].weight - round(expected_weight, 6)) < 1e-9


def test_compute_location_coef_passes_radius_param() -> None:
    """radius_m is forwarded as a bound param (psycopg v3 CAST discipline, no :p::type)."""
    db = _FakeDB([_FakeResult(scalar_value=1), _FakeResult(mapping_rows=[])])
    lc.compute_location_coef(db, lat=56.84, lon=60.6, radius_m=1500)
    _clause, params = db.executed[1]
    assert params is not None
    assert params["radius_m"] == 1500


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

    assert not re.search(r":\w+::", str(lc._NEAREST_POI_SQL.text))