Light1YT
285e8f974a
Post-фильтры парсера (по образцу ppt2018_22823): - Табл.11: отбрасываем строки без zone_name (преамбульные артефакты merged-cells давали 2 фантома при 12 реальных зонах). - Табл.13: dedup по (phase, composition, zone, area_ha) + фильтры короткого-без-зоны и нумерационного шума (~364 строк → ~30-40 реальных). Analyze-врезка: новый ppt_tep_lookup.parcel_ppt_tep — JOIN planning_projects ⋈ ekb_ppt_tep по doc_ref↔source_key/doc_full_name (best-effort, без FK). Wired в build_ird_analyze_block рядом с planning_projects/krt_requisites — DB-источники, graceful. Seed-URL остаётся placeholder с # VERIFY (ingest пропускает с WARNING). Tests: 27 в parser (3 новых), 7 lookup, 22 wiring (2 новых). All ruff/syntax green. Closes #1136
101 lines
3.8 KiB
Python
101 lines
3.8 KiB
Python
"""Тесты parcel_ppt_tep (#1136 follow-up #1133) — ТЭП ППТ/ПМТ overlap по участку.
|
||
|
||
БД не дёргается: фейковая сессия с .execute().mappings().all(). Проверяем graceful на
|
||
отсутствие участка/таблицы и проброс строк.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from sqlalchemy.exc import ProgrammingError
|
||
|
||
from app.services.site_finder.ppt_tep_lookup import parcel_ppt_tep
|
||
|
||
|
||
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 _DB:
|
||
def __init__(self, rows: list[dict[str, Any]] | Exception) -> None:
|
||
self._rows = rows
|
||
|
||
def execute(self, sql: Any, params: dict[str, Any]) -> _Result:
|
||
if isinstance(self._rows, Exception):
|
||
raise self._rows
|
||
return _Result(self._rows)
|
||
|
||
|
||
_PPT_TEP_ROW: 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_returns_ppt_tep_overlap() -> None:
|
||
"""Один match по геометрии и doc_ref↔source_key → проброс строки."""
|
||
out = parcel_ppt_tep(_DB([_PPT_TEP_ROW]), "POINT(60 56)") # type: ignore[arg-type]
|
||
assert len(out) == 1
|
||
row = out[0]
|
||
assert row["doc_ref"] == "ppt2018_22823"
|
||
assert row["source_key"] == 22823
|
||
assert row["zone_balance"] == [{"zone_name": "Зона Ж-1", "project_ha_num": 5.0}]
|
||
assert row["tep"][0]["target"] == "36,16"
|
||
assert row["phasing"][0]["zone"] == "Зона Ж-1"
|
||
|
||
|
||
def test_empty_wkt_returns_empty() -> None:
|
||
"""parcel_wkt=None → []."""
|
||
assert parcel_ppt_tep(_DB([]), None) == [] # type: ignore[arg-type]
|
||
|
||
|
||
def test_empty_string_wkt_returns_empty() -> None:
|
||
"""parcel_wkt='' → []."""
|
||
assert parcel_ppt_tep(_DB([]), "") == [] # type: ignore[arg-type]
|
||
|
||
|
||
def test_no_overlap_returns_empty() -> None:
|
||
"""Запрос отработал, но пересечений нет → []."""
|
||
assert parcel_ppt_tep(_DB([]), "POINT(60 56)") == [] # type: ignore[arg-type]
|
||
|
||
|
||
def test_missing_table_is_graceful() -> None:
|
||
"""ekb_ppt_tep ещё не задеплоена → ProgrammingError → [] (analyze не падает)."""
|
||
db = _DB(ProgrammingError("stmt", {}, Exception("relation ekb_ppt_tep does not exist")))
|
||
assert parcel_ppt_tep(db, "POINT(60 56)") == [] # type: ignore[arg-type]
|
||
|
||
|
||
def test_unexpected_exception_is_graceful() -> None:
|
||
"""Любая неожиданная ошибка БД → [] (analyze не падает)."""
|
||
db = _DB(RuntimeError("connection refused"))
|
||
assert parcel_ppt_tep(db, "POINT(60 56)") == [] # type: ignore[arg-type]
|
||
|
||
|
||
def test_multiple_rows_preserved() -> None:
|
||
"""Если match'ит несколько ППТ — все возвращаются."""
|
||
row2: dict[str, Any] = {
|
||
**_PPT_TEP_ROW,
|
||
"doc_ref": "ppt2020_99999",
|
||
"source_key": 99999,
|
||
"full_name": "ПАГЕ № 99999 от 01.06.2020",
|
||
}
|
||
out = parcel_ppt_tep(_DB([_PPT_TEP_ROW, row2]), "POINT(60 56)") # type: ignore[arg-type]
|
||
assert len(out) == 2
|
||
assert {r["doc_ref"] for r in out} == {"ppt2018_22823", "ppt2020_99999"}
|