501 lines
20 KiB
Python
501 lines
20 KiB
Python
"""Unit tests для utility_infrastructure_loader — OSM инженерные сети (#1746).
|
||
|
||
Mock-based / pure — без реального Overpass и БД. Покрывает:
|
||
- _build_overpass_query: node / way / nwr варианты + bbox ЕКБ
|
||
- _passes_tag_filters: substance-фильтр (man_made=pipeline) + tower:type (man_made=tower)
|
||
- _upsert_elements: парсинг way→LineString / node→Point, классификация kind,
|
||
per-element SAVEPOINT (битый элемент не валит батч), per-kind counts
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from contextlib import contextmanager
|
||
from typing import Any
|
||
|
||
import httpx
|
||
|
||
from app.services.site_finder import utility_infrastructure_loader as ul
|
||
|
||
# ── _build_overpass_query ─────────────────────────────────────────────────────
|
||
|
||
|
||
def test_build_query_node() -> None:
|
||
q = ul._build_overpass_query("power", "tower", "node")
|
||
assert q.startswith("[out:json][timeout:30];")
|
||
assert 'node["power"="tower"]' in q
|
||
assert "out body;" in q
|
||
|
||
|
||
def test_build_query_way_uses_out_geom() -> None:
|
||
"""way → out geom (нужны координаты узлов для LineString ЛЭП/трубы)."""
|
||
q = ul._build_overpass_query("power", "line", "way")
|
||
assert 'way["power"="line"]' in q
|
||
assert "out geom;" in q
|
||
|
||
|
||
def test_build_query_nwr_for_points() -> None:
|
||
"""nwr ловит и node, и way (точка/площадка подстанции), с out geom."""
|
||
q = ul._build_overpass_query("power", "substation", "nwr")
|
||
assert 'nwr["power"="substation"]' in q
|
||
assert "out geom;" in q
|
||
|
||
|
||
def test_build_query_bbox_present() -> None:
|
||
"""Bbox ЕКБ присутствует во всех вариантах запроса."""
|
||
q = ul._build_overpass_query("pipeline", "gas", "way")
|
||
assert "(56.7,60.5,56.95,60.75)" in q
|
||
|
||
|
||
# ── _UTILITY_QUERIES coverage ─────────────────────────────────────────────────
|
||
|
||
|
||
def test_all_required_kinds_present() -> None:
|
||
"""Все 6 видов сети покрыты запросами."""
|
||
kinds = {kind for _k, _v, _t, kind in ul._UTILITY_QUERIES}
|
||
for expected in ("power", "water", "gas", "heat", "communication", "sewage"):
|
||
assert expected in kinds, f"вид {expected!r} должен быть в _UTILITY_QUERIES"
|
||
|
||
|
||
def test_power_line_is_way_and_tower_is_node() -> None:
|
||
"""ЛЭП = way (линия), опора = node (точка)."""
|
||
by_kv = {(k, v): t for k, v, t, _ in ul._UTILITY_QUERIES}
|
||
assert by_kv[("power", "line")] == "way"
|
||
assert by_kv[("power", "tower")] == "node"
|
||
assert by_kv[("power", "substation")] == "nwr"
|
||
|
||
|
||
# ── _passes_tag_filters ───────────────────────────────────────────────────────
|
||
|
||
|
||
def test_pipeline_substance_gas_passes() -> None:
|
||
assert ul._passes_tag_filters("man_made", "pipeline", {"substance": "gas"})
|
||
|
||
|
||
def test_pipeline_substance_non_gas_filtered() -> None:
|
||
"""man_made=pipeline с substance=oil НЕ проходит под gas."""
|
||
assert not ul._passes_tag_filters("man_made", "pipeline", {"substance": "oil"})
|
||
|
||
|
||
def test_pipeline_legacy_type_tag_fallback() -> None:
|
||
"""legacy `type=gas` (без substance) тоже распознаётся."""
|
||
assert ul._passes_tag_filters("man_made", "pipeline", {"type": "gas"})
|
||
|
||
|
||
def test_tower_communication_passes() -> None:
|
||
assert ul._passes_tag_filters("man_made", "tower", {"tower:type": "communication"})
|
||
|
||
|
||
def test_tower_non_communication_filtered() -> None:
|
||
"""Водонапорная башня (tower:type=water) НЕ проходит под communication."""
|
||
assert not ul._passes_tag_filters("man_made", "tower", {"tower:type": "water"})
|
||
|
||
|
||
def test_unfiltered_key_always_passes() -> None:
|
||
"""Ключи без substance/tower фильтра проходят как есть."""
|
||
assert ul._passes_tag_filters("power", "line", {})
|
||
|
||
|
||
# ── _upsert_elements (fake DB) ────────────────────────────────────────────────
|
||
|
||
|
||
class _Scalar:
|
||
def __init__(self, value: Any) -> None:
|
||
self._value = value
|
||
|
||
def scalar(self) -> Any:
|
||
return self._value
|
||
|
||
|
||
class _FakeDB:
|
||
"""Мини-фейк SQLAlchemy session: пишет executed, отдаёт is_insert через scalar()."""
|
||
|
||
def __init__(self, *, fail_ids: set[int] | None = None) -> None:
|
||
self.executed: list[tuple[Any, dict[str, Any] | None]] = []
|
||
self.committed = False
|
||
self.closed = False
|
||
self.rolled_back = False
|
||
self._fail_ids = fail_ids or set()
|
||
|
||
@contextmanager
|
||
def begin_nested(self) -> Any:
|
||
yield
|
||
|
||
def execute(self, sql: Any, params: dict[str, Any] | None = None) -> Any:
|
||
self.executed.append((sql, params))
|
||
if params is not None and params.get("osm_id") in self._fail_ids:
|
||
raise RuntimeError("bad geometry")
|
||
return _Scalar(True) # is_insert=True
|
||
|
||
def commit(self) -> None:
|
||
self.committed = True
|
||
|
||
def rollback(self) -> None:
|
||
self.rolled_back = True
|
||
|
||
def close(self) -> None:
|
||
self.closed = True
|
||
|
||
|
||
def _way(osm_id: int, kind: str, key: str, value: str, nodes: list[dict]) -> dict:
|
||
return {
|
||
"id": osm_id,
|
||
"type": "way",
|
||
"geometry": nodes,
|
||
"tags": {"name": f"way-{osm_id}"},
|
||
"_infrastructure_kind": kind,
|
||
"_source_tag": f"{key}={value}",
|
||
"_query_key": key,
|
||
"_query_value": value,
|
||
}
|
||
|
||
|
||
def _node(osm_id: int, kind: str, key: str, value: str, tags: dict | None = None) -> dict:
|
||
return {
|
||
"id": osm_id,
|
||
"type": "node",
|
||
"lat": 56.83,
|
||
"lon": 60.6,
|
||
"tags": tags or {"name": f"node-{osm_id}"},
|
||
"_infrastructure_kind": kind,
|
||
"_source_tag": f"{key}={value}",
|
||
"_query_key": key,
|
||
"_query_value": value,
|
||
}
|
||
|
||
|
||
_TWO_NODES = [{"lat": 56.8, "lon": 60.6}, {"lat": 56.81, "lon": 60.61}]
|
||
|
||
|
||
def test_upsert_mixed_power_water_gas(monkeypatch: Any) -> None:
|
||
"""Power line (way), substation (node), water_works (node), gas pipeline (way)."""
|
||
fake_db = _FakeDB()
|
||
monkeypatch.setattr(ul, "SessionLocal", lambda: fake_db)
|
||
|
||
elements = [
|
||
_way(1, "power", "power", "line", _TWO_NODES),
|
||
_node(2, "power", "power", "substation"),
|
||
_node(3, "water", "man_made", "water_works"),
|
||
_way(4, "gas", "pipeline", "gas", _TWO_NODES),
|
||
]
|
||
result = ul._upsert_elements(elements)
|
||
|
||
assert result["fetched"] == 4
|
||
assert result["inserted"] == 4
|
||
assert result["skipped"] == 0
|
||
assert result["kind_power"] == 2
|
||
assert result["kind_water"] == 1
|
||
assert result["kind_gas"] == 1
|
||
assert fake_db.committed
|
||
assert fake_db.closed
|
||
assert len(fake_db.executed) == 4
|
||
|
||
|
||
def test_upsert_way_geometry_is_linestring(monkeypatch: Any) -> None:
|
||
"""way → WKT LINESTRING передаётся в params (geom-build из noise_loader)."""
|
||
fake_db = _FakeDB()
|
||
monkeypatch.setattr(ul, "SessionLocal", lambda: fake_db)
|
||
|
||
ul._upsert_elements([_way(1, "power", "power", "line", _TWO_NODES)])
|
||
_, params = fake_db.executed[0]
|
||
assert params is not None
|
||
assert params["wkt"] == "LINESTRING(60.6 56.8, 60.61 56.81)"
|
||
assert params["infrastructure_kind"] == "power"
|
||
assert params["source_tag"] == "power=line"
|
||
|
||
|
||
def test_upsert_node_geometry_is_point(monkeypatch: Any) -> None:
|
||
"""node → lat/lon передаются для ST_MakePoint."""
|
||
fake_db = _FakeDB()
|
||
monkeypatch.setattr(ul, "SessionLocal", lambda: fake_db)
|
||
|
||
ul._upsert_elements([_node(2, "power", "power", "substation")])
|
||
_, params = fake_db.executed[0]
|
||
assert params is not None
|
||
assert params["lat"] == 56.83
|
||
assert params["lon"] == 60.6
|
||
assert "wkt" not in params
|
||
|
||
|
||
def test_upsert_skips_way_with_too_few_nodes(monkeypatch: Any) -> None:
|
||
"""way с <2 узлами (битый bbox-обрезок) пропускается, не валит батч."""
|
||
fake_db = _FakeDB()
|
||
monkeypatch.setattr(ul, "SessionLocal", lambda: fake_db)
|
||
|
||
elements = [
|
||
_way(1, "power", "power", "line", [{"lat": 56.8, "lon": 60.6}]), # 1 узел
|
||
_node(2, "power", "power", "substation"),
|
||
]
|
||
result = ul._upsert_elements(elements)
|
||
assert result["skipped"] == 1
|
||
assert result["inserted"] == 1
|
||
assert len(fake_db.executed) == 1 # только node дошёл до UPSERT
|
||
|
||
|
||
def test_upsert_substance_filter_skips_non_gas_pipeline(monkeypatch: Any) -> None:
|
||
"""man_made=pipeline substance=oil пропускается (не газ)."""
|
||
fake_db = _FakeDB()
|
||
monkeypatch.setattr(ul, "SessionLocal", lambda: fake_db)
|
||
|
||
oil = _way(1, "gas", "man_made", "pipeline", _TWO_NODES)
|
||
oil["tags"] = {"substance": "oil"}
|
||
gas = _way(2, "gas", "man_made", "pipeline", _TWO_NODES)
|
||
gas["tags"] = {"substance": "gas"}
|
||
|
||
result = ul._upsert_elements([oil, gas])
|
||
assert result["skipped"] == 1
|
||
assert result["inserted"] == 1
|
||
assert result["kind_gas"] == 1
|
||
|
||
|
||
def test_upsert_savepoint_isolates_bad_element(monkeypatch: Any) -> None:
|
||
"""Битый элемент (execute бросает) → SAVEPOINT откатывает только его, батч жив."""
|
||
fake_db = _FakeDB(fail_ids={2})
|
||
monkeypatch.setattr(ul, "SessionLocal", lambda: fake_db)
|
||
|
||
elements = [
|
||
_node(1, "power", "power", "substation"),
|
||
_node(2, "power", "power", "substation"), # упадёт
|
||
_node(3, "water", "man_made", "water_works"),
|
||
]
|
||
result = ul._upsert_elements(elements)
|
||
assert result["inserted"] == 2
|
||
assert result["skipped"] == 1
|
||
assert fake_db.committed # outer tx НЕ откатан
|
||
assert not fake_db.rolled_back
|
||
|
||
|
||
def test_upsert_tags_serialized_json(monkeypatch: Any) -> None:
|
||
"""tags сериализуются в JSON-строку (для CAST(:tags AS jsonb))."""
|
||
fake_db = _FakeDB()
|
||
monkeypatch.setattr(ul, "SessionLocal", lambda: fake_db)
|
||
|
||
ul._upsert_elements([_node(2, "power", "power", "substation", {"name": "ТП-5"})])
|
||
_, params = fake_db.executed[0]
|
||
assert params is not None
|
||
assert json.loads(params["tags"]) == {"name": "ТП-5"}
|
||
assert params["name"] == "ТП-5"
|
||
|
||
|
||
def test_upsert_empty_returns_zero(monkeypatch: Any) -> None:
|
||
fake_db = _FakeDB()
|
||
monkeypatch.setattr(ul, "SessionLocal", lambda: fake_db)
|
||
result = ul._upsert_elements([])
|
||
assert result["fetched"] == 0
|
||
assert result["inserted"] == 0
|
||
assert len(fake_db.executed) == 0
|
||
|
||
|
||
# ── retry-backoff helpers (#1746 follow-up) ──────────────────────────────────
|
||
|
||
|
||
def test_parse_retry_after_delta_seconds() -> None:
|
||
assert ul._parse_retry_after("30") == 30.0
|
||
assert ul._parse_retry_after(" 12.5 ") == 12.5
|
||
|
||
|
||
def test_parse_retry_after_invalid_returns_none() -> None:
|
||
assert ul._parse_retry_after(None) is None
|
||
assert ul._parse_retry_after("") is None
|
||
assert ul._parse_retry_after("Wed, 21 Oct 2026 07:28:00 GMT") is None # HTTP-date
|
||
assert ul._parse_retry_after("-5") is None # отрицательное
|
||
|
||
|
||
def test_backoff_delay_uses_retry_after_when_present(monkeypatch: Any) -> None:
|
||
"""Retry-After приоритетнее табличного backoff; джиттер ≥ 0."""
|
||
monkeypatch.setattr(ul.random, "uniform", lambda _a, _b: 0.0)
|
||
assert ul._backoff_delay(0, retry_after=42.0) == 42.0
|
||
|
||
|
||
def test_backoff_delay_exponential_table(monkeypatch: Any) -> None:
|
||
"""Без Retry-After — берём _RETRY_BACKOFF_SECONDS по индексу (с клампом)."""
|
||
monkeypatch.setattr(ul.random, "uniform", lambda _a, _b: 0.0)
|
||
assert ul._backoff_delay(0, None) == ul._RETRY_BACKOFF_SECONDS[0]
|
||
assert ul._backoff_delay(1, None) == ul._RETRY_BACKOFF_SECONDS[1]
|
||
# за пределами таблицы — кламп на последний элемент
|
||
assert ul._backoff_delay(99, None) == ul._RETRY_BACKOFF_SECONDS[-1]
|
||
|
||
|
||
# ── fetch_overpass_utility retry / mirror fallback (mocked httpx) ─────────────
|
||
|
||
|
||
class _FakeResponse:
|
||
"""Минимальный stand-in для httpx.Response: status_code + headers + json()."""
|
||
|
||
def __init__(
|
||
self,
|
||
status_code: int,
|
||
elements: list[dict] | None = None,
|
||
headers: dict[str, str] | None = None,
|
||
) -> None:
|
||
self.status_code = status_code
|
||
self.headers = headers or {}
|
||
self._elements = elements or []
|
||
|
||
def json(self) -> dict[str, Any]:
|
||
return {"elements": self._elements}
|
||
|
||
def raise_for_status(self) -> None:
|
||
if self.status_code >= 400:
|
||
raise httpx.HTTPStatusError(
|
||
f"HTTP {self.status_code}", request=None, response=None # type: ignore[arg-type]
|
||
)
|
||
|
||
|
||
class _ScriptedClient:
|
||
"""Фейк httpx.AsyncClient: отдаёт ответы по сценарию per-endpoint.
|
||
|
||
``script`` — dict {endpoint_url: list[_FakeResponse | Exception]}. Каждый POST
|
||
к endpoint забирает следующий заготовленный ответ; если ответы кончились,
|
||
последний повторяется (стабильный «всегда 429» хвост).
|
||
"""
|
||
|
||
def __init__(self, script: dict[str, list[Any]]) -> None:
|
||
self._script = {k: list(v) for k, v in script.items()}
|
||
self.calls: list[str] = []
|
||
|
||
async def __aenter__(self) -> _ScriptedClient:
|
||
return self
|
||
|
||
async def __aexit__(self, *_exc: Any) -> None:
|
||
return None
|
||
|
||
async def post(self, url: str, **_kw: Any) -> _FakeResponse:
|
||
self.calls.append(url)
|
||
responses = self._script.get(url, [])
|
||
if not responses:
|
||
return _FakeResponse(200, elements=[])
|
||
item = responses.pop(0) if len(responses) > 1 else responses[0]
|
||
if isinstance(item, Exception):
|
||
raise item
|
||
return item
|
||
|
||
|
||
def _single_query(monkeypatch: Any) -> None:
|
||
"""Сводим _UTILITY_QUERIES к одному типу — тесты fetch без лишних итераций."""
|
||
monkeypatch.setattr(ul, "_UTILITY_QUERIES", [("power", "line", "way", "power")])
|
||
|
||
|
||
def _no_sleep(monkeypatch: Any) -> None:
|
||
"""Глушим asyncio.sleep — backoff/inter-query паузы не тормозят тесты."""
|
||
|
||
async def _fast(_secs: float) -> None:
|
||
return None
|
||
|
||
monkeypatch.setattr(ul.asyncio, "sleep", _fast)
|
||
monkeypatch.setattr(ul.random, "uniform", lambda _a, _b: 0.0)
|
||
|
||
|
||
_EL = {"id": 1, "type": "way", "geometry": [{"lat": 56.8, "lon": 60.6}]}
|
||
|
||
|
||
async def test_fetch_retries_429_then_succeeds(monkeypatch: Any) -> None:
|
||
"""429 затем 200 на primary → query ретраится и возвращает элементы."""
|
||
_single_query(monkeypatch)
|
||
_no_sleep(monkeypatch)
|
||
primary = ul.OVERPASS_ENDPOINTS[0]
|
||
client = _ScriptedClient(
|
||
{primary: [_FakeResponse(429, headers={"Retry-After": "1"}), _FakeResponse(200, [_EL])]}
|
||
)
|
||
monkeypatch.setattr(ul.httpx, "AsyncClient", lambda **_kw: client)
|
||
|
||
elements = await ul.fetch_overpass_utility()
|
||
|
||
assert len(elements) == 1
|
||
assert elements[0]["_infrastructure_kind"] == "power"
|
||
# обе попытки били в primary, без перехода на зеркало
|
||
assert client.calls == [primary, primary]
|
||
|
||
|
||
async def test_fetch_falls_back_to_mirror(monkeypatch: Any) -> None:
|
||
"""Primary всегда 429, mirror отдаёт 200 → используется зеркало."""
|
||
_single_query(monkeypatch)
|
||
_no_sleep(monkeypatch)
|
||
primary = ul.OVERPASS_ENDPOINTS[0]
|
||
mirror = ul.OVERPASS_ENDPOINTS[1]
|
||
client = _ScriptedClient(
|
||
{
|
||
primary: [_FakeResponse(429)], # стабильный 429 на всех попытках
|
||
mirror: [_FakeResponse(200, [_EL])],
|
||
}
|
||
)
|
||
monkeypatch.setattr(ul.httpx, "AsyncClient", lambda **_kw: client)
|
||
|
||
elements = await ul.fetch_overpass_utility()
|
||
|
||
assert len(elements) == 1
|
||
# primary исчерпан (3 попытки), затем успех на mirror
|
||
assert client.calls.count(primary) == ul._MAX_RETRIES_PER_ENDPOINT
|
||
assert mirror in client.calls
|
||
|
||
|
||
async def test_fetch_504_then_mirror(monkeypatch: Any) -> None:
|
||
"""504 Gateway Timeout тоже ретраибл и триггерит фолбэк на зеркало."""
|
||
_single_query(monkeypatch)
|
||
_no_sleep(monkeypatch)
|
||
primary = ul.OVERPASS_ENDPOINTS[0]
|
||
mirror = ul.OVERPASS_ENDPOINTS[1]
|
||
client = _ScriptedClient(
|
||
{primary: [_FakeResponse(504)], mirror: [_FakeResponse(200, [_EL])]}
|
||
)
|
||
monkeypatch.setattr(ul.httpx, "AsyncClient", lambda **_kw: client)
|
||
|
||
elements = await ul.fetch_overpass_utility()
|
||
assert len(elements) == 1
|
||
assert mirror in client.calls
|
||
|
||
|
||
async def test_fetch_all_endpoints_fail_skips_query(monkeypatch: Any) -> None:
|
||
"""Все зеркала 429 → query пропускается (лог), sync НЕ падает, [] возвращён."""
|
||
_single_query(monkeypatch)
|
||
_no_sleep(monkeypatch)
|
||
script = {ep: [_FakeResponse(429)] for ep in ul.OVERPASS_ENDPOINTS}
|
||
client = _ScriptedClient(script)
|
||
monkeypatch.setattr(ul.httpx, "AsyncClient", lambda **_kw: client)
|
||
|
||
elements = await ul.fetch_overpass_utility()
|
||
|
||
assert elements == [] # query пропущен, но без исключения
|
||
# все endpoint'ы перебраны по _MAX_RETRIES_PER_ENDPOINT попыток
|
||
for ep in ul.OVERPASS_ENDPOINTS:
|
||
assert client.calls.count(ep) == ul._MAX_RETRIES_PER_ENDPOINT
|
||
|
||
|
||
async def test_fetch_one_failing_query_does_not_abort_others(monkeypatch: Any) -> None:
|
||
"""Один query-type валится на всех зеркалах, второй проходит → второй жив."""
|
||
monkeypatch.setattr(
|
||
ul,
|
||
"_UTILITY_QUERIES",
|
||
[("power", "line", "way", "power"), ("man_made", "water_works", "nwr", "water")],
|
||
)
|
||
_no_sleep(monkeypatch)
|
||
primary = ul.OVERPASS_ENDPOINTS[0]
|
||
|
||
# power=line → way (out geom), water_works → nwr (out geom): различаем по query.
|
||
def _route(url: str, **kw: Any) -> _FakeResponse:
|
||
data = kw.get("data", {})
|
||
query = data.get("data", "")
|
||
if '"power"="line"' in query:
|
||
return _FakeResponse(429) # всегда троттлится
|
||
return _FakeResponse(200, [_EL]) # water_works проходит
|
||
|
||
class _RoutingClient(_ScriptedClient):
|
||
async def post(self, url: str, **kw: Any) -> _FakeResponse:
|
||
self.calls.append(url)
|
||
return _route(url, **kw)
|
||
|
||
client = _RoutingClient({})
|
||
monkeypatch.setattr(ul.httpx, "AsyncClient", lambda **_kw: client)
|
||
|
||
elements = await ul.fetch_overpass_utility()
|
||
|
||
# power полностью провалился, water_works дошёл — sync не абортнул
|
||
assert len(elements) == 1
|
||
assert elements[0]["_infrastructure_kind"] == "water"
|
||
# power перебрал ВСЕ зеркала по _MAX_RETRIES_PER_ENDPOINT попыток,
|
||
# water_works успешно сходил на primary 1 раз → primary вызван N*retries... +1.
|
||
expected_primary = ul._MAX_RETRIES_PER_ENDPOINT + 1 # power×3 на primary + water_works×1
|
||
assert client.calls.count(primary) == expected_primary
|
||
# каждое зеркало получило ровно _MAX_RETRIES_PER_ENDPOINT попыток от power
|
||
for mirror in ul.OVERPASS_ENDPOINTS[1:]:
|
||
assert client.calls.count(mirror) == ul._MAX_RETRIES_PER_ENDPOINT
|