286 lines
12 KiB
Python
286 lines
12 KiB
Python
"""Unit-тесты OSM-геокода городских ТП/РП ЕКБ (#2119 B2, финал).
|
||
|
||
Pure / mock-based — без реальной сети и БД. Покрывает:
|
||
- extract_tp_number: «РП-101»→101, «ТП 1591»→1591, голый ref='101'→101,
|
||
«110 кВ» БЕЗ ТП-токена/ref→None, traction-скип на уровне fetch.
|
||
- fetch_osm_substations: скип substation=traction (mock httpx.post).
|
||
- _index_osm_by_number: дубль-номер (разные координаты) → ambiguous, не матчится;
|
||
дубль-теги одной точки (way+node, те же координаты) → НЕ ambiguous.
|
||
- geocode_tp_from_osm: dry-run НЕ пишет; apply=True шлёт UPDATE с CAST-формами
|
||
и geom IS NULL-гейтом; sanity-bbox отсекает координату вне ЕКБ.
|
||
- Celery-регистрация таска sync_osm_tp_geocode + beat-строка.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from contextlib import contextmanager
|
||
from typing import Any
|
||
|
||
from app.services.site_finder import osm_tp_geocoder as og
|
||
|
||
# ── extract_tp_number ────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_extract_tp_number_rp_token() -> None:
|
||
"""«РП-101» в name → 101 (ТП-токен + число)."""
|
||
assert og.extract_tp_number({"name": "РП-101"}) == "101"
|
||
|
||
|
||
def test_extract_tp_number_tp_hyphen() -> None:
|
||
"""«ТП-180» → 180."""
|
||
assert og.extract_tp_number({"name": "ТП-180"}) == "180"
|
||
|
||
|
||
def test_extract_tp_number_tp_space() -> None:
|
||
"""«ТП 1591» (пробел-разделитель) → 1591."""
|
||
assert og.extract_tp_number({"name": "ТП 1591"}) == "1591"
|
||
|
||
|
||
def test_extract_tp_number_bare_ref() -> None:
|
||
"""Голый ref='101' (без ТП-токена) → 101 — оператор проставил ref явно."""
|
||
assert og.extract_tp_number({"ref": "101"}) == "101"
|
||
|
||
|
||
def test_extract_tp_number_ref_priority_over_name() -> None:
|
||
"""ref приоритетнее name: ref='42', name='ТП-999' → 42."""
|
||
assert og.extract_tp_number({"ref": "42", "name": "ТП-999"}) == "42"
|
||
|
||
|
||
def test_extract_tp_number_voltage_without_token_is_none() -> None:
|
||
"""«110 кВ» БЕЗ ТП-токена и без ref → None (это класс напряжения, не номер)."""
|
||
assert og.extract_tp_number({"name": "Подстанция 110 кВ"}) is None
|
||
assert og.extract_tp_number({"name": "110 кВ", "voltage": "110000"}) is None
|
||
|
||
|
||
def test_extract_tp_number_empty_tags_is_none() -> None:
|
||
"""Нет ни ref, ни name → None."""
|
||
assert og.extract_tp_number({}) is None
|
||
|
||
|
||
def test_extract_tp_number_latin_tp_token() -> None:
|
||
"""Латинский токен «TP-42» → 42."""
|
||
assert og.extract_tp_number({"name": "TP-42"}) == "42"
|
||
|
||
|
||
# ── fetch_osm_substations: traction-скип ─────────────────────────────────────
|
||
|
||
|
||
class _FakeResp:
|
||
def __init__(self, payload: dict) -> None:
|
||
self._payload = payload
|
||
|
||
def raise_for_status(self) -> None:
|
||
pass
|
||
|
||
def json(self) -> dict:
|
||
return self._payload
|
||
|
||
|
||
def test_fetch_skips_traction(monkeypatch: Any) -> None:
|
||
"""substation=traction (тяговые ж/д) отсеиваются; обычные ТП остаются."""
|
||
payload = {
|
||
"elements": [
|
||
{"type": "node", "id": 1, "lat": 56.8, "lon": 60.6, "tags": {"name": "ТП-1"}},
|
||
{
|
||
"type": "node",
|
||
"id": 2,
|
||
"lat": 56.8,
|
||
"lon": 60.6,
|
||
"tags": {"substation": "traction", "name": "Тяговая-2"},
|
||
},
|
||
]
|
||
}
|
||
|
||
def _fake_post(url: str, data: Any = None, timeout: Any = None) -> _FakeResp:
|
||
return _FakeResp(payload)
|
||
|
||
monkeypatch.setattr(og.httpx, "post", _fake_post)
|
||
kept = og.fetch_osm_substations()
|
||
assert len(kept) == 1
|
||
assert kept[0]["id"] == 1
|
||
|
||
|
||
# ── _index_osm_by_number: ambiguous ──────────────────────────────────────────
|
||
|
||
|
||
def _node(el_id: int, lat: float, lon: float, tags: dict) -> dict:
|
||
return {"type": "node", "id": el_id, "lat": lat, "lon": lon, "tags": tags}
|
||
|
||
|
||
def test_index_ambiguous_duplicate_number() -> None:
|
||
"""Один номер несут 2 фичи с РАЗНЫМИ координатами → ambiguous, из матча выпадает."""
|
||
elements = [
|
||
_node(1, 56.80, 60.60, {"name": "ТП-100"}),
|
||
_node(2, 56.85, 60.65, {"name": "ТП-100"}), # тот же номер, другая точка
|
||
_node(3, 56.82, 60.62, {"name": "ТП-200"}), # уникальный
|
||
]
|
||
by_number, ambiguous = og._index_osm_by_number(elements)
|
||
assert "100" in ambiguous
|
||
assert "100" not in by_number
|
||
assert "200" in by_number
|
||
|
||
|
||
def test_index_same_point_not_ambiguous() -> None:
|
||
"""Дубль-теги ОДНОЙ точки (way+node на тех же координатах) → НЕ ambiguous."""
|
||
elements = [
|
||
_node(1, 56.80, 60.60, {"name": "ТП-100"}),
|
||
{"type": "way", "id": 2, "center": {"lat": 56.80, "lon": 60.60}, "tags": {"ref": "100"}},
|
||
]
|
||
by_number, ambiguous = og._index_osm_by_number(elements)
|
||
assert "100" not in ambiguous
|
||
assert by_number["100"] == (56.80, 60.60)
|
||
|
||
|
||
def test_index_out_of_bbox_skipped() -> None:
|
||
"""Координата вне bbox ЕКБ (sanity-гейт) → не индексируется."""
|
||
elements = [_node(1, 55.75, 37.61, {"name": "ТП-300"})] # Москва
|
||
by_number, ambiguous = og._index_osm_by_number(elements)
|
||
assert by_number == {}
|
||
assert ambiguous == set()
|
||
|
||
|
||
def test_index_way_center_coords() -> None:
|
||
"""way без lat/lon, но с center → берём center."""
|
||
elements = [
|
||
{"type": "way", "id": 9, "center": {"lat": 56.83, "lon": 60.63}, "tags": {"ref": "555"}}
|
||
]
|
||
by_number, _ = og._index_osm_by_number(elements)
|
||
assert by_number["555"] == (56.83, 60.63)
|
||
|
||
|
||
# ── Fake DB session (execute + begin_nested), как в test_eesk_reserve_loader ──
|
||
|
||
|
||
class _FakeResult:
|
||
def __init__(self, rows: list[Any] | None = None, rowcount: int = 1) -> None:
|
||
self._rows = rows or []
|
||
self.rowcount = rowcount
|
||
|
||
def all(self) -> list[Any]:
|
||
return self._rows
|
||
|
||
|
||
class _FakeSession:
|
||
"""Мини-заглушка Session: собирает execute-вызовы, поддерживает begin_nested."""
|
||
|
||
def __init__(self, candidate_norms: list[str], rowcount: int = 1) -> None:
|
||
self.calls: list[tuple[str, Any]] = []
|
||
self._candidates = [(n,) for n in candidate_norms]
|
||
self._rowcount = rowcount
|
||
self.committed = False
|
||
|
||
@contextmanager
|
||
def begin_nested(self): # type: ignore[no-untyped-def]
|
||
yield
|
||
|
||
def execute(self, stmt: Any, params: Any = None) -> _FakeResult:
|
||
sql = str(stmt)
|
||
self.calls.append((sql, params))
|
||
if "SELECT" in sql and "power_tp_rp_reserves" in sql:
|
||
return _FakeResult(rows=self._candidates)
|
||
return _FakeResult(rowcount=self._rowcount)
|
||
|
||
def commit(self) -> None:
|
||
self.committed = True
|
||
|
||
def rollback(self) -> None:
|
||
pass
|
||
|
||
|
||
def _patch_fetch(monkeypatch: Any, elements: list[dict]) -> None:
|
||
monkeypatch.setattr(og, "fetch_osm_substations", lambda: elements)
|
||
|
||
|
||
# ── geocode_tp_from_osm ──────────────────────────────────────────────────────
|
||
|
||
|
||
def test_geocode_dry_run_does_not_write(monkeypatch: Any) -> None:
|
||
"""Dry-run (apply=False): matched считается, но UPDATE не шлётся и commit не зовётся."""
|
||
_patch_fetch(monkeypatch, [_node(1, 56.80, 60.60, {"name": "ТП-101"})])
|
||
db = _FakeSession(candidate_norms=["101"])
|
||
result = og.geocode_tp_from_osm(db=db, apply=False)
|
||
|
||
assert result["matched"] == 1
|
||
assert result["updated"] == 0
|
||
assert result["applied"] is False
|
||
assert not db.committed
|
||
# Только SELECT-кандидатов, никаких UPDATE.
|
||
assert all("UPDATE" not in sql for sql, _ in db.calls)
|
||
|
||
|
||
def test_geocode_apply_updates_with_cast(monkeypatch: Any) -> None:
|
||
"""apply=True: UPDATE шлётся с CAST-формами, geom IS NULL-гейтом, commit зовётся."""
|
||
_patch_fetch(monkeypatch, [_node(1, 56.80, 60.60, {"name": "ТП-101"})])
|
||
db = _FakeSession(candidate_norms=["101"], rowcount=1)
|
||
result = og.geocode_tp_from_osm(db=db, apply=True)
|
||
|
||
assert result["matched"] == 1
|
||
assert result["updated"] == 1
|
||
assert result["applied"] is True
|
||
assert db.committed
|
||
|
||
update_calls = [(sql, p) for sql, p in db.calls if "UPDATE" in sql]
|
||
assert len(update_calls) == 1
|
||
sql, params = update_calls[0]
|
||
# CAST-формы (psycopg3-канон), НИКОГДА :x::type.
|
||
assert "CAST(:lon AS float)" in sql
|
||
assert "CAST(:lat AS float)" in sql
|
||
assert "CAST(:name_norm AS text)" in sql
|
||
assert "::float" not in sql and "::text" not in sql
|
||
# Идемпотентность + скоуп ЕКБ.
|
||
assert "geom IS NULL" in sql
|
||
assert "geocode_status = 'osm'" in sql
|
||
assert "municipality = 'Екатеринбург'" in sql
|
||
assert params == {"name_norm": "101", "lat": 56.80, "lon": 60.60}
|
||
|
||
|
||
def test_geocode_ambiguous_not_matched(monkeypatch: Any) -> None:
|
||
"""Дубль-номер (ambiguous) не матчится: candidate '100' есть, но geom не пишется."""
|
||
_patch_fetch(
|
||
monkeypatch,
|
||
[
|
||
_node(1, 56.80, 60.60, {"name": "ТП-100"}),
|
||
_node(2, 56.85, 60.65, {"name": "ТП-100"}), # коллизия
|
||
],
|
||
)
|
||
db = _FakeSession(candidate_norms=["100"], rowcount=1)
|
||
result = og.geocode_tp_from_osm(db=db, apply=True)
|
||
|
||
assert result["ambiguous"] == 1
|
||
assert result["matched"] == 0
|
||
assert result["updated"] == 0
|
||
assert all("UPDATE" not in sql for sql, _ in db.calls)
|
||
|
||
|
||
def test_geocode_unmatched_candidate_skipped(monkeypatch: Any) -> None:
|
||
"""Кандидат без OSM-номера просто скипается (matched=0), без падения."""
|
||
_patch_fetch(monkeypatch, [_node(1, 56.80, 60.60, {"name": "ТП-777"})])
|
||
db = _FakeSession(candidate_norms=["101"], rowcount=1) # 101 нет в OSM
|
||
result = og.geocode_tp_from_osm(db=db, apply=True)
|
||
assert result["matched"] == 0
|
||
assert result["updated"] == 0
|
||
|
||
|
||
# ── Celery / beat регистрация ────────────────────────────────────────────────
|
||
|
||
|
||
def test_celery_task_registered() -> None:
|
||
"""Таск sync_osm_tp_geocode регистрируется при импорте модуля тасков.
|
||
|
||
Модуль в celery_app.include — на worker-старте импортируется и таск попадает в
|
||
celery_app.tasks. В тесте импорт делаем явно (Celery не бутит воркер).
|
||
"""
|
||
import app.workers.tasks.connection_capacity_sync # noqa: F401 — импорт регистрирует таск
|
||
from app.workers.celery_app import celery_app
|
||
|
||
assert "tasks.connection_capacity_sync.sync_osm_tp_geocode" in celery_app.tasks
|
||
|
||
|
||
def test_beat_schedule_entry() -> None:
|
||
"""Beat содержит еженедельную вторник-05:45 запись osm-tp-geocode."""
|
||
from app.workers.beat_schedule import build_beat_schedule
|
||
|
||
schedule = build_beat_schedule()
|
||
assert "osm-tp-geocode-weekly" in schedule
|
||
entry = schedule["osm-tp-geocode-weekly"]
|
||
assert entry["task"] == "tasks.connection_capacity_sync.sync_osm_tp_geocode"
|