Co-authored-by: bot-backend <bot-backend@gendsgn.local> Co-committed-by: bot-backend <bot-backend@gendsgn.local>
268 lines
10 KiB
Python
268 lines
10 KiB
Python
"""Тесты PAT-подзон (#1065 / #1067):
|
||
- репроекция координат МСК-66 → WGS84 (проверка оси, bbox Кольцово)
|
||
- построение MultiPolygon из секторов (внешние + дырки)
|
||
- lookup graceful при отсутствии таблицы / WKT
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
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:
|
||
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.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 == []
|