gendesign/backend/tests/services/test_ird_overlay_lookup.py

"""Тесты parcel_ird_overlaps (#1067 D9) — ИРД-overlay lookup для analyze."""

from __future__ import annotations

from contextlib import contextmanager
from typing import Any

import pytest
from sqlalchemy.exc import DataError, OperationalError, ProgrammingError

from app.services.site_finder.ird_overlay_lookup import parcel_ird_overlaps


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

    def mappings(self) -> _Result:
        return self

    def all(self) -> list[dict[str, Any]]:
        return self._rows


class _FakeDB:
    @contextmanager
    def begin_nested(self):  # type: ignore[no-untyped-def]
        """#2464: lookup оборачивает свой db.execute в SAVEPOINT (сессия общая с
        analyze_parcel). Здесь достаточно пустого контекст-менеджера: эти тесты про
        логику самого lookup'а. Что SAVEPOINT РЕАЛЬНО откатывает aborted-транзакцию,
        проверяет tests/services/site_finder/test_2464_ird_lookups_savepoint.py на
        двойнике с настоящей семантикой Postgres."""
        yield

    def __init__(
        self, rows: list[dict[str, Any]] | None = None, raise_exc: Exception | None = None
    ) -> None:
        self._rows = rows or []
        self._raise = raise_exc

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


_WKT = "POLYGON((60.5 56.8, 60.6 56.8, 60.6 56.9, 60.5 56.9, 60.5 56.8))"


def test_groups_overlaps_by_kind() -> None:
    rows = [
        {
            "layer_kind": "zouit_okn",
            "reg_numb_border": "66:41-6.41",
            "category": 36940,
            "subcategory": 20,
            "zone_index": None,
            "zone_type_name": None,
            "content": None,
        },
        {
            "layer_kind": "zouit_okn",
            "reg_numb_border": "66:41-6.42",
            "category": 36940,
            "subcategory": 20,
            "zone_index": None,
            "zone_type_name": None,
            "content": None,
        },
        {
            "layer_kind": "territorial_zones",
            "reg_numb_border": "66:41-7.10",
            "category": 472819,
            "subcategory": 1,
            "zone_index": "Ц-1",
            "zone_type_name": None,
            "content": None,
        },
    ]
    res = parcel_ird_overlaps(_FakeDB(rows), _WKT)
    assert len(res["ird_overlaps"]) == 3
    assert res["ird_by_kind"] == {"zouit_okn": 2, "territorial_zones": 1}
    tz = next(o for o in res["ird_overlaps"] if o["layer_kind"] == "territorial_zones")
    assert tz["reg_numb_border"] == "66:41-7.10"
    assert tz["zone_index"] == "Ц-1"


def test_content_passthrough() -> None:
    """content из gknspecial_harvest пробрасывается в каждый overlap-dict."""
    content_val = {"restrictions": "запрещена жилая застройка", "zone_state": "Действующий"}
    rows = [
        {
            "layer_kind": "water_protection",
            "reg_numb_border": "66:41-6.4125",
            "category": None,
            "subcategory": None,
            "zone_index": None,
            "zone_type_name": "Водоохранная зона р. Исеть",
            "content": content_val,
        },
    ]
    res = parcel_ird_overlaps(_FakeDB(rows), _WKT)
    assert res["ird_overlaps"][0]["content"] == content_val


def test_empty_when_no_wkt() -> None:
    res = parcel_ird_overlaps(_FakeDB([{"layer_kind": "x"}]), None)
    assert res == {"ird_overlaps": [], "ird_by_kind": {}}


def test_empty_when_no_overlap() -> None:
    res = parcel_ird_overlaps(_FakeDB([]), _WKT)
    assert res == {"ird_overlaps": [], "ird_by_kind": {}}


@pytest.mark.parametrize(
    ("exc", "case"),
    [
        (
            OperationalError("stmt", {}, Exception("no such table")),
            "operational-error: connection drop / БД недоступна",
        ),
        (
            ProgrammingError("stmt", {}, Exception('relation "ird_overlays" does not exist')),
            "programming-error: UndefinedTable — pre-migration / D9b не задеплоен",
        ),
        (
            DataError("stmt", {}, Exception("parse error - invalid geometry")),
            "data-error: malformed WKT → PostGIS ST_GeomFromText ERROR",
        ),
    ],
)
def test_graceful_when_db_fails(exc: Exception, case: str) -> None:
    """ird_overlays недоступна / malformed WKT → пусто, analyze не падает.

    Покрывает:
    - OperationalError — БД недоступна.
    - ProgrammingError — таблица ещё не задеплоена (реальный класс при UndefinedTable).
    - DataError — malformed parcel_wkt (PostGIS ST_GeomFromText бросает ERROR,
      SQLAlchemy конвертит в DataError).
    """
    db = _FakeDB(raise_exc=exc)
    res = parcel_ird_overlaps(db, _WKT)
    assert res == {"ird_overlaps": [], "ird_by_kind": {}}, case