Light1YT
50555821fb
Some checks failed
CI / changes (push) Successful in 7s
CI / frontend-tests (push) Has been skipped
CI / changes (pull_request) Successful in 7s
CI / frontend-tests (pull_request) Has been skipped
CI / backend-tests (push) Successful in 7m42s
CI / backend-tests (pull_request) Successful in 7m40s
Deploy / changes (push) Has been cancelled
Deploy / build-backend (push) Has been cancelled
Deploy / build-worker (push) Has been cancelled
Deploy / build-frontend (push) Has been cancelled
Deploy / deploy (push) Has been cancelled
#41: per-category routing-radius decay через reusable ORS /matrix client (foot-walk время + per-category пороги + decay-curves), opt-in ?decay=routing, graceful straight-line fallback (analyze не 500 при ORS down) + length-guard durations==destinations. #44: +5 OSM-категорий (transformer/gas/water/sewerage/heat) via Overpass nwr; nearest_{water_main,substation,gas,heat}_m в analyze.utilities. #255bk: ST_AsGeoJSON(CAST(geom AS geometry)) в _get_cad_zouit_overlaps → geom_geojson. Closes #41 Closes #44
638 lines
24 KiB
Python
638 lines
24 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,
|
||
_extract_features_by_layer,
|
||
_get_cad_zouit_overlaps,
|
||
_get_opportunity_parcels,
|
||
_get_red_lines,
|
||
_get_risk_zones,
|
||
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)
|
||
(
|
||
"Охранная зона трубопровода",
|
||
"Трубопровод",
|
||
"ГВС ул. Ленина",
|
||
"66-66-00/123",
|
||
42,
|
||
'{"type":"Polygon","coordinates":[[[60,56],[61,56],[61,57],[60,56]]]}',
|
||
),
|
||
]
|
||
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"] == "ГВС ул. Ленина"
|
||
# 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)
|
||
(
|
||
"Охранная зона ЛЭП 110кВ",
|
||
"Электроснабжение",
|
||
"ВЛ-110",
|
||
"66-123",
|
||
7,
|
||
'{"type":"Polygon","coordinates":[[[0,0],[1,0],[1,1],[0,0]]]}',
|
||
),
|
||
("СЗЗ промышленного объекта", "Санитарная", "Завод", "66-456", 8, 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]
|
||
assert lep["group_key"] == "cad_zouit"
|
||
assert lep["source"] == "cad_zouit"
|
||
assert lep["type_zone"] == "Охранная зона ЛЭП 110кВ"
|
||
assert lep["subcategory"] is None # всегда NULL в cad_zouit
|
||
assert lep["name"] == "ВЛ-110"
|
||
assert lep["raw_props"]["zouit_id"] == 7
|
||
# #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 не валит маппинг
|
||
|
||
|
||
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
|
||
|
||
dump_row = MagicMock()
|
||
dump_row.__getitem__ = lambda self, i: [
|
||
"66:41:0603016", # quarter_cad
|
||
datetime(2026, 1, 1, tzinfo=UTC), # 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)
|
||
("zouit_engineering", {"name": "ЛЭП", "subcategory": 17}),
|
||
]
|
||
|
||
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"
|
||
# dump.available=True т.к. свежий дамп
|
||
assert result["nspd_dump"]["available"] is True
|