gendesign/backend/tests/services/site_finder/test_riasurt_lookup.py

"""Тесты riasurt_lookup (#108) — is_in_aglomeration_but_not_ekb + parcel_riasurt_gate.

БД мокается фейк-сессией, возвращающей заранее заданные mapping-строки. Проверяем
агломерация-гейтинг (ЕКБ-сити исключён), topic-бакеты, graceful при ошибке БД.
"""

from __future__ import annotations

import contextlib
from typing import Any

from sqlalchemy.exc import ProgrammingError

from app.services.site_finder import riasurt_lookup
from app.services.site_finder.riasurt_lookup import (
    is_in_aglomeration_but_not_ekb,
    parcel_riasurt_gate,
)

_WKT = "POLYGON((60 56,60.1 56,60.1 56.1,60 56.1,60 56))"


def test_is_in_aglomeration_excludes_ekb_city() -> None:
    """ЕКБ-сити (66:41) → False."""
    assert is_in_aglomeration_but_not_ekb("66:41:0204016:10") is False
    assert is_in_aglomeration_but_not_ekb("66:41") is False


def test_is_in_aglomeration_includes_outskirts() -> None:
    """Окраины (66:35 Берёзовский, 66:62 В.Пышма, 66:25 Сысерть) → True."""
    assert is_in_aglomeration_but_not_ekb("66:35:0101001:5") is True
    assert is_in_aglomeration_but_not_ekb("66:62:1234567:1") is True
    assert is_in_aglomeration_but_not_ekb("66:25:0000001:99") is True


def test_is_in_aglomeration_rejects_non_sverdl_and_empty() -> None:
    """Не-Свердл (не 66:*) и пусто → False."""
    assert is_in_aglomeration_but_not_ekb("50:21:0000001:1") is False
    assert is_in_aglomeration_but_not_ekb(None) is False
    assert is_in_aglomeration_but_not_ekb("") is False


class _Row(dict):
    """mapping-строка: поддерживает r["key"]."""


class _Result:
    def __init__(self, rows: list[_Row]) -> None:
        self._rows = rows

    def mappings(self) -> _Result:
        return self

    def all(self) -> list[_Row]:
        return self._rows


class _FakeDB:
    def __init__(self, rows: list[_Row] | Exception) -> None:
        self._rows = rows
        self.begin_nested_calls = 0

    @contextlib.contextmanager
    def begin_nested(self) -> Any:
        """Фейковый SAVEPOINT — не глотает исключение из execute (как реальный)."""
        self.begin_nested_calls += 1
        yield

    def execute(self, sql: Any, params: dict[str, Any] | None = None) -> _Result:
        if isinstance(self._rows, Exception):
            raise self._rows
        return _Result(self._rows)


def test_gate_not_applicable_for_ekb_city() -> None:
    """ЕКБ-сити → applicable=False, БД не дёргается."""
    db = _FakeDB([])
    gate = parcel_riasurt_gate(db, _WKT, "66:41:0204016:10")  # type: ignore[arg-type]
    assert gate["applicable"] is False
    assert gate["overlaps"] == []


def test_gate_buckets_topics() -> None:
    """Пересечения раскладываются по topic-бакетам."""
    rows = [
        _Row(
            source_layer_id=845274,
            layer_topic="territorial_zone",
            mo_name="Берёзовский",
            obshnz="Ж-1",
            description="Зона застройки",
        ),
        _Row(
            source_layer_id=846381,
            layer_topic="functional_zone",
            mo_name="Берёзовский",
            obshnz=None,
            description="Жилая функц.зона",
        ),
        _Row(
            source_layer_id=844759,
            layer_topic="red_lines",
            mo_name="Берёзовский",
            obshnz=None,
            description=None,
        ),
        _Row(
            source_layer_id=846365,
            layer_topic="szz",
            mo_name="Берёзовский",
            obshnz=None,
            description=None,
        ),
        _Row(
            source_layer_id=845425,
            layer_topic="flood_zone",
            mo_name="Берёзовский",
            obshnz=None,
            description=None,
        ),
        _Row(
            source_layer_id=844478,
            layer_topic="krt",
            mo_name="Берёзовский",
            obshnz=None,
            description=None,
        ),
    ]
    db = _FakeDB(rows)
    gate = parcel_riasurt_gate(db, _WKT, "66:35:0101001:5")  # type: ignore[arg-type]

    assert gate["applicable"] is True
    assert gate["tier"] == "Зона застройки"
    assert gate["func_zone"] == "Жилая функц.зона"
    assert gate["red_lines"] is True
    assert gate["szz"] is True
    assert gate["szo"] is False
    assert gate["flood_zone"] is True
    assert gate["krt"] is True
    assert len(gate["overlaps"]) == 6


def test_gate_graceful_on_db_error() -> None:
    """Ошибка БД (pre-migration) → пустой gate, не падаем."""
    db = _FakeDB(ProgrammingError("stmt", {}, Exception("no table")))
    gate = parcel_riasurt_gate(db, _WKT, "66:35:0101001:5")  # type: ignore[arg-type]
    assert gate == riasurt_lookup.RIASURT_EMPTY_GATE


def test_gate_empty_wkt_returns_empty() -> None:
    """Нет WKT → пустой gate."""
    db = _FakeDB([])
    gate = parcel_riasurt_gate(db, None, "66:35:0101001:5")  # type: ignore[arg-type]
    assert gate["applicable"] is False


def test_gate_uses_savepoint_and_session_reusable_after_failure() -> None:
    """#2464 cluster A: db.execute обёрнут в begin_nested (SAVEPOINT) — сбой не
    отравляет сессию. Follow-up вызов на ТОЙ ЖЕ db-инстансе успевает."""
    db = _FakeDB(ProgrammingError("stmt", {}, Exception("no table")))
    gate = parcel_riasurt_gate(db, _WKT, "66:35:0101001:5")  # type: ignore[arg-type]
    assert gate == riasurt_lookup.RIASURT_EMPTY_GATE
    assert db.begin_nested_calls == 1

    # follow-up на той же сессии (таблица теперь "доступна").
    db._rows = []
    gate2 = parcel_riasurt_gate(db, _WKT, "66:35:0101001:5")  # type: ignore[arg-type]
    assert gate2["applicable"] is True
    assert db.begin_nested_calls == 2