473 lines
17 KiB
Python
473 lines
17 KiB
Python
"""Unit tests for cadastral FDW SQL-backed geocoder helpers and reverse_geocode.
|
|
|
|
Tests cover:
|
|
- _cadastral_forward_sync: returns suggestions / empty list / graceful error
|
|
- _cadastral_reverse_sync: returns address / None on no result / None on error
|
|
- reverse_geocode: cadastral primary, Nominatim fallback, both-fail → None (500-fix)
|
|
- geocode: cadastral as tier 0
|
|
- suggest: cadastral as tier 0
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
# DATABASE_URL required by config before any app import.
|
|
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
|
|
|
# WeasyPrint stub — not installed in CI without GTK.
|
|
_wp_mock = MagicMock()
|
|
sys.modules.setdefault("weasyprint", _wp_mock)
|
|
|
|
|
|
from app.services.geocoder import ( # noqa: E402
|
|
GeocodeSuggestion,
|
|
ReverseGeocodeResult,
|
|
_cadastral_forward_sync,
|
|
_cadastral_reverse_sync,
|
|
geocode,
|
|
reverse_geocode,
|
|
suggest,
|
|
)
|
|
|
|
# ── helpers ───────────────────────────────────────────────────────────────────
|
|
|
|
def _make_db_row(**kwargs: object) -> MagicMock:
|
|
"""Mock row object with dict-like access (MappingResult style)."""
|
|
row = MagicMock()
|
|
row.__getitem__ = lambda self, k: kwargs[k]
|
|
for k, v in kwargs.items():
|
|
setattr(row, k, v)
|
|
return row
|
|
|
|
|
|
def _make_db(rows: list[object]) -> MagicMock:
|
|
"""Session mock whose execute().mappings().all() returns rows."""
|
|
db = MagicMock()
|
|
result = MagicMock()
|
|
result.mappings.return_value.all.return_value = rows
|
|
result.first.return_value = rows[0] if rows else None
|
|
db.execute.return_value = result
|
|
return db
|
|
|
|
|
|
def _make_db_error() -> MagicMock:
|
|
"""Session mock whose execute() raises RuntimeError."""
|
|
db = MagicMock()
|
|
db.execute.side_effect = RuntimeError("FDW connection failed")
|
|
return db
|
|
|
|
|
|
# ── _cadastral_forward_sync ───────────────────────────────────────────────────
|
|
|
|
def test_forward_sync_returns_suggestions_when_rows_exist() -> None:
|
|
"""DB returns rows → list of GeocodeSuggestion returned."""
|
|
row = _make_db_row(
|
|
readable_address="ул. Малышева, 30, Екатеринбург",
|
|
lat=56.838,
|
|
lon=60.605,
|
|
sim=0.9,
|
|
cad_num="66:41:0101001:123",
|
|
)
|
|
db = _make_db([row])
|
|
|
|
results = _cadastral_forward_sync(db, "Малышева 30", limit=5)
|
|
|
|
assert len(results) == 1
|
|
assert results[0].full_address == "ул. Малышева, 30, Екатеринбург"
|
|
assert results[0].lat == 56.838
|
|
assert results[0].lon == 60.605
|
|
assert results[0].kind == "house"
|
|
|
|
|
|
def test_forward_sync_returns_empty_when_no_rows() -> None:
|
|
"""DB returns no rows → empty list returned, no exception."""
|
|
db = _make_db([])
|
|
|
|
results = _cadastral_forward_sync(db, "Несуществующая улица", limit=5)
|
|
|
|
assert results == []
|
|
|
|
|
|
def test_forward_sync_returns_empty_on_db_error() -> None:
|
|
"""DB raises → empty list, no exception bubbles out."""
|
|
db = _make_db_error()
|
|
|
|
results = _cadastral_forward_sync(db, "Малышева", limit=5)
|
|
|
|
assert results == []
|
|
|
|
|
|
def test_forward_sync_respects_limit() -> None:
|
|
"""Returns at most `limit` items even if DB has more."""
|
|
rows = [
|
|
_make_db_row(
|
|
readable_address=f"ул. Тестовая, {i}, Екатеринбург",
|
|
lat=56.838 + i * 0.001,
|
|
lon=60.605,
|
|
sim=0.9,
|
|
cad_num=f"66:41:01010{i:02d}:1",
|
|
)
|
|
for i in range(10)
|
|
]
|
|
db = _make_db(rows)
|
|
|
|
results = _cadastral_forward_sync(db, "Тестовая", limit=3)
|
|
|
|
assert len(results) <= 3
|
|
|
|
|
|
# ── _cadastral_reverse_sync ───────────────────────────────────────────────────
|
|
|
|
def test_reverse_sync_returns_address_when_row_found() -> None:
|
|
"""DB finds building within radius → address string returned."""
|
|
row = MagicMock()
|
|
row.readable_address = "ул. Хохрякова, 81, Екатеринбург"
|
|
|
|
db = MagicMock()
|
|
result = MagicMock()
|
|
result.first.return_value = row
|
|
db.execute.return_value = result
|
|
|
|
addr = _cadastral_reverse_sync(db, 56.838, 60.605, radius_m=200)
|
|
|
|
assert addr == "ул. Хохрякова, 81, Екатеринбург"
|
|
|
|
|
|
def test_reverse_sync_returns_none_when_no_row_in_radius() -> None:
|
|
"""No row within radius → None returned."""
|
|
db = MagicMock()
|
|
result = MagicMock()
|
|
result.first.return_value = None
|
|
db.execute.return_value = result
|
|
|
|
addr = _cadastral_reverse_sync(db, 56.838, 60.605, radius_m=200)
|
|
|
|
assert addr is None
|
|
|
|
|
|
def test_reverse_sync_returns_none_on_db_error() -> None:
|
|
"""DB raises → None returned, no exception bubbles out."""
|
|
db = _make_db_error()
|
|
|
|
addr = _cadastral_reverse_sync(db, 56.838, 60.605)
|
|
|
|
assert addr is None
|
|
|
|
|
|
# ── reverse_geocode ───────────────────────────────────────────────────────────
|
|
|
|
def _nom_result(address: str) -> ReverseGeocodeResult:
|
|
return ReverseGeocodeResult(
|
|
address=address,
|
|
snapped_lat=56.838,
|
|
snapped_lon=60.605,
|
|
precision="number",
|
|
provider="nominatim",
|
|
)
|
|
|
|
|
|
async def test_reverse_geocode_prefers_cadastral_over_nominatim() -> None:
|
|
"""Cadastral returns address → Yandex/Nominatim never called."""
|
|
db = MagicMock()
|
|
with (
|
|
patch(
|
|
"app.services.geocoder._cadastral_reverse_sync_full",
|
|
return_value=("ул. Малышева, 30, Екатеринбург", 56.8385, 60.6052),
|
|
) as mock_cad,
|
|
patch("app.services.geocoder.settings") as mock_settings,
|
|
patch(
|
|
"app.services.geocoder._yandex_reverse",
|
|
new_callable=AsyncMock,
|
|
) as mock_yandex,
|
|
patch(
|
|
"app.services.geocoder._nominatim_reverse",
|
|
new_callable=AsyncMock,
|
|
) as mock_nom,
|
|
):
|
|
mock_settings.yandex_geocoder_api_key = "fake-key"
|
|
result = await reverse_geocode(56.838, 60.605, db=db)
|
|
|
|
assert result is not None
|
|
assert result.address == "ул. Малышева, 30, Екатеринбург"
|
|
assert result.snapped_lat == 56.8385
|
|
assert result.snapped_lon == 60.6052
|
|
assert result.precision == "number"
|
|
assert result.provider == "cadastral"
|
|
mock_cad.assert_called_once_with(db, 56.838, 60.605)
|
|
mock_yandex.assert_not_called()
|
|
mock_nom.assert_not_called()
|
|
|
|
|
|
async def test_reverse_geocode_uses_yandex_when_cadastral_empty() -> None:
|
|
"""Cadastral None + Yandex key set → Yandex called, Nominatim skipped."""
|
|
db = MagicMock()
|
|
yandex_result = ReverseGeocodeResult(
|
|
address="улица Малышева, 51, Екатеринбург",
|
|
snapped_lat=56.838004,
|
|
snapped_lon=60.586155,
|
|
precision="exact",
|
|
provider="yandex",
|
|
)
|
|
with (
|
|
patch("app.services.geocoder._cadastral_reverse_sync_full", return_value=None),
|
|
patch("app.services.geocoder.settings") as mock_settings,
|
|
patch(
|
|
"app.services.geocoder._yandex_reverse",
|
|
new_callable=AsyncMock,
|
|
return_value=yandex_result,
|
|
) as mock_yandex,
|
|
patch(
|
|
"app.services.geocoder._nominatim_reverse",
|
|
new_callable=AsyncMock,
|
|
) as mock_nom,
|
|
):
|
|
mock_settings.yandex_geocoder_api_key = "fake-key"
|
|
result = await reverse_geocode(56.838, 60.586, db=db)
|
|
|
|
assert result is not None
|
|
assert result.provider == "yandex"
|
|
assert result.precision == "exact"
|
|
assert result.snapped_lat == 56.838004
|
|
mock_yandex.assert_called_once()
|
|
mock_nom.assert_not_called()
|
|
|
|
|
|
async def test_reverse_geocode_falls_back_to_nominatim_when_cadastral_none() -> None:
|
|
"""Cadastral None + no Yandex key → Nominatim called."""
|
|
db = MagicMock()
|
|
with (
|
|
patch("app.services.geocoder._cadastral_reverse_sync_full", return_value=None),
|
|
patch("app.services.geocoder.settings") as mock_settings,
|
|
patch(
|
|
"app.services.geocoder._nominatim_reverse",
|
|
new_callable=AsyncMock,
|
|
return_value=_nom_result("Трамвайный переулок, 2, Екатеринбург"),
|
|
) as mock_nom,
|
|
):
|
|
mock_settings.yandex_geocoder_api_key = None
|
|
result = await reverse_geocode(56.838, 60.605, db=db)
|
|
|
|
assert result is not None
|
|
assert result.address == "Трамвайный переулок, 2, Екатеринбург"
|
|
assert result.provider == "nominatim"
|
|
mock_nom.assert_called_once_with(56.838, 60.605)
|
|
|
|
|
|
async def test_reverse_geocode_returns_none_when_all_providers_fail() -> None:
|
|
"""Cadastral + Yandex + Nominatim — все падают → returns None, no exception."""
|
|
db = MagicMock()
|
|
with (
|
|
patch("app.services.geocoder._cadastral_reverse_sync_full", return_value=None),
|
|
patch("app.services.geocoder.settings") as mock_settings,
|
|
patch(
|
|
"app.services.geocoder._yandex_reverse",
|
|
new_callable=AsyncMock,
|
|
side_effect=RuntimeError("Yandex 429 rate-limited"),
|
|
),
|
|
patch(
|
|
"app.services.geocoder._nominatim_reverse",
|
|
new_callable=AsyncMock,
|
|
side_effect=RuntimeError("Nominatim 403 Forbidden — IP banned"),
|
|
),
|
|
):
|
|
mock_settings.yandex_geocoder_api_key = "fake-key"
|
|
result = await reverse_geocode(56.838, 60.605, db=db)
|
|
|
|
# Must return None, not raise
|
|
assert result is None
|
|
|
|
|
|
async def test_reverse_geocode_without_db_skips_cadastral() -> None:
|
|
"""When db=None, cadastral tier is skipped entirely → goes straight to Yandex/Nominatim."""
|
|
with (
|
|
patch(
|
|
"app.services.geocoder._cadastral_reverse_sync_full",
|
|
) as mock_cad,
|
|
patch("app.services.geocoder.settings") as mock_settings,
|
|
patch(
|
|
"app.services.geocoder._nominatim_reverse",
|
|
new_callable=AsyncMock,
|
|
return_value=_nom_result("ул. Ленина, 1, Екатеринбург"),
|
|
),
|
|
):
|
|
mock_settings.yandex_geocoder_api_key = None
|
|
result = await reverse_geocode(56.838, 60.605, db=None)
|
|
|
|
mock_cad.assert_not_called()
|
|
assert result is not None
|
|
assert result.address == "ул. Ленина, 1, Екатеринбург"
|
|
|
|
|
|
async def test_reverse_geocode_falls_through_yandex_to_nominatim_on_exception() -> None:
|
|
"""Yandex raises → Nominatim still tried."""
|
|
db = MagicMock()
|
|
with (
|
|
patch("app.services.geocoder._cadastral_reverse_sync_full", return_value=None),
|
|
patch("app.services.geocoder.settings") as mock_settings,
|
|
patch(
|
|
"app.services.geocoder._yandex_reverse",
|
|
new_callable=AsyncMock,
|
|
side_effect=RuntimeError("Yandex 500"),
|
|
),
|
|
patch(
|
|
"app.services.geocoder._nominatim_reverse",
|
|
new_callable=AsyncMock,
|
|
return_value=_nom_result("ул. Дублёр, 1"),
|
|
) as mock_nom,
|
|
):
|
|
mock_settings.yandex_geocoder_api_key = "fake-key"
|
|
result = await reverse_geocode(56.838, 60.605, db=db)
|
|
|
|
assert result is not None
|
|
assert result.provider == "nominatim"
|
|
mock_nom.assert_called_once()
|
|
|
|
|
|
# ── geocode: cadastral as first tier ─────────────────────────────────────────
|
|
|
|
async def test_geocode_uses_cadastral_before_yandex() -> None:
|
|
"""Cadastral FDW returns suggestion → geocode returns it without hitting Yandex."""
|
|
db = MagicMock()
|
|
# Cache miss
|
|
db.execute.side_effect = [
|
|
MagicMock(fetchone=lambda: None), # _cache_get
|
|
]
|
|
|
|
cad_suggestion = GeocodeSuggestion(
|
|
label="ул. Хохрякова, 81",
|
|
full_address="ул. Хохрякова, 81, Екатеринбург",
|
|
lat=56.840,
|
|
lon=60.610,
|
|
kind="house",
|
|
)
|
|
|
|
with (
|
|
patch("app.services.geocoder._cache_get", return_value=None),
|
|
patch(
|
|
"app.services.geocoder._cadastral_forward_sync",
|
|
return_value=[cad_suggestion],
|
|
) as mock_cad,
|
|
patch("app.services.geocoder._cache_put"),
|
|
patch("app.services.geocoder._yandex_lookup", new_callable=AsyncMock) as mock_yandex,
|
|
):
|
|
result = await geocode("Хохрякова 81", db)
|
|
|
|
assert result is not None
|
|
assert result.lat == 56.840
|
|
assert result.lon == 60.610
|
|
assert result.confidence == "exact"
|
|
mock_cad.assert_called_once()
|
|
mock_yandex.assert_not_called()
|
|
|
|
|
|
# ── suggest: cadastral as first tier ─────────────────────────────────────────
|
|
|
|
async def test_suggest_uses_cadastral_before_yandex() -> None:
|
|
"""Cadastral FDW returns results → suggest returns them without hitting Yandex."""
|
|
db = MagicMock()
|
|
cad_suggestions = [
|
|
GeocodeSuggestion(
|
|
label="ул. Малышева, 30",
|
|
full_address="ул. Малышева, 30, Екатеринбург",
|
|
lat=56.838,
|
|
lon=60.605,
|
|
kind="house",
|
|
)
|
|
]
|
|
|
|
with (
|
|
patch(
|
|
"app.services.geocoder._cadastral_forward_sync",
|
|
return_value=cad_suggestions,
|
|
) as mock_cad,
|
|
patch(
|
|
"app.services.geocoder._yandex_suggest",
|
|
new_callable=AsyncMock,
|
|
) as mock_yandex,
|
|
):
|
|
results = await suggest("Малышева", db=db, limit=8)
|
|
|
|
assert len(results) == 1
|
|
assert results[0].full_address == "ул. Малышева, 30, Екатеринбург"
|
|
mock_cad.assert_called_once()
|
|
mock_yandex.assert_not_called()
|
|
|
|
|
|
async def test_suggest_falls_back_to_yandex_when_cadastral_empty() -> None:
|
|
"""Cadastral returns [] → falls through to Yandex."""
|
|
db = MagicMock()
|
|
yandex_result = GeocodeSuggestion(
|
|
label="Малышева · Октябрьский",
|
|
full_address="улица Малышева, Екатеринбург",
|
|
lat=56.838,
|
|
lon=60.605,
|
|
kind="street",
|
|
)
|
|
|
|
with (
|
|
patch("app.services.geocoder._cadastral_forward_sync", return_value=[]),
|
|
patch("app.services.geocoder.settings") as mock_settings,
|
|
patch(
|
|
"app.services.geocoder._yandex_suggest",
|
|
new_callable=AsyncMock,
|
|
return_value=[yandex_result],
|
|
) as mock_yandex,
|
|
):
|
|
mock_settings.yandex_geocoder_api_key = "fake-key"
|
|
results = await suggest("Малышева", db=db, limit=8)
|
|
|
|
assert len(results) == 1
|
|
mock_yandex.assert_called_once()
|
|
|
|
|
|
# ── ensure_fdw_user_mapping: SQL injection / whitelist guards ─────────────────
|
|
|
|
def test_ensure_fdw_user_mapping_rejects_quote_in_password() -> None:
|
|
"""Password with embedded quote must be rejected (SQL injection guard)."""
|
|
from app.core.fdw import ensure_fdw_user_mapping
|
|
|
|
fake_settings = MagicMock()
|
|
fake_settings.gendesign_fdw_password = "x' OR '1'='1"
|
|
with patch("app.core.fdw.settings", fake_settings):
|
|
db = MagicMock()
|
|
try:
|
|
ensure_fdw_user_mapping(db)
|
|
assert False, "should have raised ValueError"
|
|
except ValueError as e:
|
|
assert "format whitelist" in str(e)
|
|
# No DB call should have happened
|
|
db.execute.assert_not_called()
|
|
|
|
|
|
def test_ensure_fdw_user_mapping_rejects_short_password() -> None:
|
|
"""Password shorter than 32 chars rejected — forces rotation away from weak defaults."""
|
|
from app.core.fdw import ensure_fdw_user_mapping
|
|
|
|
fake_settings = MagicMock()
|
|
fake_settings.gendesign_fdw_password = "short"
|
|
with patch("app.core.fdw.settings", fake_settings):
|
|
db = MagicMock()
|
|
try:
|
|
ensure_fdw_user_mapping(db)
|
|
assert False, "should have raised ValueError"
|
|
except ValueError:
|
|
pass
|
|
db.execute.assert_not_called()
|
|
|
|
|
|
def test_ensure_fdw_user_mapping_accepts_valid_password() -> None:
|
|
"""Hex password 32+ chars passes whitelist and issues CREATE USER MAPPING."""
|
|
from app.core.fdw import ensure_fdw_user_mapping
|
|
|
|
fake_settings = MagicMock()
|
|
fake_settings.gendesign_fdw_password = "a" * 64 # valid hex-like
|
|
with patch("app.core.fdw.settings", fake_settings):
|
|
db = MagicMock()
|
|
db.execute.return_value.first.return_value = None # mapping doesn't exist
|
|
ensure_fdw_user_mapping(db)
|
|
# CREATE USER MAPPING called — inspect TextClause text via str(args[0])
|
|
sql_texts = [str(c.args[0]) for c in db.execute.call_args_list]
|
|
assert any("CREATE USER MAPPING" in s for s in sql_texts)
|
|
db.commit.assert_called_once()
|