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

"""Тесты PAT-подзон (#1065 / #1067):
- репроекция координат МСК-66 → WGS84 (проверка оси, bbox Кольцово)
- построение MultiPolygon из секторов (внешние + дырки)
- lookup graceful при отсутствии таблицы / WKT
"""

from __future__ import annotations

import importlib
import pathlib
from contextlib import contextmanager
from typing import Any

import pytest
from pyproj import CRS, Transformer
from shapely.geometry import MultiPolygon
from sqlalchemy.exc import OperationalError

from app.services.site_finder.pat_lookup import parcel_pat_subzones
from app.workers.tasks.pat_subzones_load import (
    _build_subzone_geometry,
    _close_ring,
    _is_hole_sector,
)

# ── Репроекция ──────────────────────────────────────────────────────────────
_MSK66_PROJ4 = (
    "+proj=tmerc +lat_0=0 +lon_0=60 +k=1"
    " +x_0=1496944 +y_0=-5911077"
    " +ellps=krass +towgs84=23.92,-141.27,-80.9,0,0.35,0.82,-0.12"
    " +units=m +no_defs"
)
_TR = Transformer.from_crs(
    CRS.from_proj4(_MSK66_PROJ4),
    CRS.from_epsg(4326),
    always_xy=True,
)

# Известные точки из subzone 1 sector 1 (koltsovo_367p_subzones.json)
# point=[X=northing, Y=easting]; верный порядок: transform(easting=Y, northing=X)
_KOLTSOVO_POINTS = [
    ([381550.81, 1547094.51], (60.818, 56.751)),  # первая точка сектора 1
    ([381553.39, 1547585.62], (60.826, 56.752)),  # третья
    ([381562.21, 1547603.09], (60.827, 56.752)),  # четвёртая
]

_KOLTSOVO_LON_MIN = 60.5
_KOLTSOVO_LON_MAX = 61.2
_KOLTSOVO_LAT_MIN = 56.5
_KOLTSOVO_LAT_MAX = 57.0


@pytest.mark.parametrize("point,expected", _KOLTSOVO_POINTS)
def test_reproject_msk66_to_wgs84_in_koltsovo(
    point: list[float], expected: tuple[float, float]
) -> None:
    """Репроекция МСК-66 [northing, easting] → WGS84 даёт координаты в bbox Кольцово."""
    northing, easting = point[0], point[1]
    lon, lat = _TR.transform(easting, northing)  # always_xy: x=easting, y=northing

    # Грубый bbox Кольцово
    msg_lon = f"lon={lon:.4f} вне bbox Кольцово [{_KOLTSOVO_LON_MIN}, {_KOLTSOVO_LON_MAX}]"
    assert _KOLTSOVO_LON_MIN <= lon <= _KOLTSOVO_LON_MAX, msg_lon
    msg_lat = f"lat={lat:.4f} вне bbox Кольцово [{_KOLTSOVO_LAT_MIN}, {_KOLTSOVO_LAT_MAX}]"
    assert _KOLTSOVO_LAT_MIN <= lat <= _KOLTSOVO_LAT_MAX, msg_lat

    # Проверяем приблизительное совпадение с ожидаемым (±0.005°)
    exp_lon, exp_lat = expected
    assert abs(lon - exp_lon) < 0.005, f"lon расхождение: got {lon:.4f} exp {exp_lon}"
    assert abs(lat - exp_lat) < 0.005, f"lat расхождение: got {lat:.4f} exp {exp_lat}"


def test_axis_swap_gives_wrong_result() -> None:
    """Неверный порядок осей (x=northing, y=easting) выдаёт координаты вне Кольцово."""
    northing, easting = 381550.81, 1547094.51
    # Неверный порядок: x=northing, y=easting
    lon_wrong, lat_wrong = _TR.transform(northing, easting)
    # Должно быть ВНЕ bbox Кольцово
    in_koltsovo = (
        _KOLTSOVO_LON_MIN <= lon_wrong <= _KOLTSOVO_LON_MAX
        and _KOLTSOVO_LAT_MIN <= lat_wrong <= _KOLTSOVO_LAT_MAX
    )
    msg = (
        f"Неверный порядок осей дал координаты в Кольцово: lon={lon_wrong:.4f} lat={lat_wrong:.4f}"
    )
    assert not in_koltsovo, msg


# ── Построение геометрии ────────────────────────────────────────────────────


def _make_sector(label: str | None, pts: list[tuple[float, float]]) -> dict[str, Any]:
    """Вспомогательная фикстура: сектор с точками уже в WGS84-диапазоне.

    Используем точки в МСК-66 которые проецируются в WGS84 Кольцово.
    """
    # Конвертируем WGS84-like точки в «псевдо МСК-66» для реального прогона через _build.
    # Проще передать реальные МСК-66 точки из JSON.
    return {
        "label": label,
        "points": [list(p) for p in pts],
    }


