Light1YT
50555821fb
Some checks failed
CI / changes (push) Successful in 7s
CI / frontend-tests (push) Has been skipped
CI / changes (pull_request) Successful in 7s
CI / frontend-tests (pull_request) Has been skipped
CI / backend-tests (push) Successful in 7m42s
CI / backend-tests (pull_request) Successful in 7m40s
Deploy / changes (push) Has been cancelled
Deploy / build-backend (push) Has been cancelled
Deploy / build-worker (push) Has been cancelled
Deploy / build-frontend (push) Has been cancelled
Deploy / deploy (push) Has been cancelled
#41: per-category routing-radius decay через reusable ORS /matrix client (foot-walk время + per-category пороги + decay-curves), opt-in ?decay=routing, graceful straight-line fallback (analyze не 500 при ORS down) + length-guard durations==destinations. #44: +5 OSM-категорий (transformer/gas/water/sewerage/heat) via Overpass nwr; nearest_{water_main,substation,gas,heat}_m в analyze.utilities. #255bk: ST_AsGeoJSON(CAST(geom AS geometry)) в _get_cad_zouit_overlaps → geom_geojson. Closes #41 Closes #44
335 lines
13 KiB
Python
335 lines
13 KiB
Python
"""Tests for POI weighted score service (B6).
|
||
|
||
Юнит-тесты для чистой функции — без DB.
|
||
"""
|
||
|
||
import pytest
|
||
|
||
from app.services.site_finder import ors_client, poi_score
|
||
from app.services.site_finder.poi_score import (
|
||
CATEGORY_WEIGHTS,
|
||
PoiScoreResponse,
|
||
_category_radius_min,
|
||
_category_weight,
|
||
_decay_linear,
|
||
_decay_piecewise,
|
||
_decay_step,
|
||
compute_poi_routing_decay,
|
||
compute_poi_weighted_top7,
|
||
)
|
||
|
||
# ── unit: _category_weight ─────────────────────────────────────────────────────
|
||
|
||
|
||
def test_category_weight_metro():
|
||
"""Метро имеет наибольший вес из всех категорий."""
|
||
metro_w = _category_weight("metro_stop")
|
||
for cat in CATEGORY_WEIGHTS:
|
||
if cat != "metro_stop" and cat != "default":
|
||
assert metro_w >= _category_weight(
|
||
cat
|
||
), f"metro_stop weight {metro_w} должен быть >= {cat} weight {_category_weight(cat)}"
|
||
|
||
|
||
def test_category_weight_unknown_returns_default():
|
||
w = _category_weight("unknown_category_xyz")
|
||
assert w == CATEGORY_WEIGHTS["default"]
|
||
|
||
|
||
def test_category_weight_all_positive():
|
||
"""Все веса в CATEGORY_WEIGHTS должны быть положительными (B6 — ranking, не штраф)."""
|
||
for cat, w in CATEGORY_WEIGHTS.items():
|
||
assert w > 0, f"Вес {cat}={w} должен быть > 0"
|
||
|
||
|
||
# ── unit: weight formula ratio ─────────────────────────────────────────────────
|
||
|
||
|
||
def test_weight_formula_ratio():
|
||
"""Ближний объект той же категории должен иметь больший вес."""
|
||
cat = "school"
|
||
cw = _category_weight(cat)
|
||
w_near = (1.0 / (100.0 + 100.0)) * cw # 100м
|
||
w_far = (1.0 / (1000.0 + 100.0)) * cw # 1000м
|
||
assert w_near > w_far
|
||
|
||
|
||
def test_weight_formula_category_dominates_at_equal_distance():
|
||
"""При одинаковом расстоянии метро должно быть впереди автобусной остановки."""
|
||
dist = 500.0
|
||
w_metro = (1.0 / (dist + 100.0)) * _category_weight("metro_stop")
|
||
w_bus = (1.0 / (dist + 100.0)) * _category_weight("bus_stop")
|
||
assert w_metro > w_bus
|
||
|
||
|
||
# ── unit: compute_poi_weighted_top7 with mock DB ───────────────────────────────
|
||
|
||
|
||
class _MockMappings:
|
||
def __init__(self, data: list[dict]) -> None:
|
||
self._data = data
|
||
|
||
def all(self) -> list[dict]:
|
||
return self._data # type: ignore[return-value]
|
||
|
||
|
||
class _MockResult:
|
||
def __init__(self, data: list[dict]) -> None:
|
||
self._data = data
|
||
|
||
def mappings(self) -> "_MockMappings":
|
||
return _MockMappings(self._data)
|
||
|
||
|
||
class _MockDb:
|
||
"""Минимальный мок SQLAlchemy Session для тестирования без БД."""
|
||
|
||
def __init__(self, rows: list[dict]) -> None:
|
||
self._rows = rows
|
||
|
||
def execute(self, *_args: object, **_kwargs: object) -> _MockResult:
|
||
return _MockResult(self._rows)
|
||
|
||
|
||
def _make_row(name: str, category: str, distance_m: float) -> dict:
|
||
return {
|
||
"name": name,
|
||
"category": category,
|
||
"tags": {},
|
||
"distance_m": distance_m,
|
||
}
|
||
|
||
|
||
def test_top7_returns_at_most_7():
|
||
rows = [_make_row(f"POI {i}", "school", float(i * 50)) for i in range(1, 20)]
|
||
db = _MockDb(rows)
|
||
result = compute_poi_weighted_top7(db, "66:41:0204016:10", 56.838, 60.605)
|
||
assert isinstance(result, PoiScoreResponse)
|
||
assert len(result.top_poi) <= 7
|
||
|
||
|
||
def test_top7_sorted_by_weight_desc():
|
||
rows = [
|
||
_make_row("Дальняя школа", "school", 1500.0),
|
||
_make_row("Метро", "metro_stop", 300.0),
|
||
_make_row("Близкая школа", "school", 100.0),
|
||
]
|
||
db = _MockDb(rows)
|
||
result = compute_poi_weighted_top7(db, "66:41:0204016:10", 56.838, 60.605)
|
||
weights = [item.weight for item in result.top_poi]
|
||
assert weights == sorted(weights, reverse=True), "top_poi должны быть по weight DESC"
|
||
|
||
|
||
def test_metro_beats_school_at_equal_distance():
|
||
"""Метро в 300м должно быть на первом месте перед школой в 300м (равное расстояние)."""
|
||
rows = [
|
||
_make_row("Школа №1", "school", 300.0),
|
||
_make_row("Метро Чкаловская", "metro_stop", 300.0),
|
||
]
|
||
db = _MockDb(rows)
|
||
result = compute_poi_weighted_top7(db, "66:41:0204016:10", 56.838, 60.605)
|
||
assert (
|
||
result.top_poi[0].category == "metro_stop"
|
||
), "При равном расстоянии метро (category_weight=6.0) должно быть выше школы (5.0)"
|
||
|
||
|
||
def test_metro_first_when_close():
|
||
"""Метро в 50м должно быть на первом месте перед школой в 300м."""
|
||
rows = [
|
||
_make_row("Школа №1", "school", 300.0),
|
||
_make_row("Метро Чкаловская", "metro_stop", 50.0),
|
||
]
|
||
db = _MockDb(rows)
|
||
result = compute_poi_weighted_top7(db, "66:41:0204016:10", 56.838, 60.605)
|
||
assert result.top_poi[0].category == "metro_stop", (
|
||
"Метро (weight=6.0) в 50м должно быть впереди школы (weight=5.0) в 300м — "
|
||
f"metro_weight={(1/(50+100))*6:.5f} vs school_weight={(1/(300+100))*5:.5f}"
|
||
)
|
||
|
||
|
||
def test_empty_db_returns_empty_top_poi():
|
||
db = _MockDb([])
|
||
result = compute_poi_weighted_top7(db, "66:41:0204016:10", 56.838, 60.605)
|
||
assert result.top_poi == []
|
||
assert result.cad_num == "66:41:0204016:10"
|
||
assert result.radius_m == 2000
|
||
|
||
|
||
def test_address_built_from_tags():
|
||
rows = [
|
||
{
|
||
"name": "Магазин",
|
||
"category": "shop_small",
|
||
"tags": {"addr:street": "ул. Ленина", "addr:housenumber": "10"},
|
||
"distance_m": 200.0,
|
||
}
|
||
]
|
||
db = _MockDb(rows)
|
||
result = compute_poi_weighted_top7(db, "test", 56.838, 60.605)
|
||
assert result.top_poi[0].address == "ул. Ленина, 10"
|
||
|
||
|
||
def test_address_none_when_no_tags():
|
||
rows = [_make_row("Парк", "park", 400.0)]
|
||
db = _MockDb(rows)
|
||
result = compute_poi_weighted_top7(db, "test", 56.838, 60.605)
|
||
assert result.top_poi[0].address is None
|
||
|
||
|
||
# ── #41: per-category decay curves (pure) ────────────────────────────────────
|
||
|
||
|
||
def test_decay_linear_bounds():
|
||
assert _decay_linear(0.0, 20.0) == pytest.approx(1.0)
|
||
assert _decay_linear(10.0, 20.0) == pytest.approx(0.5)
|
||
assert _decay_linear(20.0, 20.0) == 0.0
|
||
assert _decay_linear(25.0, 20.0) == 0.0 # за порогом → 0
|
||
|
||
|
||
def test_decay_piecewise_full_then_drop():
|
||
"""Школа: полный вес близко, спад к краю, 0 за порогом."""
|
||
r = 15.0
|
||
assert _decay_piecewise(2.0, r) == pytest.approx(1.0) # 0..5 мин полный
|
||
assert _decay_piecewise(5.0, r) == pytest.approx(1.0)
|
||
mid = _decay_piecewise(10.0, r)
|
||
assert 0.0 < mid < 1.0
|
||
assert _decay_piecewise(15.0, r) == 0.0
|
||
assert _decay_piecewise(99.0, r) == 0.0
|
||
|
||
|
||
def test_decay_step_binary():
|
||
"""Парк: есть в радиусе (1.0) либо нет (0.0)."""
|
||
assert _decay_step(9.0, 10.0) == 1.0
|
||
assert _decay_step(10.0, 10.0) == 1.0
|
||
assert _decay_step(10.1, 10.0) == 0.0
|
||
|
||
|
||
def test_category_radius_min_defaults():
|
||
assert _category_radius_min("metro_stop") == 20.0
|
||
assert _category_radius_min("park") == 10.0
|
||
# неизвестная категория → дефолт
|
||
assert _category_radius_min("unknown_xyz") == poi_score._DEFAULT_RADIUS_MIN
|
||
|
||
|
||
# ── #41: compute_poi_routing_decay (mock DB + monkeypatched ORS) ─────────────
|
||
|
||
|
||
def _make_routing_row(name: str, category: str, distance_m: float, lon=60.6, lat=56.8) -> dict:
|
||
return {
|
||
"name": name,
|
||
"category": category,
|
||
"tags": {},
|
||
"lon": lon,
|
||
"lat": lat,
|
||
"distance_m": distance_m,
|
||
}
|
||
|
||
|
||
def test_routing_decay_drops_poi_beyond_radius(monkeypatch):
|
||
"""POI за порогом доступности (минут) выбывает из результата."""
|
||
rows = [
|
||
_make_routing_row("Близкое метро", "metro_stop", 400.0),
|
||
_make_routing_row("Далёкая школа", "school", 1800.0),
|
||
]
|
||
db = _MockDb(rows)
|
||
|
||
# ORS: метро 5 мин (в пределах 20), школа 30 мин (за порогом 15) → школа выбывает.
|
||
def fake_matrix(_lon, _lat, dests, **_kw):
|
||
return [5.0, 30.0]
|
||
|
||
monkeypatch.setattr(ors_client, "matrix_durations_min", fake_matrix)
|
||
monkeypatch.setattr(ors_client, "is_configured", lambda: True)
|
||
|
||
result = compute_poi_routing_decay(db, "66:41:0204016:10", 56.838, 60.605)
|
||
cats = [i.category for i in result.top_poi]
|
||
assert "metro_stop" in cats
|
||
assert "school" not in cats, "школа в 30 мин (> 15 мин порог) должна выбыть"
|
||
|
||
|
||
def test_routing_decay_closer_time_higher_weight(monkeypatch):
|
||
"""При одинаковой категории меньшее время в пути → больший вес."""
|
||
rows = [
|
||
_make_routing_row("Школа дальняя", "school", 900.0),
|
||
_make_routing_row("Школа ближняя", "school", 200.0),
|
||
]
|
||
db = _MockDb(rows)
|
||
|
||
def fake_matrix(_lon, _lat, dests, **_kw):
|
||
return [12.0, 3.0] # дальняя 12 мин, ближняя 3 мин
|
||
|
||
monkeypatch.setattr(ors_client, "matrix_durations_min", fake_matrix)
|
||
monkeypatch.setattr(ors_client, "is_configured", lambda: True)
|
||
|
||
result = compute_poi_routing_decay(db, "cad", 56.838, 60.605)
|
||
assert result.top_poi[0].name == "Школа ближняя"
|
||
|
||
|
||
def test_routing_decay_falls_back_to_straight_line(monkeypatch):
|
||
"""ORS недоступен → straight-line fallback, без исключения наружу."""
|
||
rows = [_make_routing_row("Метро", "metro_stop", 300.0)]
|
||
db = _MockDb(rows)
|
||
|
||
def raising_matrix(*_a, **_kw):
|
||
raise ors_client.OrsUnavailableError("no key")
|
||
|
||
monkeypatch.setattr(ors_client, "matrix_durations_min", raising_matrix)
|
||
|
||
result = compute_poi_routing_decay(db, "cad", 56.838, 60.605)
|
||
# 300м / 83 м/мин ≈ 3.6 мин — в пределах 20 мин радиуса метро → остаётся.
|
||
assert len(result.top_poi) == 1
|
||
assert result.top_poi[0].category == "metro_stop"
|
||
|
||
|
||
def test_routing_decay_none_route_fallback(monkeypatch):
|
||
"""ORS вернул None (нет маршрута) для точки → straight-line оценка времени."""
|
||
rows = [_make_routing_row("Парк", "park", 400.0)]
|
||
db = _MockDb(rows)
|
||
|
||
def fake_matrix(_lon, _lat, dests, **_kw):
|
||
return [None] # ORS не построил маршрут
|
||
|
||
monkeypatch.setattr(ors_client, "matrix_durations_min", fake_matrix)
|
||
monkeypatch.setattr(ors_client, "is_configured", lambda: True)
|
||
|
||
result = compute_poi_routing_decay(db, "cad", 56.838, 60.605)
|
||
# 400м / 83 ≈ 4.8 мин < 10 мин (park step) → парк остаётся.
|
||
assert len(result.top_poi) == 1
|
||
|
||
|
||
def test_routing_decay_empty_db(monkeypatch):
|
||
db = _MockDb([])
|
||
monkeypatch.setattr(ors_client, "is_configured", lambda: True)
|
||
result = compute_poi_routing_decay(db, "cad", 56.838, 60.605)
|
||
assert result.top_poi == []
|
||
|
||
|
||
def test_routing_decay_score_spread_wider_than_straight_line(monkeypatch):
|
||
"""#41 acceptance: routing-decay даёт более широкий разброс между хорошим и
|
||
плохим участком, чем straight-line (1/(d+100)).
|
||
|
||
Хороший участок: метро 4 мин, школа 6 мин. Плохой: метро 18 мин, школа 14 мин.
|
||
"""
|
||
|
||
def total(rows, times):
|
||
db = _MockDb(rows)
|
||
monkeypatch.setattr(ors_client, "is_configured", lambda: True)
|
||
monkeypatch.setattr(
|
||
ors_client, "matrix_durations_min", lambda *_a, **_k: list(times)
|
||
)
|
||
res = compute_poi_routing_decay(db, "cad", 56.8, 60.6)
|
||
return sum(i.weight for i in res.top_poi)
|
||
|
||
good_rows = [
|
||
_make_routing_row("Метро", "metro_stop", 300.0),
|
||
_make_routing_row("Школа", "school", 450.0),
|
||
]
|
||
bad_rows = [
|
||
_make_routing_row("Метро", "metro_stop", 1400.0),
|
||
_make_routing_row("Школа", "school", 1100.0),
|
||
]
|
||
good = total(good_rows, [4.0, 6.0])
|
||
bad = total(bad_rows, [18.0, 14.0])
|
||
|
||
assert good > 0 and bad > 0
|
||
# routing-decay усиливает контраст: хороший минимум вдвое выше плохого.
|
||
assert good / bad > 2.0, f"ожидали разброс >2×, получили {good / bad:.2f}×"
|