634 lines
26 KiB
Python
634 lines
26 KiB
Python
"""Тесты воркера okn_objects_sync (#1066, #1159) — без живой сети.
|
||
|
||
Покрывает:
|
||
- _extract_coords: правильный порядок [lat, lon] → (lon, lat) для PostGIS.
|
||
- _extract_coords: невалидные / выходящие за bbox координаты → None.
|
||
- _PASSES: расширенный список адресных подстрок + декартово с категориями.
|
||
- sync_okn_objects: дедуп source_id across проходов (federal wins).
|
||
- sync_okn_objects: coverage — фичи из district-passes добавляются к city-pass.
|
||
- sync_okn_objects: upsert идемпотентность.
|
||
- _enrich_with_detail: merge name/address в raw_props; пустой detail → noop.
|
||
- _enrich_with_detail: detail-фейл (raise) не валит прогон.
|
||
- _enrich_with_detail: limit ограничивает число fetch'ей.
|
||
- _enrich_with_detail: circuit breaker — N детейл-фейлов подряд прерывает цикл,
|
||
сбрасывается успехом, дефолт берётся из _DETAIL_FAIL_BREAKER (#2445 item D3).
|
||
- parcel_okn_objects: name/address возвращаются из БД-строки.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from typing import Any
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
from app.workers.tasks.okn_objects_sync import (
|
||
_ADDRESS_PASSES,
|
||
_CATEGORY_ORDER,
|
||
_PASSES,
|
||
_enrich_with_detail,
|
||
_extract_coords,
|
||
)
|
||
|
||
# ── _extract_coords ───────────────────────────────────────────────────────────
|
||
|
||
|
||
def _make_feature_coords(lat: float, lon: float) -> dict[str, Any]:
|
||
"""Фича ЕГРКН с coordinates=[lat, lon] (Яндекс-порядок)."""
|
||
return {
|
||
"id": "test",
|
||
"geometry": {"type": "Point", "coordinates": [lat, lon]},
|
||
"properties": {},
|
||
}
|
||
|
||
|
||
def test_extract_coords_ekb_center() -> None:
|
||
"""Центр ЕКБ (56.83N 60.61E): lat=coords[0], lon=coords[1] → (lon, lat) для MakePoint."""
|
||
feature = _make_feature_coords(lat=56.83, lon=60.61)
|
||
result = _extract_coords(feature)
|
||
assert result is not None
|
||
lon, lat = result
|
||
# MakePoint(lon, lat): lon=60.61°E (Свердл. обл.), lat=56.83°N (ЕКБ)
|
||
assert abs(lon - 60.61) < 0.001, f"lon={lon} ≠ 60.61 — coord swap сломан"
|
||
assert abs(lat - 56.83) < 0.001, f"lat={lat} ≠ 56.83 — coord swap сломан"
|
||
|
||
|
||
def test_extract_coords_inside_ekb_bbox() -> None:
|
||
"""Любая точка в ЕКБ-bbox [55–58.5°N, 58–63°E] проходит санитарный фильтр."""
|
||
feature = _make_feature_coords(lat=56.50, lon=60.10)
|
||
result = _extract_coords(feature)
|
||
assert result is not None
|
||
lon, lat = result
|
||
assert 58.0 <= lon <= 63.0
|
||
assert 55.0 <= lat <= 58.5
|
||
|
||
|
||
def test_extract_coords_outside_bbox_returns_none() -> None:
|
||
"""Координаты вне Урал-bbox → None (некорректная/тестовая точка из ЕГРКН)."""
|
||
# Москва (~55.75N 37.62E) — lon=37.62 вне [58–63]
|
||
feature = _make_feature_coords(lat=55.75, lon=37.62)
|
||
result = _extract_coords(feature)
|
||
assert result is None
|
||
|
||
|
||
def test_extract_coords_inverted_would_fail_bbox() -> None:
|
||
"""Тест критически важной coord swap логики.
|
||
|
||
Если бы coords[0] трактовался как lon (а не lat),
|
||
то для ЕКБ-точки (lat=56.83, lon=60.61):
|
||
wrong_lon = 56.83 → вне bbox [58–63]
|
||
→ _extract_coords вернул бы None → баг обнаружен.
|
||
|
||
Этот тест ловит регрессию если кто-то случайно поменяет местами coords[0]/[1].
|
||
"""
|
||
# Проверяем что ПРАВИЛЬНАЯ точка ЕКБ проходит только при lat=coords[0], lon=coords[1]
|
||
feature = _make_feature_coords(lat=56.83, lon=60.61)
|
||
# Симулируем неверную трактовку: coords[0]=56.83 как lon
|
||
inverted_lon = feature["geometry"]["coordinates"][0] # 56.83 → не в bbox [58–63]
|
||
inverted_lat = feature["geometry"]["coordinates"][1] # 60.61 → не в bbox [55–58.5]
|
||
inverted_inside = (55.0 <= inverted_lat <= 58.5) and (58.0 <= inverted_lon <= 63.0)
|
||
assert (
|
||
not inverted_inside
|
||
), "Перепутанные координаты прошли bbox-фильтр — _extract_coords не обнаружит баг coord swap"
|
||
|
||
# Проверяем что реальный _extract_coords возвращает КОРРЕКТНЫЕ (lon, lat)
|
||
result = _extract_coords(feature)
|
||
assert result is not None
|
||
lon, _lat = result
|
||
assert abs(lon - 60.61) < 0.001
|
||
|
||
|
||
def test_extract_coords_missing_geometry_returns_none() -> None:
|
||
"""Фича без geometry → None (graceful)."""
|
||
feature: dict[str, Any] = {"id": "x", "geometry": None, "properties": {}}
|
||
assert _extract_coords(feature) is None
|
||
|
||
|
||
def test_extract_coords_missing_coordinates_returns_none() -> None:
|
||
"""Geometry без coordinates → None."""
|
||
feature: dict[str, Any] = {
|
||
"id": "x",
|
||
"geometry": {"type": "Point"},
|
||
"properties": {},
|
||
}
|
||
assert _extract_coords(feature) is None
|
||
|
||
|
||
def test_extract_coords_single_coordinate_returns_none() -> None:
|
||
"""Только одна координата (len<2) → None."""
|
||
feature: dict[str, Any] = {
|
||
"id": "x",
|
||
"geometry": {"type": "Point", "coordinates": [56.83]},
|
||
"properties": {},
|
||
}
|
||
assert _extract_coords(feature) is None
|
||
|
||
|
||
# ── _PASSES / _ADDRESS_PASSES структура (#1159 coverage) ──────────────────────
|
||
|
||
|
||
def test_address_passes_include_ekb_districts() -> None:
|
||
"""Поверх «Екатеринбург»/«Свердловская» добавлены районы ЕКБ.
|
||
|
||
Defect 1 (#1159): searchMap не возвращает объекты с адресом без слова
|
||
«Екатеринбург». Запросы по районам ловят такие записи.
|
||
"""
|
||
expected_districts = {
|
||
"Кировский",
|
||
"Ленинский",
|
||
"Октябрьский",
|
||
"Орджоникидзевский",
|
||
"Чкаловский",
|
||
"Верх-Исетский",
|
||
"Железнодорожный",
|
||
}
|
||
assert expected_districts <= set(
|
||
_ADDRESS_PASSES
|
||
), f"Не все районы ЕКБ в _ADDRESS_PASSES: missing={expected_districts - set(_ADDRESS_PASSES)}"
|
||
|
||
|
||
def test_address_passes_order_city_before_district() -> None:
|
||
"""Городские запросы идут раньше районных.
|
||
|
||
При дедупе source_id first-wins: если объект всплыл в проходе «Екатеринбург»
|
||
как federal, проход «Кировский» не должен перетереть его category.
|
||
"""
|
||
city_idx = _ADDRESS_PASSES.index("Екатеринбург")
|
||
district_idx = _ADDRESS_PASSES.index("Кировский")
|
||
assert city_idx < district_idx
|
||
|
||
|
||
def test_passes_cartesian_product_address_x_category() -> None:
|
||
"""_PASSES = декартово (address × category_order)."""
|
||
expected_size = len(_ADDRESS_PASSES) * len(_CATEGORY_ORDER)
|
||
assert len(_PASSES) == expected_size
|
||
# Каждый адрес покрыт всеми тремя категориями (federal/regional/local).
|
||
for addr in _ADDRESS_PASSES:
|
||
cats = [cat for a, cat in _PASSES if a == addr]
|
||
assert sorted(cats) == sorted(_CATEGORY_ORDER), f"{addr}: {cats}"
|
||
|
||
|
||
def test_passes_federal_before_local_per_address() -> None:
|
||
"""Для каждого address federal появляется раньше regional раньше local.
|
||
|
||
Это гарантирует что при конфликте source_id в одной address-страте
|
||
более «высокая» категория выигрывает.
|
||
"""
|
||
for addr in _ADDRESS_PASSES:
|
||
positions = [i for i, (a, _c) in enumerate(_PASSES) if a == addr]
|
||
cats_for_addr = [_PASSES[i][1] for i in positions]
|
||
assert cats_for_addr == _CATEGORY_ORDER, f"{addr}: {cats_for_addr}"
|
||
|
||
|
||
# ── sync_okn_objects: дедуп source_id ────────────────────────────────────────
|
||
|
||
|
||
def _make_feature_full(
|
||
feature_id: str,
|
||
lat: float = 56.83,
|
||
lon: float = 60.61,
|
||
) -> dict[str, Any]:
|
||
return {
|
||
"id": feature_id,
|
||
"geometry": {"type": "Point", "coordinates": [lat, lon]},
|
||
"properties": {"id": feature_id},
|
||
}
|
||
|
||
|
||
def _build_mock_db(captured: list[dict[str, Any]]) -> MagicMock:
|
||
mock_db = MagicMock()
|
||
mock_db.__enter__ = MagicMock(return_value=mock_db)
|
||
mock_db.__exit__ = MagicMock(return_value=False)
|
||
mock_db.begin_nested.return_value.__enter__ = MagicMock(return_value=None)
|
||
mock_db.begin_nested.return_value.__exit__ = MagicMock(return_value=False)
|
||
|
||
def _exec(sql: Any, row: Any = None) -> MagicMock:
|
||
if row:
|
||
captured.append(dict(row))
|
||
return MagicMock()
|
||
|
||
mock_db.execute = _exec
|
||
return mock_db
|
||
|
||
|
||
def test_sync_okn_objects_dedup_source_id_across_passes() -> None:
|
||
"""Один source_id из нескольких проходов → category из первого прохода (federal wins)."""
|
||
from app.workers.tasks.okn_objects_sync import sync_okn_objects
|
||
|
||
# «001» встречается в проходе (Екатеринбург, 1) = federal
|
||
# и в проходе (Екатеринбург, 2) = regional → category должна быть federal.
|
||
feat = _make_feature_full("001")
|
||
|
||
def _mock_fetch(address: str, category_type: int) -> list[dict[str, Any]]:
|
||
if address == "Екатеринбург" and category_type in (1, 2):
|
||
return [feat]
|
||
return []
|
||
|
||
upserted_rows: list[dict[str, Any]] = []
|
||
mock_db = _build_mock_db(upserted_rows)
|
||
|
||
with (
|
||
patch("app.workers.tasks.okn_objects_sync.fetch_okn_points", side_effect=_mock_fetch),
|
||
patch(
|
||
"app.workers.tasks.okn_objects_sync.fetch_okn_detail",
|
||
return_value={},
|
||
),
|
||
patch("app.workers.tasks.okn_objects_sync.SessionLocal", return_value=mock_db),
|
||
):
|
||
result = sync_okn_objects()
|
||
|
||
# Один объект, один upsert
|
||
assert result == {"okn": 1, "enriched": 0}
|
||
assert len(upserted_rows) == 1
|
||
assert upserted_rows[0]["source_id"] == "001"
|
||
assert upserted_rows[0]["category"] == "federal"
|
||
|
||
|
||
def test_sync_okn_objects_coverage_collects_from_multiple_address_passes() -> None:
|
||
"""Фичи, появляющиеся только в district-проходе, добавляются к city-проходу (#1159).
|
||
|
||
Симулируем реальный кейс: 1 объект ловится «Екатеринбург», 1 — только
|
||
«Кировский» (улицный адрес без слова Екатеринбург), 1 — только
|
||
«Свердловская» (за городом). После прогона все три попадают в upsert.
|
||
"""
|
||
from app.workers.tasks.okn_objects_sync import sync_okn_objects
|
||
|
||
city_feat = _make_feature_full("city-001")
|
||
district_feat = _make_feature_full("district-002")
|
||
region_feat = _make_feature_full("region-003")
|
||
|
||
def _mock_fetch(address: str, category_type: int) -> list[dict[str, Any]]:
|
||
if category_type != 2: # для краткости — все «региональные»
|
||
return []
|
||
if address == "Екатеринбург":
|
||
return [city_feat]
|
||
if address == "Кировский":
|
||
return [city_feat, district_feat] # дубль city-001 + новый
|
||
if address == "Свердловская":
|
||
return [region_feat]
|
||
return []
|
||
|
||
upserted_rows: list[dict[str, Any]] = []
|
||
mock_db = _build_mock_db(upserted_rows)
|
||
|
||
with (
|
||
patch("app.workers.tasks.okn_objects_sync.fetch_okn_points", side_effect=_mock_fetch),
|
||
patch(
|
||
"app.workers.tasks.okn_objects_sync.fetch_okn_detail",
|
||
return_value={},
|
||
),
|
||
patch("app.workers.tasks.okn_objects_sync.SessionLocal", return_value=mock_db),
|
||
):
|
||
result = sync_okn_objects()
|
||
|
||
source_ids = {r["source_id"] for r in upserted_rows}
|
||
assert source_ids == {"city-001", "district-002", "region-003"}, source_ids
|
||
assert result["okn"] == 3
|
||
|
||
|
||
def test_sync_okn_objects_upsert_idempotent() -> None:
|
||
"""Два вызова с одними source_id → два upsert (ON CONFLICT в БД, не Python-фильтр)."""
|
||
from app.workers.tasks.okn_objects_sync import sync_okn_objects
|
||
|
||
features = [_make_feature_full("001"), _make_feature_full("002")]
|
||
|
||
def _mock_fetch(address: str, category_type: int) -> list[dict[str, Any]]:
|
||
if address == "Екатеринбург" and category_type == 1:
|
||
return features
|
||
return []
|
||
|
||
counts: list[int] = []
|
||
|
||
def _run() -> dict[str, int]:
|
||
upserted: list[dict[str, Any]] = []
|
||
mock_db = _build_mock_db(upserted)
|
||
with (
|
||
patch(
|
||
"app.workers.tasks.okn_objects_sync.fetch_okn_points",
|
||
side_effect=_mock_fetch,
|
||
),
|
||
patch(
|
||
"app.workers.tasks.okn_objects_sync.fetch_okn_detail",
|
||
return_value={},
|
||
),
|
||
patch("app.workers.tasks.okn_objects_sync.SessionLocal", return_value=mock_db),
|
||
):
|
||
res = sync_okn_objects()
|
||
counts.append(len(upserted))
|
||
return res
|
||
|
||
r1 = _run()
|
||
r2 = _run()
|
||
assert r1["okn"] == 2
|
||
assert r2["okn"] == 2
|
||
assert counts == [2, 2]
|
||
|
||
|
||
def test_sync_okn_objects_returns_total_unique_count() -> None:
|
||
"""Возвращает {"okn": N, "enriched": M}: N уникальных, M обогащённых."""
|
||
from app.workers.tasks.okn_objects_sync import sync_okn_objects
|
||
|
||
feats_1 = [_make_feature_full("A"), _make_feature_full("B")]
|
||
feats_2 = [_make_feature_full("C"), _make_feature_full("A")] # A дубль
|
||
|
||
def _mock_fetch(address: str, category_type: int) -> list[dict[str, Any]]:
|
||
if address == "Екатеринбург" and category_type == 1:
|
||
return feats_1
|
||
if address == "Екатеринбург" and category_type == 2:
|
||
return feats_2
|
||
return []
|
||
|
||
upserted_rows: list[dict[str, Any]] = []
|
||
mock_db = _build_mock_db(upserted_rows)
|
||
|
||
with (
|
||
patch(
|
||
"app.workers.tasks.okn_objects_sync.fetch_okn_points",
|
||
side_effect=_mock_fetch,
|
||
),
|
||
patch(
|
||
"app.workers.tasks.okn_objects_sync.fetch_okn_detail",
|
||
return_value={},
|
||
),
|
||
patch("app.workers.tasks.okn_objects_sync.SessionLocal", return_value=mock_db),
|
||
):
|
||
result = sync_okn_objects()
|
||
|
||
assert result["okn"] == 3
|
||
assert result["enriched"] == 0
|
||
|
||
|
||
# ── _enrich_with_detail (#1159 acceptance 2) ─────────────────────────────────
|
||
|
||
|
||
def _row(source_id: str, base_props: dict[str, Any] | None = None) -> dict[str, Any]:
|
||
return {
|
||
"source_id": source_id,
|
||
"category": "regional",
|
||
"lon": 60.61,
|
||
"lat": 56.83,
|
||
"raw_props": json.dumps(base_props or {"id": source_id}, ensure_ascii=False),
|
||
}
|
||
|
||
|
||
def test_enrich_merges_name_address_into_raw_props() -> None:
|
||
"""detail (name, address, …) подмёрджен в raw_props; baseline id сохранён."""
|
||
rows = [_row("001"), _row("002")]
|
||
details = {
|
||
"001": {"name": "Дом Севастьянова", "address": "пр. Ленина, 35"},
|
||
"002": {"name": "Гостиница «Большой Урал»", "address": "ул. Красноармейская, 1"},
|
||
}
|
||
|
||
enriched = _enrich_with_detail(rows, detail_fetcher=lambda sid: details[sid])
|
||
assert enriched == 2
|
||
|
||
parsed_0 = json.loads(rows[0]["raw_props"])
|
||
assert parsed_0 == {"id": "001", "name": "Дом Севастьянова", "address": "пр. Ленина, 35"}
|
||
parsed_1 = json.loads(rows[1]["raw_props"])
|
||
assert parsed_1["address"] == "ул. Красноармейская, 1"
|
||
assert parsed_1["id"] == "002"
|
||
|
||
|
||
def test_enrich_with_empty_detail_keeps_baseline() -> None:
|
||
"""Пустой detail → raw_props не меняется, enriched=0."""
|
||
rows = [_row("001")]
|
||
original = rows[0]["raw_props"]
|
||
|
||
enriched = _enrich_with_detail(rows, detail_fetcher=lambda _sid: {})
|
||
assert enriched == 0
|
||
assert rows[0]["raw_props"] == original
|
||
|
||
|
||
def test_enrich_tolerates_detail_exception() -> None:
|
||
"""Detail-fetcher raises → строка пропускается, прогон не падает."""
|
||
rows = [_row("001"), _row("002")]
|
||
|
||
def _fetcher(source_id: str) -> dict[str, Any]:
|
||
if source_id == "001":
|
||
raise RuntimeError("network down")
|
||
return {"name": "OK", "address": "Х"}
|
||
|
||
enriched = _enrich_with_detail(rows, detail_fetcher=_fetcher)
|
||
assert enriched == 1
|
||
# 001 baseline, 002 enriched
|
||
parsed_1 = json.loads(rows[1]["raw_props"])
|
||
assert parsed_1["name"] == "OK"
|
||
|
||
|
||
# ── circuit breaker: N detail-фейлов подряд → прерывание (#2445 item D3) ────
|
||
|
||
|
||
def test_enrich_circuit_breaker_stops_after_consecutive_failures() -> None:
|
||
"""fail_breaker детейл-фейлов ПОДРЯД → цикл прерывается, остаток не запрашивается.
|
||
|
||
Аналогично gisogd66._CARD_FAIL_BREAKER: устойчивый WAF-бан не должен приводить
|
||
к тому, что все оставшиеся до _DETAIL_FETCH_LIMIT объекты долбятся впустую.
|
||
"""
|
||
rows = [_row(f"{i:03d}") for i in range(20)]
|
||
calls: list[str] = []
|
||
|
||
def _fetcher(source_id: str) -> dict[str, Any]:
|
||
calls.append(source_id)
|
||
raise RuntimeError("WAF ban simulated")
|
||
|
||
enriched = _enrich_with_detail(rows, detail_fetcher=_fetcher, fail_breaker=5)
|
||
|
||
assert enriched == 0
|
||
# Breaker останавливает цикл ровно после fail_breaker вызовов подряд — остаток
|
||
# (15 строк) НЕ запрашивается.
|
||
assert len(calls) == 5
|
||
|
||
|
||
def test_enrich_circuit_breaker_resets_on_success() -> None:
|
||
"""Успешный вызов (даже без detail-полей) сбрасывает счётчик подряд-фейлов.
|
||
|
||
Чередование fail/success/fail/... никогда не должно достичь fail_breaker,
|
||
если между фейлами есть успехи.
|
||
"""
|
||
rows = [_row(f"{i:03d}") for i in range(12)]
|
||
calls: list[str] = []
|
||
|
||
def _fetcher(source_id: str) -> dict[str, Any]:
|
||
calls.append(source_id)
|
||
idx = int(source_id)
|
||
if idx % 2 == 0:
|
||
raise RuntimeError("transient")
|
||
return {"name": f"N-{source_id}"}
|
||
|
||
enriched = _enrich_with_detail(rows, detail_fetcher=_fetcher, fail_breaker=3)
|
||
|
||
# Ни одного случая 3 фейлов подряд — breaker не сработал, все 12 строк обработаны.
|
||
assert len(calls) == 12
|
||
assert enriched == 6
|
||
|
||
|
||
def test_enrich_circuit_breaker_disabled_via_high_threshold() -> None:
|
||
"""fail_breaker выше числа строк → breaker не мешает нормальному прогону с фейлами."""
|
||
rows = [_row(f"{i:03d}") for i in range(5)]
|
||
|
||
def _fetcher(source_id: str) -> dict[str, Any]:
|
||
raise RuntimeError("always fails")
|
||
|
||
enriched = _enrich_with_detail(rows, detail_fetcher=_fetcher, fail_breaker=1000)
|
||
assert enriched == 0
|
||
# Без early-stop все 5 строк должны были быть попытаны — raw_props не изменился.
|
||
for row in rows:
|
||
parsed = json.loads(row["raw_props"])
|
||
assert "name" not in parsed
|
||
|
||
|
||
def test_enrich_uses_default_fail_breaker_constant() -> None:
|
||
"""Дефолт fail_breaker берётся из _DETAIL_FAIL_BREAKER (модульная константа)."""
|
||
from app.workers.tasks.okn_objects_sync import _DETAIL_FAIL_BREAKER
|
||
|
||
rows = [_row(f"{i:03d}") for i in range(_DETAIL_FAIL_BREAKER + 10)]
|
||
calls: list[str] = []
|
||
|
||
def _fetcher(source_id: str) -> dict[str, Any]:
|
||
calls.append(source_id)
|
||
raise RuntimeError("persistent failure")
|
||
|
||
enriched = _enrich_with_detail(rows, detail_fetcher=_fetcher)
|
||
assert enriched == 0
|
||
assert len(calls) == _DETAIL_FAIL_BREAKER
|
||
|
||
|
||
def test_enrich_respects_limit() -> None:
|
||
"""limit=N → только первые N source_id запрошены."""
|
||
rows = [_row(f"{i:03d}") for i in range(10)]
|
||
calls: list[str] = []
|
||
|
||
def _fetcher(source_id: str) -> dict[str, Any]:
|
||
calls.append(source_id)
|
||
return {"name": f"N-{source_id}"}
|
||
|
||
enriched = _enrich_with_detail(rows, detail_fetcher=_fetcher, limit=3)
|
||
assert enriched == 3
|
||
assert len(calls) == 3
|
||
assert calls == ["000", "001", "002"]
|
||
|
||
|
||
def test_enrich_handles_non_dict_base_props() -> None:
|
||
"""raw_props с битым JSON (не dict) → пересоздаётся как пустой dict + merge detail."""
|
||
row = {
|
||
"source_id": "001",
|
||
"category": "regional",
|
||
"lon": 60.61,
|
||
"lat": 56.83,
|
||
"raw_props": json.dumps(["broken", "list"]), # не dict
|
||
}
|
||
enriched = _enrich_with_detail([row], detail_fetcher=lambda _sid: {"name": "X", "address": "Y"})
|
||
assert enriched == 1
|
||
parsed = json.loads(row["raw_props"])
|
||
assert parsed == {"name": "X", "address": "Y"}
|
||
|
||
|
||
# ── parcel_okn_objects lookup (#1159 acceptance 2: name/address) ─────────────
|
||
|
||
|
||
def test_parcel_okn_objects_returns_nearby_with_name_address() -> None:
|
||
"""parcel_okn_objects: мок DB → [{source_id, category, distance_m, name, address}]."""
|
||
from app.services.site_finder.okn_lookup import parcel_okn_objects
|
||
|
||
rows = [
|
||
("ekb-001", "federal", 42, "Дом Севастьянова", "пр. Ленина, 35"),
|
||
("ekb-002", "regional", 95, None, None), # detail не догружен
|
||
]
|
||
|
||
mock_result = MagicMock()
|
||
mock_result.fetchall.return_value = rows
|
||
|
||
db = MagicMock()
|
||
db.execute.return_value = mock_result
|
||
|
||
result = parcel_okn_objects(db, "POLYGON((60 56,61 56,61 57,60 57,60 56))", radius_m=100)
|
||
|
||
assert len(result) == 2
|
||
assert result[0] == {
|
||
"source_id": "ekb-001",
|
||
"category": "federal",
|
||
"distance_m": 42,
|
||
"name": "Дом Севастьянова",
|
||
"address": "пр. Ленина, 35",
|
||
}
|
||
assert result[1]["name"] is None
|
||
assert result[1]["address"] is None
|
||
assert result[1]["source_id"] == "ekb-002"
|
||
|
||
|
||
def test_parcel_okn_objects_empty_wkt_returns_empty() -> None:
|
||
"""Пустой parcel_wkt → пустой список без DB-запроса."""
|
||
from app.services.site_finder.okn_lookup import parcel_okn_objects
|
||
|
||
db = MagicMock()
|
||
result = parcel_okn_objects(db, None)
|
||
assert result == []
|
||
db.execute.assert_not_called()
|
||
|
||
|
||
def test_parcel_okn_objects_empty_string_wkt_returns_empty() -> None:
|
||
"""Пустая строка parcel_wkt → пустой список без DB-запроса."""
|
||
from app.services.site_finder.okn_lookup import parcel_okn_objects
|
||
|
||
db = MagicMock()
|
||
result = parcel_okn_objects(db, "")
|
||
assert result == []
|
||
db.execute.assert_not_called()
|
||
|
||
|
||
def test_parcel_okn_objects_no_table_returns_empty() -> None:
|
||
"""ProgrammingError (таблица не создана) → пустой список (graceful)."""
|
||
from sqlalchemy.exc import ProgrammingError
|
||
|
||
from app.services.site_finder.okn_lookup import parcel_okn_objects
|
||
|
||
db = MagicMock()
|
||
db.execute.side_effect = ProgrammingError("relation does not exist", None, None)
|
||
|
||
result = parcel_okn_objects(db, "POLYGON((60 56,61 56,61 57,60 57,60 56))")
|
||
assert result == []
|
||
|
||
|
||
def test_parcel_okn_objects_db_error_returns_empty() -> None:
|
||
"""OperationalError → пустой список (graceful)."""
|
||
from sqlalchemy.exc import OperationalError
|
||
|
||
from app.services.site_finder.okn_lookup import parcel_okn_objects
|
||
|
||
db = MagicMock()
|
||
db.execute.side_effect = OperationalError("connection lost", None, None)
|
||
|
||
result = parcel_okn_objects(db, "POLYGON((60 56,61 56,61 57,60 57,60 56))")
|
||
assert result == []
|
||
|
||
|
||
def test_parcel_okn_objects_no_results_returns_empty() -> None:
|
||
"""Нет объектов в радиусе → пустой список."""
|
||
from app.services.site_finder.okn_lookup import parcel_okn_objects
|
||
|
||
mock_result = MagicMock()
|
||
mock_result.fetchall.return_value = []
|
||
|
||
db = MagicMock()
|
||
db.execute.return_value = mock_result
|
||
|
||
result = parcel_okn_objects(db, "POLYGON((60 56,61 56,61 57,60 57,60 56))", radius_m=50)
|
||
assert result == []
|
||
|
||
|
||
def test_parcel_okn_objects_distance_mapping() -> None:
|
||
"""distance_m / name / address берутся из колонок [2..4] (порядок SELECT)."""
|
||
from app.services.site_finder.okn_lookup import parcel_okn_objects
|
||
|
||
rows = [("id-x", "local", 7, "Памятник Х", "ул. Y")]
|
||
mock_result = MagicMock()
|
||
mock_result.fetchall.return_value = rows
|
||
|
||
db = MagicMock()
|
||
db.execute.return_value = mock_result
|
||
|
||
result = parcel_okn_objects(db, "POINT(60.61 56.83)", radius_m=200)
|
||
assert result[0]["distance_m"] == 7
|
||
assert result[0]["source_id"] == "id-x"
|
||
assert result[0]["category"] == "local"
|
||
assert result[0]["name"] == "Памятник Х"
|
||
assert result[0]["address"] == "ул. Y"
|