# Набор точек в МСК-66 (northing, easting) образующих маленький треугольник около Кольцово
_TRI_OUTER = [
    [381500.0, 1547000.0],
    [381600.0, 1547000.0],
    [381600.0, 1547100.0],
    [381500.0, 1547000.0],  # замкнуто
]

_TRI_HOLE = [
    [381520.0, 1547020.0],
    [381560.0, 1547020.0],
    [381560.0, 1547060.0],
    [381520.0, 1547020.0],
]


def test_build_single_exterior_sector() -> None:
    """Один внешний сектор → валидный MultiPolygon."""
    sectors = [{"label": "Сектор 1", "points": _TRI_OUTER}]
    geom = _build_subzone_geometry(sectors)
    assert isinstance(geom, MultiPolygon)
    assert not geom.is_empty
    assert geom.is_valid


def test_build_exterior_with_hole() -> None:
    """Внешний сектор + внутренняя граница → Polygon с дыркой."""
    sectors = [
        {"label": "Сектор 1. Внешняя граница", "points": _TRI_OUTER},
        {"label": "Сектор 1. Внутренняя граница 1", "points": _TRI_HOLE},
    ]
    geom = _build_subzone_geometry(sectors)
    assert isinstance(geom, MultiPolygon)
    assert not geom.is_empty
    assert geom.is_valid
    # Площадь с дыркой должна быть меньше только внешней
    outer_only = _build_subzone_geometry([sectors[0]])
    assert geom.area < outer_only.area


def test_build_two_exterior_sectors() -> None:
    """Два внешних сектора → MultiPolygon из двух Polygon-ов."""
    tri2 = [
        [382000.0, 1548000.0],
        [382100.0, 1548000.0],
        [382100.0, 1548100.0],
        [382000.0, 1548000.0],
    ]
    sectors = [
        {"label": "Сектор 1", "points": _TRI_OUTER},
        {"label": "Сектор 2", "points": tri2},
    ]
    geom = _build_subzone_geometry(sectors)
    assert isinstance(geom, MultiPolygon)
    assert not geom.is_empty
    assert geom.is_valid


def test_build_empty_sector_skipped() -> None:
    """Сектор без точек не ломает построение."""
    sectors = [
        {"label": "Пустой", "points": []},
        {"label": "Сектор 1", "points": _TRI_OUTER},
    ]
    geom = _build_subzone_geometry(sectors)
    assert isinstance(geom, MultiPolygon)
    assert not geom.is_empty


def test_is_hole_sector_detection() -> None:
    """Детекция внутренних секторов по label."""
    assert _is_hole_sector("Сектор 1. Внутренняя граница 1") is True
    assert _is_hole_sector("Внутренняя граница") is True
    assert _is_hole_sector("internal boundary") is True
    assert _is_hole_sector("Исключённая территория") is True
    assert _is_hole_sector("Сектор 1. Внешняя граница") is False
    assert _is_hole_sector("Сектор 1") is False
    assert _is_hole_sector(None) is False


def test_close_ring_adds_closing_point() -> None:
    """Незамкнутое кольцо дополняется копией первой точки."""
    coords = [(1.0, 2.0), (3.0, 4.0), (5.0, 6.0)]
    closed = _close_ring(coords)
    assert closed[-1] == closed[0]
    assert len(closed) == 4


def test_close_ring_already_closed() -> None:
    """Уже замкнутое кольцо не изменяется."""
    coords = [(1.0, 2.0), (3.0, 4.0), (1.0, 2.0)]
    closed = _close_ring(coords)
    assert len(closed) == 3


# ── Lookup graceful ─────────────────────────────────────────────────────────


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:
    """begin_nested — реальный (не-swallow) CM: зеркалит SAVEPOINT-фикс #2464
    cluster A finding 7 (parcel_pat_subzones); исключение из db.execute должно
    долетать до except в проде, не глушиться на выходе из ``with``."""

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

    @contextmanager
    def begin_nested(self):  # type: ignore[no-untyped-def]
        yield

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


_WKT = "POLYGON((60.8 56.7, 60.9 56.7, 60.9 56.8, 60.8 56.8, 60.8 56.7))"


def test_pat_lookup_returns_subzones() -> None:
    rows = [
        {
            "subzone_no": 2,
            "name": "Вторая подзона",
            "restriction": "Ограничения второй подзоны",
            "aerodrome": "Екатеринбург (Кольцово)",
        }
    ]
    result = parcel_pat_subzones(_FakeDB(rows), _WKT)
    assert len(result) == 1
    assert result[0]["subzone_no"] == 2
    assert result[0]["name"] == "Вторая подзона"
    assert result[0]["aerodrome"] == "Екатеринбург (Кольцово)"


