gendesign/tradein-mvp/backend/tests/services/test_cadastral_reverse.py

"""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,
    _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 ───────────────────────────────────────────────────────────

async def test_reverse_geocode_prefers_cadastral_over_nominatim() -> None:
    """Cadastral returns address → Nominatim is never called."""
    db = MagicMock()
    with (
        patch(
            "app.services.geocoder._cadastral_reverse_sync",
            return_value="ул. Малышева, 30, Екатеринбург",
        ) as mock_cad,
        patch(
            "app.services.geocoder._nominatim_reverse",
            new_callable=AsyncMock,
        ) as mock_nom,
    ):
        result = await reverse_geocode(56.838, 60.605, db=db)

    assert result == "ул. Малышева, 30, Екатеринбург"
    mock_cad.assert_called_once_with(db, 56.838, 60.605)
    mock_nom.assert_not_called()


async def test_reverse_geocode_falls_back_to_nominatim_when_cadastral_none() -> None:
    """Cadastral returns None → Nominatim called and result returned."""
    db = MagicMock()
    with (
        patch("app.services.geocoder._cadastral_reverse_sync", return_value=None),
        patch(
            "app.services.geocoder._nominatim_reverse",
            new_callable=AsyncMock,
            return_value="Трамвайный переулок, 2, Екатеринбург",
        ) as mock_nom,
    ):
        result = await reverse_geocode(56.838, 60.605, db=db)

    assert result == "Трамвайный переулок, 2, Екатеринбург"
    mock_nom.assert_called_once_with(56.838, 60.605)


async def test_reverse_geocode_returns_none_when_both_fail() -> None:
    """Both cadastral and Nominatim fail → returns None, no exception raised (500 fix)."""
    db = MagicMock()
    with (
        patch("app.services.geocoder._cadastral_reverse_sync", return_value=None),
        patch(
            "app.services.geocoder._nominatim_reverse",
            new_callable=AsyncMock,
            side_effect=RuntimeError("Nominatim 403 Forbidden — IP banned"),
        ),
    ):
        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 Nominatim."""
    with (
        patch(
            "app.services.geocoder._cadastral_reverse_sync",
        ) as mock_cad,
        patch(
            "app.services.geocoder._nominatim_reverse",
            new_callable=AsyncMock,
            return_value="ул. Ленина, 1, Екатеринбург",
        ),
    ):
        result = await reverse_geocode(56.838, 60.605, db=None)

    mock_cad.assert_not_called()
    assert result == "ул. Ленина, 1, Екатеринбург"


# ── 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()