833 lines
33 KiB
Python
833 lines
33 KiB
Python
"""Тесты для quarter_dump_lookup.py — risk zones + generic layer extraction + TIER 4.
|
||
|
||
Покрывает:
|
||
- _extract_features_by_layer: filter by prefix
|
||
- _get_risk_zones / extract through get_quarter_dump_data: intersect, no-intersect,
|
||
no-risks-in-dump
|
||
- _get_opportunity_parcels: auction/scheme/free/future/oopt layers, distance sort, empty
|
||
- _get_red_lines: intersecting, nearby-only, empty, early-exit
|
||
- EMPTY_DUMP_RESULT / make_empty_result: new keys присутствуют
|
||
Не требует реального PostgreSQL — мокает db.execute().
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
from unittest.mock import MagicMock
|
||
|
||
import pytest
|
||
|
||
from app.services.site_finder.quarter_dump_lookup import (
|
||
EMPTY_DUMP_RESULT,
|
||
_build_structure_characteristics,
|
||
_clean_struct_str,
|
||
_extract_features_by_layer,
|
||
_get_cad_zouit_overlaps,
|
||
_get_engineering_structures_by_boundary,
|
||
_get_opportunity_parcels,
|
||
_get_red_lines,
|
||
_get_risk_zones,
|
||
_get_zouit_overlaps,
|
||
_zouit_subcategory_type,
|
||
derive_quarter_cad,
|
||
get_quarter_dump_data,
|
||
make_empty_result,
|
||
)
|
||
|
||
# ── Fixtures & helpers ────────────────────────────────────────────────────────
|
||
|
||
|
||
def _make_feature(layer: str, geom: dict[str, Any] | None = None) -> dict[str, Any]:
|
||
"""Минимальный feature dict в формате features_json."""
|
||
return {
|
||
"layer": layer,
|
||
"feature_id": "test-id",
|
||
"geometry": geom or {"type": "Polygon", "coordinates": [[[0, 0], [1, 0], [1, 1], [0, 0]]]},
|
||
"properties": {"name": f"test-{layer}", "type_zone": f"zone-{layer}"},
|
||
}
|
||
|
||
|
||
# ── _extract_features_by_layer ────────────────────────────────────────────────
|
||
|
||
|
||
def test_extract_features_by_layer_prefix_match() -> None:
|
||
"""Возвращает только features с нужным prefix."""
|
||
features = [
|
||
_make_feature("risk_flooding"),
|
||
_make_feature("risk_burns"),
|
||
_make_feature("zouit_engineering"),
|
||
_make_feature("parcels"),
|
||
_make_feature("risk_landslide"),
|
||
]
|
||
result = _extract_features_by_layer(features, "risk_")
|
||
assert len(result) == 3
|
||
layers = {f["layer"] for f in result}
|
||
assert layers == {"risk_flooding", "risk_burns", "risk_landslide"}
|
||
|
||
|
||
def test_extract_features_by_layer_zouit_prefix() -> None:
|
||
"""Работает для zouit_ prefix — generic re-use."""
|
||
features = [
|
||
_make_feature("zouit_engineering"),
|
||
_make_feature("zouit_okn"),
|
||
_make_feature("risk_flooding"),
|
||
]
|
||
result = _extract_features_by_layer(features, "zouit_")
|
||
assert len(result) == 2
|
||
assert all(f["layer"].startswith("zouit_") for f in result)
|
||
|
||
|
||
def test_extract_features_by_layer_empty_list() -> None:
|
||
result = _extract_features_by_layer([], "risk_")
|
||
assert result == []
|
||
|
||
|
||
def test_extract_features_by_layer_no_match() -> None:
|
||
features = [_make_feature("parcels"), _make_feature("buildings")]
|
||
result = _extract_features_by_layer(features, "risk_")
|
||
assert result == []
|
||
|
||
|
||
def test_extract_features_by_layer_non_string_layer() -> None:
|
||
"""Gracefully skips features with non-string layer."""
|
||
features: list[dict[str, Any]] = [
|
||
{"layer": None, "feature_id": "x", "geometry": None, "properties": {}},
|
||
_make_feature("risk_flooding"),
|
||
]
|
||
result = _extract_features_by_layer(features, "risk_")
|
||
assert len(result) == 1
|
||
assert result[0]["layer"] == "risk_flooding"
|
||
|
||
|
||
# ── _get_risk_zones ───────────────────────────────────────────────────────────
|
||
|
||
|
||
def _make_mock_db_with_rows(rows: list[tuple[Any, ...]]) -> MagicMock:
|
||
"""Мок Session где execute().fetchall() возвращает rows."""
|
||
db = MagicMock()
|
||
mock_result = MagicMock()
|
||
mock_result.fetchall.return_value = rows
|
||
db.execute.return_value = mock_result
|
||
return db
|
||
|
||
|
||
def test_get_risk_zones_intersects() -> None:
|
||
"""Три risk features → три записи в результате."""
|
||
rows: list[tuple[Any, ...]] = [
|
||
# (layer, props, geom_wkt, intersection_area_sqm)
|
||
(
|
||
"risk_flooding",
|
||
{"type_zone": "Затопление 1%"},
|
||
"POLYGON((0 0,1 0,1 1,0 0))",
|
||
1234.5,
|
||
),
|
||
(
|
||
"risk_burns",
|
||
{"name": "Гарь"},
|
||
"POLYGON((0 0,2 0,2 2,0 0))",
|
||
500.0,
|
||
),
|
||
(
|
||
"risk_landslide",
|
||
{},
|
||
"POLYGON((0 0,3 0,3 3,0 0))",
|
||
999.9,
|
||
),
|
||
]
|
||
db = _make_mock_db_with_rows(rows)
|
||
|
||
result = _get_risk_zones(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
|
||
assert len(result) == 3
|
||
|
||
flooding = next(r for r in result if r["layer"] == "risk_flooding")
|
||
assert flooding["subtype"] == "Затопление 1%"
|
||
assert flooding["intersection_area_sqm"] == 1234.5
|
||
assert flooding["geom_wkt"] == "POLYGON((0 0,1 0,1 1,0 0))"
|
||
|
||
burns = next(r for r in result if r["layer"] == "risk_burns")
|
||
assert burns["subtype"] == "Гарь"
|
||
|
||
landslide = next(r for r in result if r["layer"] == "risk_landslide")
|
||
# No properties name/type_zone → falls back to _RISK_SUBTYPE_LABELS
|
||
assert landslide["subtype"] == "Обвально-осыпные процессы"
|
||
|
||
|
||
def test_get_risk_zones_no_intersect() -> None:
|
||
"""Пустой fetchall → пустой список."""
|
||
db = _make_mock_db_with_rows([])
|
||
result = _get_risk_zones(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
assert result == []
|
||
|
||
|
||
def test_get_risk_zones_no_risks_in_dump() -> None:
|
||
"""risks_count == 0 → early exit, db.execute не вызывается."""
|
||
db = MagicMock()
|
||
layer_counts = {"risks_count": 0}
|
||
result = _get_risk_zones(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))", layer_counts)
|
||
assert result == []
|
||
db.execute.assert_not_called()
|
||
|
||
|
||
def test_get_risk_zones_db_exception_returns_empty() -> None:
|
||
"""DB exception → пустой список (не propagate)."""
|
||
db = MagicMock()
|
||
db.execute.side_effect = Exception("DB connection lost")
|
||
result = _get_risk_zones(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
assert result == []
|
||
|
||
|
||
def test_get_risk_zones_null_area() -> None:
|
||
"""NULL intersection_area_sqm → intersection_area_sqm=None в результате."""
|
||
rows = [("risk_erosion_water", {}, "POINT(0 0)", None)]
|
||
db = _make_mock_db_with_rows(rows)
|
||
result = _get_risk_zones(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
assert len(result) == 1
|
||
assert result[0]["intersection_area_sqm"] is None
|
||
|
||
|
||
# ── make_empty_result includes nspd_risk_zones ────────────────────────────────
|
||
|
||
|
||
def test_make_empty_result_has_risk_zones_key() -> None:
|
||
"""make_empty_result возвращает nspd_risk_zones: []."""
|
||
result = make_empty_result()
|
||
assert "nspd_risk_zones" in result
|
||
assert result["nspd_risk_zones"] == []
|
||
|
||
|
||
# ── _get_opportunity_parcels ─────────────────────────────────────────────────
|
||
|
||
|
||
def test_get_opportunity_parcels_finds_auction() -> None:
|
||
"""3 features (1 auction + 2 other types) — все возвращаются, auction в нужном layer."""
|
||
rows: list[tuple[Any, ...]] = [
|
||
# (layer, props, geom_wkt, distance_m)
|
||
(
|
||
"opportunity_auction_parcels",
|
||
{"cad_num": "66:41:0204016:101"},
|
||
"POLYGON((0 0,1 0,1 1,0 0))",
|
||
50.0,
|
||
),
|
||
(
|
||
"opportunity_scheme_parcels",
|
||
{"cadastral_number": "66:41:0204016:102"},
|
||
"POLYGON((1 1,2 1,2 2,1 1))",
|
||
150.0,
|
||
),
|
||
(
|
||
"opportunity_oopt",
|
||
{},
|
||
"POLYGON((2 2,3 2,3 3,2 2))",
|
||
300.0,
|
||
),
|
||
]
|
||
db = _make_mock_db_with_rows(rows)
|
||
|
||
result = _get_opportunity_parcels(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
|
||
assert len(result) == 3
|
||
|
||
auction = next(r for r in result if r["layer"] == "auction_parcels")
|
||
assert auction["cad_num"] == "66:41:0204016:101"
|
||
assert auction["distance_m"] == 50.0
|
||
|
||
scheme = next(r for r in result if r["layer"] == "scheme_parcels")
|
||
assert scheme["cad_num"] == "66:41:0204016:102"
|
||
|
||
oopt = next(r for r in result if r["layer"] == "oopt")
|
||
assert oopt["cad_num"] is None
|
||
assert oopt["distance_m"] == 300.0
|
||
|
||
|
||
def test_get_opportunity_parcels_distance_sort() -> None:
|
||
"""Результаты отсортированы по distance_m ASC (SQL ORDER BY передаётся DB)."""
|
||
rows: list[tuple[Any, ...]] = [
|
||
# Порядок: ближайший первым (SQL уже сортирует — тест проверяет что порядок сохраняется)
|
||
(
|
||
"opportunity_free_parcels",
|
||
{"cad_num": "66:41:0204016:10"},
|
||
"POINT(0 0)",
|
||
10.0,
|
||
),
|
||
(
|
||
"opportunity_auction_parcels",
|
||
{"cad_num": "66:41:0204016:20"},
|
||
"POINT(1 1)",
|
||
200.0,
|
||
),
|
||
(
|
||
"opportunity_future_parcels",
|
||
{},
|
||
"POINT(2 2)",
|
||
450.0,
|
||
),
|
||
]
|
||
db = _make_mock_db_with_rows(rows)
|
||
|
||
result = _get_opportunity_parcels(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
|
||
assert len(result) == 3
|
||
# Порядок из DB (mock возвращает в порядке rows) — ближайший первый
|
||
assert result[0]["distance_m"] == 10.0
|
||
assert result[1]["distance_m"] == 200.0
|
||
assert result[2]["distance_m"] == 450.0
|
||
|
||
|
||
def test_get_opportunity_parcels_empty() -> None:
|
||
"""Нет opportunity features — пустой список."""
|
||
db = _make_mock_db_with_rows([])
|
||
result = _get_opportunity_parcels(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
assert result == []
|
||
|
||
|
||
def test_get_opportunity_parcels_early_exit() -> None:
|
||
"""opportunity_count == 0 → early exit, db.execute не вызывается."""
|
||
db = MagicMock()
|
||
layer_counts = {"opportunity_count": 0}
|
||
result = _get_opportunity_parcels(
|
||
db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))", layer_counts
|
||
)
|
||
assert result == []
|
||
db.execute.assert_not_called()
|
||
|
||
|
||
def test_get_opportunity_parcels_db_exception_returns_empty() -> None:
|
||
"""DB exception → пустой список (не propagate)."""
|
||
db = MagicMock()
|
||
db.execute.side_effect = Exception("connection lost")
|
||
result = _get_opportunity_parcels(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
assert result == []
|
||
|
||
|
||
# ── _get_red_lines ────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_get_red_lines_intersects() -> None:
|
||
"""Red line intersecting parcel → intersection_length_m filled, distance_m=None."""
|
||
rows: list[tuple[Any, ...]] = [
|
||
# (geom_wkt, does_intersect, intersection_length_m, distance_m)
|
||
(
|
||
"LINESTRING(0 0, 1 1)",
|
||
True,
|
||
45.3, # длина пересечения в метрах
|
||
0.0, # intersecting → distance_m becomes None in result
|
||
),
|
||
]
|
||
db = _make_mock_db_with_rows(rows)
|
||
|
||
result = _get_red_lines(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
|
||
assert len(result) == 1
|
||
r = result[0]
|
||
assert r["geom_wkt"] == "LINESTRING(0 0, 1 1)"
|
||
assert r["intersection_length_m"] == 45.3
|
||
assert r["distance_m"] is None # null when intersecting
|
||
|
||
|
||
def test_get_red_lines_nearby_only() -> None:
|
||
"""Red line nearby only (не intersect) → distance_m filled, intersection_length_m=None."""
|
||
rows: list[tuple[Any, ...]] = [
|
||
# (geom_wkt, does_intersect, intersection_length_m, distance_m)
|
||
(
|
||
"LINESTRING(10 10, 20 20)",
|
||
False,
|
||
0.0, # нет пересечения
|
||
85.5,
|
||
),
|
||
]
|
||
db = _make_mock_db_with_rows(rows)
|
||
|
||
result = _get_red_lines(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
|
||
assert len(result) == 1
|
||
r = result[0]
|
||
assert r["intersection_length_m"] is None
|
||
assert r["distance_m"] == 85.5
|
||
|
||
|
||
def test_get_red_lines_db_exception_returns_empty() -> None:
|
||
"""DB exception → пустой список (не propagate)."""
|
||
db = MagicMock()
|
||
db.execute.side_effect = Exception("DB connection lost")
|
||
result = _get_red_lines(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
assert result == []
|
||
|
||
|
||
def test_get_red_lines_empty() -> None:
|
||
"""Нет red lines features — пустой список."""
|
||
db = _make_mock_db_with_rows([])
|
||
result = _get_red_lines(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
assert result == []
|
||
|
||
|
||
def test_get_red_lines_early_exit() -> None:
|
||
"""red_lines_count == 0 → early exit, db.execute не вызывается."""
|
||
db = MagicMock()
|
||
layer_counts = {"red_lines_count": 0}
|
||
result = _get_red_lines(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))", layer_counts)
|
||
assert result == []
|
||
db.execute.assert_not_called()
|
||
|
||
|
||
# ── EMPTY_DUMP_RESULT / make_empty_result includes new TIER 4 keys ────────────
|
||
|
||
|
||
def test_empty_dump_result_has_opportunity_key() -> None:
|
||
"""EMPTY_DUMP_RESULT содержит nspd_opportunity_parcels: []."""
|
||
assert "nspd_opportunity_parcels" in EMPTY_DUMP_RESULT
|
||
assert EMPTY_DUMP_RESULT["nspd_opportunity_parcels"] == []
|
||
|
||
|
||
def test_empty_dump_result_has_red_lines_key() -> None:
|
||
"""EMPTY_DUMP_RESULT содержит nspd_red_lines: []."""
|
||
assert "nspd_red_lines" in EMPTY_DUMP_RESULT
|
||
assert EMPTY_DUMP_RESULT["nspd_red_lines"] == []
|
||
|
||
|
||
def test_make_empty_result_has_tier4_keys() -> None:
|
||
"""make_empty_result() возвращает nspd_opportunity_parcels и nspd_red_lines."""
|
||
result = make_empty_result()
|
||
assert "nspd_opportunity_parcels" in result
|
||
assert result["nspd_opportunity_parcels"] == []
|
||
assert "nspd_red_lines" in result
|
||
assert result["nspd_red_lines"] == []
|
||
|
||
|
||
# ── Issue #234: harvest_eta_seconds + SETNX dedupe ────────────────────────────
|
||
|
||
|
||
def test_empty_dump_result_has_harvest_eta_seconds_key() -> None:
|
||
"""EMPTY_DUMP_RESULT.nspd_dump содержит harvest_eta_seconds (issue #234)."""
|
||
assert "harvest_eta_seconds" in EMPTY_DUMP_RESULT["nspd_dump"]
|
||
assert EMPTY_DUMP_RESULT["nspd_dump"]["harvest_eta_seconds"] is None
|
||
|
||
|
||
def test_make_empty_result_passes_harvest_eta_seconds() -> None:
|
||
"""make_empty_result(harvest_eta_seconds=60) → пробрасывает в nspd_dump."""
|
||
result = make_empty_result(harvest_triggered=True, harvest_eta_seconds=60)
|
||
assert result["nspd_dump"]["harvest_eta_seconds"] == 60
|
||
assert result["nspd_dump"]["harvest_triggered"] is True
|
||
|
||
|
||
def test_make_empty_result_default_eta_is_none() -> None:
|
||
"""Если harvest_eta_seconds не передан — поле None (для available=False случая)."""
|
||
result = make_empty_result()
|
||
assert result["nspd_dump"]["harvest_eta_seconds"] is None
|
||
|
||
|
||
def test_acquire_harvest_lock_succeeds_first_call(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""SETNX lock: первый вызов получает lock (True), второй — нет (False).
|
||
|
||
Issue #234: burst N concurrent analyze на один свежетриггеренный квартал
|
||
дедуплицируется через Redis SETNX. Мокаем redis.Redis.from_url.
|
||
"""
|
||
from app.services.site_finder import quarter_dump_lookup as qdl
|
||
|
||
# Mock redis: первый client.set возвращает True, второй — None (key existed).
|
||
calls = {"n": 0}
|
||
|
||
class _MockRedis:
|
||
def set(self, key: str, value: str, *, nx: bool, ex: int) -> bool | None:
|
||
calls["n"] += 1
|
||
return True if calls["n"] == 1 else None
|
||
|
||
monkeypatch.setattr(
|
||
"redis.Redis.from_url",
|
||
lambda _url: _MockRedis(),
|
||
)
|
||
|
||
assert qdl._acquire_harvest_lock("66:41:0204016") is True
|
||
assert qdl._acquire_harvest_lock("66:41:0204016") is False
|
||
assert calls["n"] == 2
|
||
|
||
|
||
def test_acquire_harvest_lock_redis_unavailable_returns_false(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""Redis недоступен → _acquire_harvest_lock возвращает False (graceful).
|
||
|
||
Лучше не запустить дубль, чем нагрузить WAF при transient redis-fail.
|
||
"""
|
||
from app.services.site_finder import quarter_dump_lookup as qdl
|
||
|
||
def _boom(_url: str) -> None:
|
||
raise ConnectionError("redis down")
|
||
|
||
monkeypatch.setattr("redis.Redis.from_url", _boom)
|
||
assert qdl._acquire_harvest_lock("66:41:0204016") is False
|
||
|
||
|
||
# ── derive_quarter_cad (smoke tests) ──────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"cad,expected",
|
||
[
|
||
("66:41:0204016", "66:41:0204016"),
|
||
("66:41:0204016:10", "66:41:0204016"),
|
||
("66:41:0204016:10:1", "66:41:0204016"),
|
||
("invalid", None),
|
||
("66:41", None),
|
||
],
|
||
)
|
||
def test_derive_quarter_cad(cad: str, expected: str | None) -> None:
|
||
assert derive_quarter_cad(cad) == expected
|
||
|
||
|
||
# ── #243: cad_zouit fallback when dump absent ─────────────────────────────────
|
||
|
||
|
||
def _make_mock_db_no_dump_with_cad_zouit(
|
||
cad_zouit_rows: list[tuple[Any, ...]],
|
||
) -> MagicMock:
|
||
"""Мок Session: nspd_quarter_dumps.first() → None; cad_zouit.fetchall() → rows."""
|
||
db = MagicMock()
|
||
first_result = MagicMock()
|
||
first_result.first.return_value = None # нет дампа
|
||
|
||
fetchall_result = MagicMock()
|
||
fetchall_result.fetchall.return_value = cad_zouit_rows
|
||
|
||
# Первый вызов execute (SELECT nspd_quarter_dumps) → first()=None
|
||
# Второй вызов execute (SELECT cad_zouit) → fetchall()=rows
|
||
db.execute.side_effect = [first_result, fetchall_result]
|
||
return db
|
||
|
||
|
||
def test_cad_zouit_fallback_fires_when_no_dump() -> None:
|
||
"""#243: cad_zouit fallback срабатывает когда nspd_quarter_dumps пуст глобально.
|
||
|
||
Ожидаем: nspd_zouit_overlaps непустой, source='cad_zouit'.
|
||
"""
|
||
cad_zouit_rows: list[tuple[Any, ...]] = [
|
||
# (type_zone, category_name, name_by_doc, reg_numb_border, id, geom_geojson, coverage_pct)
|
||
(
|
||
"Охранная зона трубопровода",
|
||
"Трубопровод",
|
||
"ГВС ул. Ленина",
|
||
"66-66-00/123",
|
||
42,
|
||
'{"type":"Polygon","coordinates":[[[60,56],[61,56],[61,57],[60,56]]]}',
|
||
0.15,
|
||
),
|
||
]
|
||
db = _make_mock_db_no_dump_with_cad_zouit(cad_zouit_rows)
|
||
|
||
wkt = "POLYGON((60 56,61 56,61 57,60 57,60 56))"
|
||
result = get_quarter_dump_data(db, "66:41:0603016:194", wkt)
|
||
|
||
overlaps: list[dict[str, Any]] = result["nspd_zouit_overlaps"]
|
||
assert len(overlaps) == 1
|
||
assert overlaps[0]["source"] == "cad_zouit"
|
||
assert overlaps[0]["type_zone"] == "Охранная зона трубопровода"
|
||
assert overlaps[0]["name"] == "ГВС ул. Ленина"
|
||
assert overlaps[0]["coverage_pct"] == 0.15
|
||
# dump.available остаётся False (нет дампа)
|
||
assert result["nspd_dump"]["available"] is False
|
||
|
||
|
||
def test_cad_zouit_fallback_no_dump_no_parcel_wkt() -> None:
|
||
"""#243: dump=None + parcel_wkt=None → overlaps пустой, не вызываем cad_zouit."""
|
||
db = MagicMock()
|
||
first_result = MagicMock()
|
||
first_result.first.return_value = None
|
||
db.execute.return_value = first_result
|
||
|
||
result = get_quarter_dump_data(db, "66:41:0603016:194", None)
|
||
|
||
assert result["nspd_zouit_overlaps"] == []
|
||
# cad_zouit SELECT не должен вызываться (только один execute для dump lookup)
|
||
assert db.execute.call_count == 1
|
||
|
||
|
||
def test_cad_zouit_fallback_empty_when_no_overlaps() -> None:
|
||
"""#243: dump=None + cad_zouit возвращает 0 строк → nspd_zouit_overlaps пуст."""
|
||
db = _make_mock_db_no_dump_with_cad_zouit([])
|
||
|
||
wkt = "POLYGON((60 56,61 56,61 57,60 57,60 56))"
|
||
result = get_quarter_dump_data(db, "66:41:0603016:194", wkt)
|
||
|
||
assert result["nspd_zouit_overlaps"] == []
|
||
|
||
|
||
def test_get_cad_zouit_overlaps_returns_correct_schema() -> None:
|
||
"""_get_cad_zouit_overlaps: поля group_key/layer/source/type_zone корректны."""
|
||
rows: list[tuple[Any, ...]] = [
|
||
# (type_zone, category_name, name_by_doc, reg_numb_border, id, geom_geojson, coverage_pct)
|
||
(
|
||
"Охранная зона ЛЭП 110кВ",
|
||
"Электроснабжение",
|
||
"ВЛ-110",
|
||
"66-123",
|
||
7,
|
||
'{"type":"Polygon","coordinates":[[[0,0],[1,0],[1,1],[0,0]]]}',
|
||
0.42,
|
||
),
|
||
("СЗЗ промышленного объекта", "Санитарная", "Завод", "66-456", 8, None, None),
|
||
]
|
||
db = _make_mock_db_with_rows(rows)
|
||
|
||
result = _get_cad_zouit_overlaps(db, "POLYGON((0 0,1 0,1 1,0 0))")
|
||
|
||
assert len(result) == 2
|
||
lep = result[0]
|
||
# #1957: group_key унифицирован на 'protected' (был сырой 'cad_zouit' без RU-метки).
|
||
assert lep["group_key"] == "protected"
|
||
# source остаётся 'cad_zouit' — gate_verdict.py ветвится по нему (keyword-классификация).
|
||
assert lep["source"] == "cad_zouit"
|
||
assert lep["type_zone"] == "Охранная зона ЛЭП 110кВ"
|
||
assert lep["subcategory"] is None # всегда NULL в cad_zouit
|
||
assert lep["name"] == "ВЛ-110"
|
||
# #1957: top-level reg_numb_border — единый shape с nspd-dump path.
|
||
assert lep["reg_numb_border"] == "66-123"
|
||
assert lep["raw_props"]["zouit_id"] == 7
|
||
# area-gate: coverage_pct проброшен (0..1) и нормализован
|
||
assert lep["coverage_pct"] == 0.42
|
||
# #255: geom_geojson проброшен для ZouitLayer (string из ST_AsGeoJSON)
|
||
assert "geom_geojson" in lep
|
||
assert lep["geom_geojson"].startswith('{"type":"Polygon"')
|
||
assert result[1]["geom_geojson"] is None # NULL geom не валит маппинг
|
||
assert result[1]["coverage_pct"] is None # NULL coverage (делитель 0) → None
|
||
|
||
|
||
# ── #1957: NSPD subcategory → human type map + dump-path enrichment ───────────
|
||
|
||
|
||
def test_zouit_subcategory_type_known_codes() -> None:
|
||
"""Известный subcategory-код → (int, RU-тип). #1957 СЗЗ=26 обязателен."""
|
||
assert _zouit_subcategory_type(26) == (26, "Санитарно-защитная зона")
|
||
assert _zouit_subcategory_type("26") == (26, "Санитарно-защитная зона") # str из dump
|
||
assert _zouit_subcategory_type(17) == (17, "Охранная зона инженерных коммуникаций")
|
||
|
||
|
||
def test_zouit_subcategory_type_unknown_and_invalid() -> None:
|
||
"""Неизвестный/невалидный код → тип None (graceful), int нормализован если возможно."""
|
||
assert _zouit_subcategory_type(9999) == (9999, None) # код вне карты
|
||
assert _zouit_subcategory_type("garbage") == (None, None) # не int
|
||
assert _zouit_subcategory_type(None) == (None, None)
|
||
|
||
|
||
def test_dump_zouit_overlap_maps_subcategory_26_to_szz() -> None:
|
||
"""#1957: dump-фича с subcategory=26 (СЗЗ) и БЕЗ name/type_zone больше не blank.
|
||
|
||
type_zone заполняется из карты, reg читается из props.options.reg_numb_border,
|
||
group_key унифицирован ('protected'), name fallback → type_zone.
|
||
"""
|
||
props = {
|
||
"subcategory": 26,
|
||
"descr": "66:41-6.15834",
|
||
"label": "66:41-6.15834",
|
||
"options": {"reg_numb_border": "66:41-6.15834"},
|
||
}
|
||
row = MagicMock()
|
||
row.__getitem__ = lambda self, i: {0: "zouit_protected", 1: props, 2: 0.12}[i]
|
||
|
||
db = MagicMock()
|
||
exec_result = MagicMock()
|
||
exec_result.fetchall.return_value = [row]
|
||
db.execute.return_value = exec_result
|
||
|
||
# layer_counts с zouit_count>0 → идём по dump-path (не cad_zouit fallback).
|
||
overlaps = _get_zouit_overlaps(
|
||
db, "66:41:0205010", "POLYGON((0 0,1 0,1 1,0 0))", {"zouit_count": 3}
|
||
)
|
||
assert len(overlaps) == 1
|
||
o = overlaps[0]
|
||
assert o["group_key"] == "protected"
|
||
assert o["subcategory"] == 26 # нормализован к int
|
||
assert o["type_zone"] == "Санитарно-защитная зона" # из карты (был null → blank)
|
||
assert o["name"] == "Санитарно-защитная зона" # name отсутствовал → fallback на тип
|
||
assert o["reg_numb_border"] == "66:41-6.15834" # из options
|
||
assert o["source"] == "nspd-quarter-dump"
|
||
|
||
|
||
def test_dump_zouit_overlap_unknown_subcategory_graceful() -> None:
|
||
"""Неизвестный subcategory-код → type_zone None, но фактор не теряется."""
|
||
props = {"subcategory": 9999, "options": {"reg_numb_border": "66:41-9.99"}}
|
||
row = MagicMock()
|
||
row.__getitem__ = lambda self, i: {0: "zouit_other", 1: props, 2: None}[i]
|
||
|
||
db = MagicMock()
|
||
exec_result = MagicMock()
|
||
exec_result.fetchall.return_value = [row]
|
||
db.execute.return_value = exec_result
|
||
|
||
overlaps = _get_zouit_overlaps(
|
||
db, "66:41:0205010", "POLYGON((0 0,1 0,1 1,0 0))", {"zouit_count": 1}
|
||
)
|
||
assert len(overlaps) == 1
|
||
assert overlaps[0]["group_key"] == "other"
|
||
assert overlaps[0]["subcategory"] == 9999
|
||
assert overlaps[0]["type_zone"] is None
|
||
assert overlaps[0]["reg_numb_border"] == "66:41-9.99"
|
||
|
||
|
||
def test_cad_zouit_overlaps_distinct_on_in_sql() -> None:
|
||
"""#1957: cad_zouit SELECT использует DISTINCT ON (reg_numb_border) для дедупа.
|
||
|
||
Дубль-строки (одна физ. зона 2× с разным category_name) схлопывает Postgres
|
||
через DISTINCT ON — проверяем, что SQL действительно его содержит (мок не
|
||
выполняет SQL, поэтому верифицируем форму запроса)."""
|
||
db = MagicMock()
|
||
exec_result = MagicMock()
|
||
exec_result.fetchall.return_value = []
|
||
db.execute.return_value = exec_result
|
||
|
||
_get_cad_zouit_overlaps(db, "POLYGON((0 0,1 0,1 1,0 0))")
|
||
|
||
sql_text = str(db.execute.call_args[0][0])
|
||
assert "DISTINCT ON (reg_numb_border)" in sql_text
|
||
assert "ORDER BY reg_numb_border, id" in sql_text
|
||
|
||
|
||
def test_nspd_path_used_when_zouit_count_nonzero() -> None:
|
||
"""#243 backward compat: когда dump существует с zouit_count>0 — NSPD path, не cad_zouit.
|
||
|
||
Мок: row.zouit_count=3, fetchall возвращает NSPD features.
|
||
Проверяем source='nspd-quarter-dump' (не cad_zouit).
|
||
"""
|
||
# Строка дампа: все поля через MagicMock
|
||
from datetime import UTC, datetime, timedelta
|
||
|
||
dump_row = MagicMock()
|
||
dump_row.__getitem__ = lambda self, i: [
|
||
"66:41:0603016", # quarter_cad
|
||
# Свежесть ОТНОСИТЕЛЬНО now: был хардкод datetime(2026, 1, 1) → date-bomb
|
||
# (с 2026-06-30 перевалил _DUMP_MAX_AGE_DAYS=180 → stale-ветка → _trigger_harvest
|
||
# → _empty_unavailable → overlaps=[] → assert 0 == 1). now − 1 день всегда свеж.
|
||
datetime.now(UTC) - timedelta(days=1), # fetched_at_utc (свежий)
|
||
100, # total_features
|
||
None, # harvest_error
|
||
1, # territorial_zones_count
|
||
3, # zouit_count > 0 → NSPD path
|
||
0, # engineering_count
|
||
0, # risks_count
|
||
0, # opportunity_count
|
||
0, # red_lines_count
|
||
][i]
|
||
|
||
nspd_zouit_rows: list[tuple[Any, ...]] = [
|
||
# (layer, props, coverage_pct)
|
||
("zouit_engineering", {"name": "ЛЭП", "subcategory": 17}, 0.33),
|
||
]
|
||
|
||
db = MagicMock()
|
||
first_result = MagicMock()
|
||
first_result.first.return_value = dump_row
|
||
|
||
# Последующие вызовы execute (spatial queries) — возвращаем пустые кроме zouit
|
||
zouit_result = MagicMock()
|
||
zouit_result.fetchall.return_value = nspd_zouit_rows
|
||
|
||
empty_result = MagicMock()
|
||
empty_result.fetchall.return_value = []
|
||
empty_result.first.return_value = None
|
||
|
||
# [0]=dumps lookup, [1]=zoning(first), [2]=zouit(fetchall), [3+]=others
|
||
db.execute.side_effect = [
|
||
first_result,
|
||
empty_result, # _get_zoning first()
|
||
zouit_result, # _get_zouit_overlaps fetchall()
|
||
empty_result, # _get_engineering_nearby fetchall()
|
||
empty_result, # _get_risk_zones fetchall()
|
||
empty_result, # _get_opportunity_parcels fetchall()
|
||
empty_result, # _get_red_lines fetchall()
|
||
]
|
||
|
||
result = get_quarter_dump_data(
|
||
db, "66:41:0603016:194", "POLYGON((60 56,61 56,61 57,60 57,60 56))"
|
||
)
|
||
|
||
overlaps: list[dict[str, Any]] = result["nspd_zouit_overlaps"]
|
||
assert len(overlaps) == 1
|
||
assert overlaps[0]["source"] == "nspd-quarter-dump"
|
||
# area-gate: coverage_pct проброшен из SQL (r[2]) и нормализован
|
||
assert overlaps[0]["coverage_pct"] == 0.33
|
||
# dump.available=True т.к. свежий дамп
|
||
assert result["nspd_dump"]["available"] is True
|
||
|
||
|
||
# ── Engineering-structure options.* extraction (cat 36328, network popups) ──────
|
||
|
||
|
||
def test_clean_struct_str_normalizes_and_drops_empty() -> None:
|
||
assert _clean_struct_str(" КЛ 10кВ ") == "КЛ 10кВ"
|
||
assert _clean_struct_str("-") is None
|
||
assert _clean_struct_str("") is None
|
||
assert _clean_struct_str(None) is None
|
||
assert _clean_struct_str(2006) == "2006"
|
||
|
||
|
||
def test_build_structure_characteristics_full() -> None:
|
||
# Год · протяжённость · собственность из options НСПД-дампа.
|
||
opts = {
|
||
"params_year_built": "2006",
|
||
"params_extension": 192,
|
||
"ownership_type": "Частная",
|
||
}
|
||
assert (
|
||
_build_structure_characteristics(opts)
|
||
== "2006 г. · протяжённость 192 м · частная собственность"
|
||
)
|
||
|
||
|
||
def test_build_structure_characteristics_partial_and_empty() -> None:
|
||
# Только год → одна часть; нулевая протяжённость и пустые поля отбрасываются.
|
||
assert _build_structure_characteristics({"params_year_built": "1971"}) == "1971 г."
|
||
assert (
|
||
_build_structure_characteristics(
|
||
{"params_extension": 0, "ownership_type": "-", "params_year_built": ""}
|
||
)
|
||
is None
|
||
)
|
||
assert _build_structure_characteristics({}) is None
|
||
|
||
|
||
# ── #2111: engineering-structure purpose (dedup-vs-type) ───────────────────────
|
||
|
||
|
||
def _eng_structures(opts: dict[str, Any]) -> list[dict[str, Any]]:
|
||
"""Прогнать _get_engineering_structures_by_boundary на одной строке с данными opts."""
|
||
geom = '{"type":"Point","coordinates":[0,0]}'
|
||
db = _make_mock_db_with_rows([({"options": opts}, geom, 12.3)])
|
||
return _get_engineering_structures_by_boundary(
|
||
db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))", 500
|
||
)
|
||
|
||
|
||
def test_engineering_purpose_set_when_differs_from_type() -> None:
|
||
# params_name → type; params_purpose отличается → purpose проброшен.
|
||
result = _eng_structures(
|
||
{
|
||
"params_name": "КЛ 10кВ ТП 1077",
|
||
"params_purpose": "1.1. Сооружения электроэнергетики",
|
||
"object_type_value": "Сооружение",
|
||
}
|
||
)
|
||
assert len(result) == 1
|
||
s = result[0]
|
||
# type берётся из params_purpose (первый в fallback-цепочке ptype).
|
||
assert s["type"] == "1.1. Сооружения электроэнергетики"
|
||
# purpose == type → дедуп → None (purpose первым тоже params_purpose).
|
||
assert s["purpose"] is None
|
||
|
||
|
||
def test_engineering_purpose_dedup_when_equal_to_type() -> None:
|
||
# Нет params_name → type = params_purpose; purpose тоже = params_purpose → дедуп None.
|
||
result = _eng_structures({"params_purpose": "Сооружение", "object_type_value": "Сооружение"})
|
||
assert result[0]["type"] == "Сооружение"
|
||
assert result[0]["purpose"] is None
|
||
|
||
|
||
def test_engineering_purpose_from_object_type_value_distinct() -> None:
|
||
# Нет params_purpose → type = object_type_value; purpose тоже object_type_value → дедуп.
|
||
result = _eng_structures({"params_name": "ВЛ-110", "object_type_value": "Сооружение"})
|
||
s = result[0]
|
||
assert s["type"] == "Сооружение" # object_type_value (params_purpose отсутствует)
|
||
assert s["purpose"] is None # purpose == type → дедуп
|
||
|
||
|
||
def test_engineering_purpose_none_when_no_options() -> None:
|
||
result = _eng_structures({})
|
||
assert result[0]["purpose"] is None
|