def test_pat_lookup_empty_when_no_wkt() -> None:
    result = parcel_pat_subzones(_FakeDB([{"subzone_no": 1}]), None)
    assert result == []


def test_pat_lookup_empty_when_no_intersections() -> None:
    result = parcel_pat_subzones(_FakeDB([]), _WKT)
    assert result == []


def test_pat_lookup_graceful_table_missing() -> None:
    """pat_subzones не задеплоена → пусто, без падения."""
    db = _FakeDB(raise_exc=OperationalError("stmt", {}, Exception("relation does not exist")))
    result = parcel_pat_subzones(db, _WKT)
    assert result == []


def test_pat_lookup_graceful_programming_error() -> None:
    """ProgrammingError (колонка не существует) → graceful."""
    from sqlalchemy.exc import ProgrammingError

    db = _FakeDB(raise_exc=ProgrammingError("stmt", {}, Exception("column not found")))
    result = parcel_pat_subzones(db, _WKT)
    assert result == []


def test_pat_lookup_db_error_leaves_session_usable_for_next_query() -> None:
    """#2464 cluster A finding 7: db shared с caller (analyze_parcel) — сбой здесь
    не должен отравлять session для persist_analysis_run и прочих db.execute дальше
    по /analyze. Симулируем: execute кидает ОДИН раз, затем следующий (не связанный)
    execute на ТОЙ ЖЕ session отрабатывает нормально."""
    from sqlalchemy.exc import OperationalError

    db = _FakeDB(raise_exc=OperationalError("stmt", {}, Exception("relation does not exist")))
    result = parcel_pat_subzones(db, _WKT)
    assert result == []
    assert db.execute_calls == 1

    # Сессия осталась usable: сбрасываем "raise" (имитируя, что savepoint откатился
    # и не поломала outer-транзакцию) и проверяем, что db.execute снова отрабатывает.
    db._raise = None
    db._rows = [{"subzone_no": 3, "name": "N", "restriction": "R", "aerodrome": "A"}]
    next_result = parcel_pat_subzones(db, _WKT)
    assert next_result == [{"subzone_no": 3, "name": "N", "restriction": "R", "aerodrome": "A"}]
    assert db.execute_calls == 2


# ── Резолв пути к JSON (#1150) ──────────────────────────────────────────────


def test_json_path_candidates_non_empty() -> None:
    """_JSON_PATH_CANDIDATES содержит хотя бы один кандидат."""
    from app.workers.tasks.pat_subzones_load import _JSON_PATH_CANDIDATES

    assert len(_JSON_PATH_CANDIDATES) >= 1


def test_json_path_candidates_end_with_expected_suffix() -> None:
    """Каждый кандидат заканчивается на data/pat/koltsovo_367p_subzones.json."""
    import pathlib

    from app.workers.tasks.pat_subzones_load import _JSON_PATH_CANDIDATES

    expected_suffix = pathlib.Path("data") / "pat" / "koltsovo_367p_subzones.json"
    for candidate in _JSON_PATH_CANDIDATES:
        # последние 3 части пути совпадают с ожидаемыми
        assert (
            candidate.parts[-3:] == expected_suffix.parts
        ), f"Кандидат {candidate} не заканчивается на {expected_suffix}"


def test_resolve_json_path_finds_real_file() -> None:
    """_resolve_json_path() находит JSON-файл в репо (data/pat/ на уровне repo root)."""
    from app.workers.tasks.pat_subzones_load import _resolve_json_path

    resolved = _resolve_json_path()
    assert resolved.exists(), (
        f"JSON-файл подзон ПАТ не найден по пути {resolved}. "
        "Убедитесь что data/pat/koltsovo_367p_subzones.json присутствует в репо."
    )
    assert resolved.name == "koltsovo_367p_subzones.json"


def test_resolve_json_path_env_override(
    monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
) -> None:
    """PAT_JSON_PATH env-переменная переопределяет путь к JSON."""
    import app.workers.tasks.pat_subzones_load as mod

    # Создаём временный файл
    fake_json = tmp_path / "fake_subzones.json"
    fake_json.write_text("{}", encoding="utf-8")

    monkeypatch.setenv("PAT_JSON_PATH", str(fake_json))

    # Перезагружаем модуль чтобы env-переменная была подхвачена на уровне модуля
    mod_reloaded = importlib.reload(mod)

    resolved = mod_reloaded._resolve_json_path()
    assert resolved == fake_json

    # Откат — перезагрузить без env
    monkeypatch.delenv("PAT_JSON_PATH", raising=False)
    importlib.reload(mod)