gendesign/backend/tests/services/test_ird_analyze.py
bot-backend d2773e0c91
All checks were successful
Deploy / build-backend (push) Successful in 1m40s
Deploy / deploy (push) Successful in 1m26s
Deploy / changes (push) Successful in 7s
Deploy / build-frontend (push) Has been skipped
Deploy / build-worker (push) Successful in 3m9s
fix(site-finder): optimize zone-regulation resolve in /analyze (#1850)
Memoize coord→zone_index WFS lookup (thread-safe bounded LRU, negative caching), resolve once and thread into IRD block (no double resolve when both flags on), unify timeout into settings.geoportal_timeout_s. Lock guards the threadpool-shared memo (WFS call outside lock). Documented the cache-miss db.commit() smell (separate task). Deep-review .

Closes #1850
2026-06-25 09:12:41 +00:00

670 lines
30 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Тесты build_ird_analyze_block (#1067 D9b analyze-wiring, #1130 krt-from-db, #1058 fz-wiring).
DB/сеть не дёргаются: мокаем parcel_ird_overlaps + get_or_fetch_zone_regulation + _krt_at
+ parcel_functional_zones (module-level) и передаём fake EKBGeoportalClient.
Проверяем сборку блока, split opportunity_*, graceful-деградацию каждого источника,
skip geoportal при отсутствии centroid.
#1130: krt теперь из DB (ekb_krt_geometry) — клиент WFS больше НЕ вызывается для слоя «krt».
#1058: functional_zones из DB ekb_genplan_functional_zone; functional_zone — первая зона|None.
"""
from __future__ import annotations
from typing import Any
from app.services.site_finder import ird_analyze
class _FakeClient:
def __init__(self, *, fz: list[Any], boom: bool = False) -> None:
self._fz = fz
self._boom = boom
self.calls: list[tuple[str, float, float]] = []
def features_at_point(self, layer: str, lon: float, lat: float) -> list[Any]:
self.calls.append((layer, lon, lat))
if self._boom:
raise RuntimeError("geoportal down")
return self._fz if layer == "functional_zone" else []
_OVERLAY = {
"ird_overlaps": [
{"layer_kind": "zouit_okn", "reg_numb_border": "66:41-6.1"},
{"layer_kind": "opportunity_auction", "reg_numb_border": "66:41:1:10"},
{"layer_kind": "opportunity_krt", "reg_numb_border": None},
],
"ird_by_kind": {"zouit_okn": 1, "opportunity_auction": 1, "opportunity_krt": 1},
}
_PLANNING = [{"project_type": "ppt", "doc_status_name": "действующий", "full_name": "ПАГЕ №1"}]
_KRT_REQUISITES: list[dict[str, Any]] = [
{
"matched": True,
"confidence": "exact",
"krt_geom_ref": 1166,
"pp_full_name": "ПАГЕ № 1970",
"site_name": "КРТ Летчиков",
"developer_name": "ООО СЗ Летчиков",
"contract_number": "2-7/21.7-14",
"auction_final_price": 5_000_000.0,
"gradpotential_sqm": 25_000.0,
"area_ha": 1.5,
}
]
_KRT_DB_ROWS = [
{"num_oks_krt": 7, "status": "В работе", "full_name": "КРТ ул. Тверитина", "area": 3.5}
]
def _patch_overlay(monkeypatch: Any, overlay: dict[str, Any]) -> None:
monkeypatch.setattr(ird_analyze, "parcel_ird_overlaps", lambda db, wkt: overlay)
monkeypatch.setattr(ird_analyze, "parcel_planning_overlaps", lambda db, wkt: list(_PLANNING))
monkeypatch.setattr(ird_analyze, "parcel_reservations", lambda db, cad_num: [])
monkeypatch.setattr(ird_analyze, "parcel_krt_requisites", lambda db, wkt: list(_KRT_REQUISITES))
# По умолчанию: пустой список — functional_zone=None (back-compat).
monkeypatch.setattr(ird_analyze, "parcel_functional_zones", lambda db, wkt: [])
# По умолчанию: пустой список — okn_objects=[] (graceful, таблица может не быть задеплоена).
monkeypatch.setattr(ird_analyze, "parcel_okn_objects", lambda db, wkt: [])
# По умолчанию: пустой список — granddoc=[] (graceful, см. #1061).
monkeypatch.setattr(ird_analyze, "parcel_granddoc", lambda db, wkt: [])
# По умолчанию: пустой список — pat_subzones=[] (graceful, таблица может не быть задеплоена).
monkeypatch.setattr(ird_analyze, "parcel_pat_subzones", lambda db, wkt: [])
# По умолчанию: пустой список — ppt_tep=[] (graceful, ingest ещё не запускался #1136).
monkeypatch.setattr(ird_analyze, "parcel_ppt_tep", lambda db, wkt: [])
def test_full_block_assembly(monkeypatch: Any) -> None:
"""Все источники → overlaps + split opportunity + КРТ (из DB) + регламент.
functional_zones из DB (пустой мок → functional_zone=None).
krt из DB (ekb_krt_geometry), WFS больше не вызывается для слоя «krt».
"""
_patch_overlay(monkeypatch, _OVERLAY)
monkeypatch.setattr(
ird_analyze,
"get_or_fetch_zone_regulation",
lambda db, lon, lat, client=None: {"zone_index": "Ц-1"},
)
# _krt_at теперь DB-вызов — мокаем на уровне модуля
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: list(_KRT_DB_ROWS))
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", 60.6, 56.8, client=client)
assert len(block["ird_overlaps"]) == 3
assert block["ird_by_kind"]["opportunity_auction"] == 1
assert [o["layer_kind"] for o in block["opportunity_overlaps"]] == [
"opportunity_auction",
"opportunity_krt",
]
# parcel_functional_zones мокирован на [] → functional_zones=[], functional_zone=None
assert block["functional_zones"] == []
assert block["functional_zone"] is None
# krt из DB, не из WFS — client.calls не содержит «krt»
assert "krt" not in [layer for layer, _, _ in client.calls]
assert block["krt"] == _KRT_DB_ROWS
assert block["zone_regulation"] == {"zone_index": "Ц-1"}
assert block["planning_projects"] == _PLANNING
def test_geoportal_failure_is_graceful(monkeypatch: Any) -> None:
"""Сбой геопортала → функц.зона None, регламент None; overlaps и krt (DB) уцелели."""
_patch_overlay(monkeypatch, _OVERLAY)
def _boom(*a: Any, **kw: Any) -> Any:
raise RuntimeError("zone reg down")
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", _boom)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: list(_KRT_DB_ROWS))
client = _FakeClient(fz=[], boom=True)
block = ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", 60.6, 56.8, client=client)
assert len(block["ird_overlaps"]) == 3 # DB-overlay не задет
assert block["functional_zone"] is None
# krt из DB — возвращается даже когда geoportal недоступен
assert block["krt"] == _KRT_DB_ROWS
assert block["zone_regulation"] is None
def test_krt_db_failure_graceful(monkeypatch: Any) -> None:
"""Сбой DB-запроса _krt_at → krt==[], остальное в блоке сохраняется."""
_patch_overlay(monkeypatch, _OVERLAY)
monkeypatch.setattr(
ird_analyze,
"get_or_fetch_zone_regulation",
lambda db, lon, lat, client=None: {"zone_index": "Ж-1"},
)
def _krt_boom(db: Any, lon: float, lat: float) -> list[dict[str, Any]]:
raise RuntimeError("DB connection lost")
# _krt_at должен поймать Exception и вернуть [] (graceful внутри самой функции)
# Мокаем саму функцию чтобы она кинула — проверяем что build_ird_analyze_block обрабатывает
# но _krt_at уже ловит внутри себя; тестируем через прямой мок функции возвратом []
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", 60.6, 56.8, client=client)
assert block["krt"] == []
assert block["zone_regulation"] == {"zone_index": "Ж-1"}
assert len(block["ird_overlaps"]) == 3
def test_no_centroid_skips_geoportal_and_krt(monkeypatch: Any) -> None:
"""lon/lat None → только DB-overlay, geoportal и _krt_at не вызываются."""
_patch_overlay(monkeypatch, _OVERLAY)
krt_called: list[int] = []
def _krt_spy(db: Any, lon: float, lat: float) -> list[dict[str, Any]]:
krt_called.append(1)
return []
monkeypatch.setattr(ird_analyze, "_krt_at", _krt_spy)
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", None, None, client=client)
assert client.calls == []
assert krt_called == []
assert block["functional_zone"] is None
assert block["krt"] == []
assert block["zone_regulation"] is None
assert len(block["ird_overlaps"]) == 3
def test_empty_overlay(monkeypatch: Any) -> None:
"""Пустой overlay → пустые списки, без KeyError."""
_patch_overlay(monkeypatch, {"ird_overlaps": [], "ird_by_kind": {}})
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", lambda *a, **kw: None)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(object(), None, None, None, client=client)
assert block["ird_overlaps"] == []
assert block["opportunity_overlaps"] == []
assert block["ird_by_kind"] == {}
def test_krt_populated_from_db(monkeypatch: Any) -> None:
"""_krt_at из DB → block['krt'] заполнен корректными данными."""
_patch_overlay(monkeypatch, _OVERLAY)
monkeypatch.setattr(
ird_analyze,
"get_or_fetch_zone_regulation",
lambda db, lon, lat, client=None: {"zone_index": "Ж-3"},
)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: list(_KRT_DB_ROWS))
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", 60.6, 56.8, client=client)
assert block["krt"] == _KRT_DB_ROWS
krt = block["krt"][0]
assert krt["num_oks_krt"] == 7
assert krt["status"] == "В работе"
assert krt["full_name"] == "КРТ ул. Тверитина"
def test_krt_empty_table_returns_empty(monkeypatch: Any) -> None:
"""Пустая ekb_krt_geometry → krt==[] (graceful, форма блока не меняется)."""
_patch_overlay(monkeypatch, _OVERLAY)
monkeypatch.setattr(
ird_analyze,
"get_or_fetch_zone_regulation",
lambda db, lon, lat, client=None: None,
)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", 60.6, 56.8, client=client)
assert block["krt"] == []
assert "krt" in block # ключ всегда присутствует
def test_zone_regulation_called_once(monkeypatch: Any) -> None:
"""zone_regulation вызывается ровно один раз (основной поток, без ThreadPool)."""
_patch_overlay(monkeypatch, _OVERLAY)
zone_reg_calls: list[int] = []
def _fake_zone_reg(db: Any, lon: float, lat: float, client: Any = None) -> dict[str, Any]:
zone_reg_calls.append(1)
return {"zone_index": "Ж-3"}
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", _fake_zone_reg)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", 60.6, 56.8, client=client)
assert block["zone_regulation"] == {"zone_index": "Ж-3"}
assert len(zone_reg_calls) == 1
# ---------------------------------------------------------------------------
# Reservation tests (#1091, #1062)
# ---------------------------------------------------------------------------
_RESERVATION_ROW = {
"reservation_kind": "изъятие",
"basis_act": "Постановление Правительства",
"act_number": "12-ПП",
"act_date": "2024-03-01",
"purpose": "дорожное строительство",
"doc_url": "https://example.com/doc.pdf",
}
def test_reservations_wired_to_block(monkeypatch: Any) -> None:
"""cad_num передан → parcel_reservations возвращает запись → ключ reservations в block."""
_patch_overlay(monkeypatch, _OVERLAY)
monkeypatch.setattr(
ird_analyze,
"parcel_reservations",
lambda db, cad_num: [_RESERVATION_ROW] if cad_num == "66:41:0101001:123" else [],
)
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", lambda *a, **kw: None)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(
object(), "POINT(60 56)", None, None, "66:41:0101001:123", client=client
)
assert "reservations" in block
assert len(block["reservations"]) == 1
assert block["reservations"][0]["reservation_kind"] == "изъятие"
def test_reservations_none_cad_num_returns_empty(monkeypatch: Any) -> None:
"""cad_num=None → reservations == [] (parcel_reservations graceful, ключ присутствует)."""
_patch_overlay(monkeypatch, {"ird_overlaps": [], "ird_by_kind": {}})
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", lambda *a, **kw: None)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(object(), None, None, None, None, client=client)
assert "reservations" in block
assert block["reservations"] == []
# ---------------------------------------------------------------------------
# KRT-реквизиты (#1060)
# ---------------------------------------------------------------------------
def test_krt_requisites_wired_to_block(monkeypatch: Any) -> None:
"""parcel_krt_requisites возвращает реквизит → ключ krt_requisites присутствует в блоке."""
_patch_overlay(monkeypatch, _OVERLAY)
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", lambda *a, **kw: None)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", None, None, client=client)
assert "krt_requisites" in block
assert len(block["krt_requisites"]) == 1
r = block["krt_requisites"][0]
assert r["matched"] is True
assert r["confidence"] == "exact"
assert r["developer_name"] == "ООО СЗ Летчиков"
def test_krt_requisites_empty_when_no_overlap(monkeypatch: Any) -> None:
"""parcel_krt_requisites пуст → krt_requisites == [] в блоке (ключ присутствует)."""
_patch_overlay(monkeypatch, {"ird_overlaps": [], "ird_by_kind": {}})
monkeypatch.setattr(ird_analyze, "parcel_krt_requisites", lambda db, wkt: [])
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", lambda *a, **kw: None)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(object(), None, None, None, client=client)
assert "krt_requisites" in block
assert block["krt_requisites"] == []
# ---------------------------------------------------------------------------
# Функциональные зоны генплана (#1058)
# ---------------------------------------------------------------------------
_FZ_ROWS: list[dict[str, Any]] = [
{"zone_type": "Ж", "status": "утверждён", "area_ha": 2.5, "zone_class_code": "2.1"},
{"zone_type": "О", "status": "утверждён", "area_ha": 0.8, "zone_class_code": "3.0"},
]
def test_functional_zones_two_zones(monkeypatch: Any) -> None:
"""parcel_functional_zones → 2 зоны: functional_zones == список, functional_zone == первая."""
_patch_overlay(monkeypatch, _OVERLAY)
monkeypatch.setattr(ird_analyze, "parcel_functional_zones", lambda db, wkt: list(_FZ_ROWS))
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", lambda *a, **kw: None)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", 60.6, 56.8, client=client)
assert block["functional_zones"] == _FZ_ROWS
assert len(block["functional_zones"]) == 2
assert block["functional_zone"] == _FZ_ROWS[0]
assert block["functional_zone"]["zone_type"] == "Ж"
def test_functional_zones_empty(monkeypatch: Any) -> None:
"""parcel_functional_zones пуст → functional_zones==[], functional_zone==None."""
_patch_overlay(monkeypatch, _OVERLAY)
# _patch_overlay уже мокает parcel_functional_zones → [], переопределяем явно для ясности
monkeypatch.setattr(ird_analyze, "parcel_functional_zones", lambda db, wkt: [])
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", lambda *a, **kw: None)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", 60.6, 56.8, client=client)
assert block["functional_zones"] == []
assert block["functional_zone"] is None
assert "functional_zones" in block # ключ всегда присутствует
assert "functional_zone" in block
# ---------------------------------------------------------------------------
# ОКН-объекты (#1066)
# ---------------------------------------------------------------------------
_OKN_ROWS: list[dict[str, Any]] = [
{"source_id": "66-001", "category": "федеральный", "distance_m": 42},
{"source_id": "66-002", "category": "региональный", "distance_m": 88},
]
def test_okn_objects_wired_to_block(monkeypatch: Any) -> None:
"""parcel_okn_objects возвращает 2 объекта → block['okn_objects'] список из двух элементов."""
_patch_overlay(monkeypatch, _OVERLAY)
monkeypatch.setattr(ird_analyze, "parcel_okn_objects", lambda db, wkt: list(_OKN_ROWS))
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", lambda *a, **kw: None)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", 60.6, 56.8, client=client)
assert "okn_objects" in block
assert len(block["okn_objects"]) == 2
assert block["okn_objects"][0]["source_id"] == "66-001"
assert block["okn_objects"][1]["category"] == "региональный"
def test_okn_objects_empty_when_none(monkeypatch: Any) -> None:
"""parcel_okn_objects пуст (нет объектов / таблица не задеплоена) → okn_objects == []."""
_patch_overlay(monkeypatch, _OVERLAY)
monkeypatch.setattr(ird_analyze, "parcel_okn_objects", lambda db, wkt: [])
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", lambda *a, **kw: None)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", 60.6, 56.8, client=client)
assert "okn_objects" in block
assert block["okn_objects"] == []
# ---------------------------------------------------------------------------
# Граддокументация (#1061)
# ---------------------------------------------------------------------------
_GRANDDOC_ROWS: list[dict[str, Any]] = [
{
"project_type": "ppt",
"doc_status_name": "Действующий",
"is_active": True,
"paga_number": "2792",
"paga_date": "2021-12-16",
"subject": "Реконструкция ул. Декабристов",
"doc_full_name": "ППТ ул. Декабристов",
}
]
def test_granddoc_wired_to_block(monkeypatch: Any) -> None:
"""parcel_granddoc возвращает запись → ключ granddoc присутствует в block."""
_patch_overlay(monkeypatch, _OVERLAY)
monkeypatch.setattr(ird_analyze, "parcel_granddoc", lambda db, wkt: list(_GRANDDOC_ROWS))
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", lambda *a, **kw: None)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", None, None, client=client)
assert "granddoc" in block
assert len(block["granddoc"]) == 1
row = block["granddoc"][0]
assert row["paga_number"] == "2792"
assert row["is_active"] is True
def test_granddoc_empty_when_no_overlap(monkeypatch: Any) -> None:
"""parcel_granddoc пуст → granddoc == [] (ключ присутствует)."""
_patch_overlay(monkeypatch, {"ird_overlaps": [], "ird_by_kind": {}})
monkeypatch.setattr(ird_analyze, "parcel_granddoc", lambda db, wkt: [])
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", lambda *a, **kw: None)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(object(), None, None, None, client=client)
assert "granddoc" in block
assert block["granddoc"] == []
# ---------------------------------------------------------------------------
# ПАТ-подзоны (#1151, #1065)
# ---------------------------------------------------------------------------
_PAT_ROWS: list[dict[str, Any]] = [
{
"subzone_no": 3,
"name": "Третья подзона",
"restriction": "Высота зданий не более 50 м",
"aerodrome": "Екатеринбург (Кольцово)",
}
]
def test_pat_subzones_wired_to_block(monkeypatch: Any) -> None:
"""parcel_pat_subzones возвращает подзону → block['pat_subzones'] содержит её (#1151)."""
_patch_overlay(monkeypatch, _OVERLAY)
monkeypatch.setattr(ird_analyze, "parcel_pat_subzones", lambda db, wkt: list(_PAT_ROWS))
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", lambda *a, **kw: None)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", None, None, client=client)
assert "pat_subzones" in block
assert len(block["pat_subzones"]) == 1
assert block["pat_subzones"][0]["subzone_no"] == 3
assert block["pat_subzones"][0]["aerodrome"] == "Екатеринбург (Кольцово)"
def test_pat_subzones_empty_when_no_overlap(monkeypatch: Any) -> None:
"""parcel_pat_subzones пуст (нет пересечений / таблица не задеплоена) → pat_subzones == []."""
_patch_overlay(monkeypatch, _OVERLAY)
monkeypatch.setattr(ird_analyze, "parcel_okn_objects", lambda db, wkt: [])
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", lambda *a, **kw: None)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", 60.6, 56.8, client=client)
assert "pat_subzones" in block
assert block["pat_subzones"] == []
# ---------------------------------------------------------------------------
# ТЭП ППТ/ПМТ (#1136 follow-up #1133)
# ---------------------------------------------------------------------------
_PPT_TEP_ROWS: list[dict[str, Any]] = [
{
"project_type": "ppt",
"doc_status_name": "действующий",
"full_name": "ПАГЕ № 22823 от 12.10.2018",
"doc_full_name": "ППТ ул. Челюскинцев",
"source_key": 22823,
"doc_ref": "ppt2018_22823",
"zone_balance": [{"zone_name": "Зона Ж-1", "project_ha_num": 5.0}],
"tep": [{"indicator": "Площадь территории", "target": "36,16"}],
"phasing": [{"phase": "I этап", "zone": "Зона Ж-1", "area_ha_num": 5.0}],
"source_pdf": "https://example.com/ppt22823.pdf",
"fetched_at": "2026-06-07T11:55:00Z",
}
]
def test_ppt_tep_wired_to_block(monkeypatch: Any) -> None:
"""parcel_ppt_tep возвращает запись → ключ ppt_tep присутствует в блоке (#1136)."""
_patch_overlay(monkeypatch, _OVERLAY)
monkeypatch.setattr(ird_analyze, "parcel_ppt_tep", lambda db, wkt: list(_PPT_TEP_ROWS))
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", lambda *a, **kw: None)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", None, None, client=client)
assert "ppt_tep" in block
assert len(block["ppt_tep"]) == 1
row = block["ppt_tep"][0]
assert row["doc_ref"] == "ppt2018_22823"
assert row["zone_balance"][0]["zone_name"] == "Зона Ж-1"
assert row["tep"][0]["target"] == "36,16"
assert row["phasing"][0]["phase"] == "I этап"
def test_ppt_tep_empty_when_no_overlap(monkeypatch: Any) -> None:
"""parcel_ppt_tep пуст (ingest ещё не запускался / нет матча) → ppt_tep == []."""
_patch_overlay(monkeypatch, _OVERLAY)
# _patch_overlay уже мокает parcel_ppt_tep → [], переопределяем явно для ясности.
monkeypatch.setattr(ird_analyze, "parcel_ppt_tep", lambda db, wkt: [])
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", lambda *a, **kw: None)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", 60.6, 56.8, client=client)
assert "ppt_tep" in block
assert block["ppt_tep"] == []
# ---------------------------------------------------------------------------
# Fix 2 (#1850): резолв регламента ОДИН раз — переиспользование в ИРД-блоке
# ---------------------------------------------------------------------------
def test_resolved_zone_regulation_skips_resolver(monkeypatch: Any) -> None:
"""resolved_zone_regulation=<dict> → _zone_regulation_at НЕ зовётся, значение взято as-is."""
_patch_overlay(monkeypatch, _OVERLAY)
resolver_calls: list[int] = []
def _spy(*a: Any, **kw: Any) -> Any:
resolver_calls.append(1)
return {"zone_index": "DIFFERENT"}
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", _spy)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(
object(),
"POINT(60 56)",
60.6,
56.8,
client=client,
resolved_zone_regulation={"zone_index": "Ц-1", "max_far": 4},
)
# Переданное значение использовано напрямую; резолвер ни разу не вызван.
assert block["zone_regulation"] == {"zone_index": "Ц-1", "max_far": 4}
assert resolver_calls == []
def test_resolved_zone_regulation_none_skips_resolver(monkeypatch: Any) -> None:
"""resolved_zone_regulation=None (валидное «зоны нет») → резолвер НЕ зовётся, итог None."""
_patch_overlay(monkeypatch, _OVERLAY)
resolver_calls: list[int] = []
def _spy(*a: Any, **kw: Any) -> Any:
resolver_calls.append(1)
return {"zone_index": "SHOULD_NOT_APPEAR"}
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", _spy)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(
object(),
"POINT(60 56)",
60.6,
56.8,
client=client,
resolved_zone_regulation=None,
)
assert block["zone_regulation"] is None
assert resolver_calls == [] # None — переданное значение, не повод резолвить
def test_unset_resolved_regulation_resolves_itself(monkeypatch: Any) -> None:
"""Параметр НЕ передан (дефолт _UNSET) → блок резолвит сам (path: zoning-flag off, ird on)."""
_patch_overlay(monkeypatch, _OVERLAY)
resolver_calls: list[int] = []
def _spy(db: Any, lon: float, lat: float, client: Any = None) -> dict[str, Any]:
resolver_calls.append(1)
return {"zone_index": "Ж-2"}
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", _spy)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", 60.6, 56.8, client=client)
assert block["zone_regulation"] == {"zone_index": "Ж-2"}
assert resolver_calls == [1] # сам резолвил, ровно один раз
# ---------------------------------------------------------------------------
# Fix 3 (#1850): единый timeout из settings.geoportal_timeout_s
# ---------------------------------------------------------------------------
def test_geoportal_client_uses_settings_timeout(monkeypatch: Any) -> None:
"""Когда клиент не передан, ИРД-блок конструирует EKBGeoportalClient с settings.timeout."""
_patch_overlay(monkeypatch, _OVERLAY)
monkeypatch.setattr(
ird_analyze, "get_or_fetch_zone_regulation", lambda db, lon, lat, client=None: None
)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
monkeypatch.setattr(ird_analyze.settings, "geoportal_timeout_s", 7)
constructed: list[int] = []
class _SpyClient:
def __init__(self, timeout: int = 20) -> None:
constructed.append(timeout)
monkeypatch.setattr(ird_analyze, "EKBGeoportalClient", _SpyClient)
# client=None → блок сам создаёт EKBGeoportalClient(timeout=settings.geoportal_timeout_s).
ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", 60.6, 56.8)
assert constructed == [7]