Light1YT
2e428405d2
parcel_ird_overlaps ловил только (OperationalError, ProgrammingError). При D9b-wiring в analyze malformed WKT в ST_GeomFromText давал PostGIS ERROR → SQLAlchemy DataError, который пробивал try/except → analyze падал вместо graceful-degrade. - ird_overlay_lookup: + DataError в imports и в except tuple. - quarter_dump_lookup: симметричное расширение в _get_engineering_*. - test_ird_overlay_lookup: parametrized test_graceful_when_db_fails покрывает 3 класса (Operational/Programming + DataError для WKT). Closes #1095
134 lines
4.4 KiB
Python
134 lines
4.4 KiB
Python
"""Тесты parcel_ird_overlaps (#1067 D9) — ИРД-overlay lookup для analyze."""
|
||
|
||
from __future__ import annotations
|
||
|
||
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:
|
||
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
|