7f06bd5608
All checks were successful
CI / changes (pull_request) Successful in 9s
CI / openapi-codegen-check (pull_request) Has been skipped
CI Trade-In / changes (pull_request) Successful in 11s
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI Trade-In / backend-tests (pull_request) Successful in 52s
The normalization used an unreachable all-POI-at-0m maximum, compressing real scores into ~0.975-0.98 so every address read as "location reduces value". Normalize against a realistic reference distance instead; strong central ~ 1.0, poor locations stay below. Informational only (estimator does not use it). Kept the uncalibrated-heuristic caveat. LOW audit R2 (#7a).
229 lines
10 KiB
Python
229 lines
10 KiB
Python
"""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 weight-sum constant (31.5)."""
|
||
assert lc._TOP7_WEIGHT_SUM == 31.5
|
||
|
||
|
||
def test_max_straight_score_normalized_at_reference_distance_not_zero() -> None:
|
||
"""Audit R2 #7a: denominator uses _REF_DISTANCE_M (100m), not an unreachable d=0 max.
|
||
|
||
Old d=0 normalization was _TOP7_WEIGHT_SUM / 100 == 0.315 — the theoretical max with all
|
||
top-7 categories AT the doorstep. That compressed realistic scores (~25-30/100 for even a
|
||
strong central address) into the bottom third of the range. The new normalization halves
|
||
the max (denominator distance+100 doubles from 100 to 200 at REF=100m), doubling realistic
|
||
scores instead.
|
||
"""
|
||
assert lc._REF_DISTANCE_M == 100.0
|
||
assert abs(lc._MAX_STRAIGHT_SCORE - 0.1575) < 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_strong_central_fixture_lands_near_one() -> None:
|
||
"""Audit R2 #7a: a realistic strong central-EKB POI profile now reads as ~1.0, not ~0.98.
|
||
|
||
Fixture mirrors a real strong central address audit: near transit/kindergarten/pharmacy,
|
||
but school/metro/mall noticeably farther (a typical quarter profile, NOT everything at
|
||
the doorstep). Under the OLD d=0 normalization this fixture scores ~28/100 → coef ~0.978
|
||
(verified separately against the pre-fix formula). After re-centering on _REF_DISTANCE_M
|
||
it should land close to 1.0 (within the documented ±0.01 calibration target).
|
||
"""
|
||
rows = [
|
||
{"name": "Остановка", "category": "bus_stop", "distance_m": 80.0},
|
||
{"name": "Аптека", "category": "pharmacy", "distance_m": 120.0},
|
||
{"name": "Супермаркет", "category": "shop_supermarket", "distance_m": 180.0},
|
||
{"name": "Детсад №5", "category": "kindergarten", "distance_m": 220.0},
|
||
{"name": "Школа №10", "category": "school", "distance_m": 350.0},
|
||
{"name": "Метро Геологическая", "category": "metro_stop", "distance_m": 500.0},
|
||
{"name": "ТЦ Гринвич", "category": "shop_mall", "distance_m": 700.0},
|
||
]
|
||
db = _FakeDB([_FakeResult(scalar_value=len(rows)), _FakeResult(mapping_rows=rows)])
|
||
result = lc.compute_location_coef(db, lat=56.838, lon=60.605, top_n=7)
|
||
|
||
assert result.geo_source == "osm_poi_ekb"
|
||
assert len(result.factors) == 7
|
||
assert abs(result.coef - 1.0) <= 0.01
|
||
|
||
|
||
def test_compute_location_coef_weak_fixture_stays_well_below_one() -> None:
|
||
"""A sparse/poor location (only far, low-weight POI) must still stay well below 1.0."""
|
||
rows = [
|
||
{"name": "Магазинчик", "category": "shop_small", "distance_m": 950.0},
|
||
{"name": "Прочее", "category": "some_unknown_tag", "distance_m": 1100.0},
|
||
]
|
||
db = _FakeDB([_FakeResult(scalar_value=len(rows)), _FakeResult(mapping_rows=rows)])
|
||
result = lc.compute_location_coef(db, lat=56.838, lon=60.605, top_n=7)
|
||
|
||
assert result.geo_source == "osm_poi_ekb"
|
||
assert result.coef < 0.98
|
||
|
||
|
||
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))
|