bot-backend
91b04a1851
#1786 merged before the test fix landed → main red on test_geocode_uses_cadastral_before_yandex. The new geoportal house-match tier runs first in geocode(); this legacy cadastral test must mock it to None so it exercises the cadastral forward path (else the MagicMock db row leaks through the new tier as lat=1.0).
521 lines
19 KiB
Python
521 lines
19 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
|
||
|
||
import pytest
|
||
|
||
# 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),
|
||
# geoportal house-match (new primary local tier) misses → cadastral path exercised
|
||
patch("app.services.geocoder._geoportal_house_match", return_value=None),
|
||
# anchored house-match (new primary cadastral step) misses → legacy forward used
|
||
patch("app.services.geocoder._cadastral_house_match", 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 [] → DaData not configured → 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._dadata_suggest",
|
||
new_callable=AsyncMock,
|
||
return_value=[], # DaData empty → falls through
|
||
),
|
||
patch(
|
||
"app.services.geocoder._yandex_suggest",
|
||
new_callable=AsyncMock,
|
||
return_value=[yandex_result],
|
||
) as mock_yandex,
|
||
):
|
||
mock_settings.dadata_api_token = "fake-dadata-token" # gate to call _dadata_suggest
|
||
mock_settings.yandex_geocoder_api_key = "fake-key"
|
||
results = await suggest("Малышева", db=db, limit=8)
|
||
|
||
assert len(results) == 1
|
||
mock_yandex.assert_called_once()
|
||
|
||
|
||
async def test_suggest_uses_dadata_before_yandex_when_cadastral_empty() -> None:
|
||
"""Cadastral returns [] → DaData tier returns results → Yandex never called (PR Q2)."""
|
||
db = MagicMock()
|
||
dadata_result = GeocodeSuggestion(
|
||
label="г Екатеринбург, ул Малышева, д 30",
|
||
full_address="620075, Свердловская обл, г Екатеринбург, ул Малышева, д 30",
|
||
lat=56.838011,
|
||
lon=60.601023,
|
||
kind="house",
|
||
)
|
||
|
||
with (
|
||
patch("app.services.geocoder._cadastral_forward_sync", return_value=[]),
|
||
patch("app.services.geocoder.settings") as mock_settings,
|
||
patch(
|
||
"app.services.geocoder._dadata_suggest",
|
||
new_callable=AsyncMock,
|
||
return_value=[dadata_result],
|
||
) as mock_dadata,
|
||
patch(
|
||
"app.services.geocoder._yandex_suggest",
|
||
new_callable=AsyncMock,
|
||
) as mock_yandex,
|
||
):
|
||
mock_settings.dadata_api_token = "fake-dadata-token"
|
||
mock_settings.yandex_geocoder_api_key = "fake-yandex-key"
|
||
results = await suggest("Малышева", db=db, limit=8)
|
||
|
||
assert len(results) == 1
|
||
assert results[0].lat == 56.838011
|
||
mock_dadata.assert_called_once()
|
||
mock_yandex.assert_not_called()
|
||
|
||
|
||
# ── 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()
|
||
with pytest.raises(ValueError, match="format whitelist"):
|
||
ensure_fdw_user_mapping(db)
|
||
# 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()
|
||
with pytest.raises(ValueError):
|
||
ensure_fdw_user_mapping(db)
|
||
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()
|