gendesign/backend/tests/workers/test_nspd_geo.py
bot-backend be70cacf18
All checks were successful
CI / changes (push) Successful in 7s
CI / changes (pull_request) Successful in 9s
CI / frontend-tests (push) Successful in 1m4s
CI / frontend-tests (pull_request) Successful in 57s
CI / openapi-codegen-check (push) Successful in 2m7s
CI / openapi-codegen-check (pull_request) Successful in 1m41s
CI / backend-tests (push) Successful in 8m59s
CI / backend-tests (pull_request) Successful in 8m58s
test: выровнять weather_cache/nspd_geo тесты под #1377/#1336 (CI red→green)
backend-tests CI упал на 4 стейл-тестах — код автофиксов обновили, тесты нет:
- test_weather_cache: фикстура _make_air_response отдавала hourly-форму, а код
  (#1377) теперь читает current-bucket → current.time пуст → None. Фикстура → current.
- test_nspd_geo: 2 теста стерегли ошибочный ST_Transform(3857→4326), который #1336
  убрал (геометрия уже WGS84, иначе Null Island). Ассерты через _strip_sql_comments
  (как сиблинг parcel-тест) → ST_SetSRID(...,4326) без ST_Transform. Переименованы.

Локально: 38 passed (оба файла).
2026-06-16 08:35:14 +03:00

604 lines
24 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Тесты для backend/app/workers/tasks/nspd_geo.py — on-demand НСПД-ингест.
Покрывает 3 prod-confirmed бага «пользователь вводит новый/произвольный реальный участок»:
- Finding 1: _save_parcel / _save_quarter должны ST_Multi() geom (колонки MultiPolygon,
migration 93 / 58) — иначе "Geometry type (Polygon) does not match column type (MultiPolygon)".
- Finding 2: _save_building пишет в широкую схему cad_buildings (migration 92):
quarter_cad_number (не quarter_cad_num), source NOT NULL, floors через _safe_int.
Mock DB через unittest.mock — без реального PostgreSQL / NSPD сети.
"""
from __future__ import annotations
from typing import Any
from unittest.mock import MagicMock
import pytest
from celery.exceptions import SoftTimeLimitExceeded
from app.workers.tasks import nspd_geo as nspd_geo_mod
from app.workers.tasks.nspd_geo import (
_ZOMBIE_PAUSED_THRESHOLD,
_ZOMBIE_RUNNING_THRESHOLD,
_save_building,
_save_parcel,
_save_quarter,
cleanup_zombies,
process_nspd_geo_job,
)
# ── Фикстуры payload (форма ответа _persist_target: {"data": {"features": [...]}}) ──
def _parcel_payload(cad_num: str = "66:41:0610029:83") -> dict[str, Any]:
"""NSPD-ответ для ЗУ (одноконтурный Polygon — триггер бага type-mismatch)."""
return {
"data": {
"features": [
{
"type": "Feature",
"geometry": {
"type": "Polygon",
"coordinates": [
[
[60.6, 56.8],
[60.61, 56.8],
[60.61, 56.81],
[60.6, 56.81],
[60.6, 56.8],
]
],
},
"properties": {"options": {"cad_num": cad_num}},
}
]
}
}
def _quarter_payload(cad_num: str = "66:41:0610029") -> dict[str, Any]:
"""NSPD-ответ для квартала (одноконтурный Polygon в EPSG:3857)."""
return {
"data": {
"features": [
{
"type": "Feature",
"geometry": {
"type": "Polygon",
"coordinates": [
[
[6090000.0, 7590000.0],
[6090200.0, 7590000.0],
[6090200.0, 7590200.0],
[6090000.0, 7590200.0],
[6090000.0, 7590000.0],
]
],
},
"properties": {"options": {}},
}
]
}
}
def _building_payload(
cad_num: str = "66:41:0610029:83",
floors: Any = "5-7",
quarter: str | None = "66:41:0610029",
) -> dict[str, Any]:
"""NSPD-ответ для здания. floors по умолчанию range-строка "5-7" (триггер бага floors INT)."""
opts: dict[str, Any] = {
"cad_num": cad_num,
"purpose": "Жилое",
"building_name": "Тестовое здание",
"readable_address": "г Екатеринбург, ул Тестовая, д 1",
"area": "1234.56",
"floors": floors,
}
if quarter is not None:
opts["quarter_cad_number"] = quarter
return {
"data": {
"features": [
{
"type": "Feature",
"geometry": {
"type": "Polygon",
"coordinates": [
[
[6090000.0, 7590000.0],
[6090050.0, 7590000.0],
[6090050.0, 7590050.0],
[6090000.0, 7590050.0],
[6090000.0, 7590000.0],
]
],
},
"properties": {"options": opts},
}
]
}
}
def _capturing_db() -> tuple[MagicMock, list[tuple[str, Any]]]:
"""Mock Session, фиксирующий (sql_text, params) каждого db.execute."""
captured: list[tuple[str, Any]] = []
db = MagicMock()
db.execute = lambda stmt, params=None: captured.append((str(stmt), params)) or MagicMock()
return db, captured
def _strip_sql_comments(sql: str) -> str:
"""Убрать `-- ...` комментарии — чтобы assert'ить ИСПОЛНЯЕМЫЙ SQL, не текст комментов.
Комментарии легитимно упоминают ST_Transform/ST_Multi, поэтому проверяем выражения
только в реальном SQL.
"""
lines = []
for line in sql.splitlines():
idx = line.find("--")
lines.append(line if idx == -1 else line[:idx])
return "\n".join(lines)
# ── Finding 1: ST_Multi для MultiPolygon-колонок ─────────────────────────────
def test_save_parcel_sql_wraps_geom_in_st_multi() -> None:
"""cad_parcels.geom = MultiPolygon (migration 93) → SQL должен ST_Multi() обернуть geom.
Иначе одноконтурный Polygon из rosreestr2coord → "Geometry type (Polygon) does not
match column type (MultiPolygon)" → on-demand ингест участка крашится.
"""
db, captured = _capturing_db()
n = _save_parcel(db, _parcel_payload(), "66:41:0610029:83")
assert n == 1
assert len(captured) == 1
sql, _params = captured[0]
assert "ST_Multi(" in sql, "SQL _save_parcel должен содержать ST_Multi для MultiPolygon schema"
def test_save_parcel_sql_no_transform_parcel_is_4326() -> None:
"""parcel GeoJSON уже WGS84 (rosreestr2coord) — только ST_SetSRID(...,4326), без ST_Transform.
ST_Transform 3857→4326 на 4326-геометрии схлопнул бы участок в точку у (0,0).
"""
db, captured = _capturing_db()
_save_parcel(db, _parcel_payload(), "66:41:0610029:83")
sql, _params = captured[0]
executable = _strip_sql_comments(sql)
assert "ST_Transform" not in executable, "parcel уже 4326 — ST_Transform схлопнет в точку"
assert "ST_SetSRID(ST_GeomFromGeoJSON(:g), 4326)" in executable
def test_save_quarter_sql_wraps_geom_in_st_multi() -> None:
"""cad_quarters_geom.geom = MultiPolygon NOT NULL (migration 58) → ST_Multi() обязателен."""
db, captured = _capturing_db()
n = _save_quarter(db, _quarter_payload(), "66:41:0610029")
assert n == 1
assert len(captured) == 1
sql, _params = captured[0]
assert "ST_Multi(" in sql, "SQL _save_quarter должен содержать ST_Multi для MultiPolygon schema"
def test_save_quarter_sql_no_transform_wgs84() -> None:
"""quarter GeoJSON уже в WGS84 (rosreestr2coord) → ST_SetSRID(...,4326) БЕЗ ST_Transform
(#1336: ST_Transform из 3857 трактовал градусы как метры → коллапс к Null Island).
ST_Multi сохраняется — колонка cad_quarters_geom.geom = MultiPolygon."""
db, captured = _capturing_db()
_save_quarter(db, _quarter_payload(), "66:41:0610029")
sql, _params = captured[0]
executable = _strip_sql_comments(sql)
assert "ST_Multi(ST_SetSRID(ST_GeomFromGeoJSON(:g), 4326))" in executable
assert "ST_Transform" not in executable
def test_save_parcel_empty_features_returns_zero() -> None:
"""Пустой features → 0, без db.execute."""
db, captured = _capturing_db()
assert _save_parcel(db, {"data": {"features": []}}, "66:41:0610029:83") == 0
assert captured == []
# ── Finding 2: _save_building → широкая схема cad_buildings (migration 92) ─────
def test_save_building_uses_quarter_cad_number_column() -> None:
"""migration 92: колонка quarter_cad_number (НЕ quarter_cad_num) → иначе UndefinedColumn."""
db, captured = _capturing_db()
n = _save_building(db, _building_payload(), "66:41:0610029:83")
assert n == 1
sql, params = captured[0]
assert "quarter_cad_number" in sql
# старое имя колонки не должно встречаться (как отдельное слово quarter_cad_num)
assert "quarter_cad_num " not in sql
assert "quarter_cad_num," not in sql
assert "quarter_cad_num=" not in sql
assert params["qcad"] == "66:41:0610029"
def test_save_building_sets_source_not_null() -> None:
"""cad_buildings.source NOT NULL без дефолта (migration 92) → INSERT должен биндить source."""
db, captured = _capturing_db()
_save_building(db, _building_payload(), "66:41:0610029:83")
sql, params = captured[0]
assert ":source" in sql
assert params["source"] == "nspd"
def test_save_building_safe_int_floors_range_string() -> None:
"""floors INT: NSPD range "5-7" → _safe_int → None (иначе InvalidTextRepresentation).
Сырой "5-7" в floors INT крашил бы INSERT.
"""
db, captured = _capturing_db()
_save_building(db, _building_payload(floors="5-7"), "66:41:0610029:83")
_sql, params = captured[0]
assert params["floors"] is None # "5-7" не парсится → None, не сырая строка
def test_save_building_safe_int_floors_numeric() -> None:
"""floors "9" → int 9 (валидное число парсится)."""
db, captured = _capturing_db()
_save_building(db, _building_payload(floors="9"), "66:41:0610029:83")
_sql, params = captured[0]
assert params["floors"] == 9
def test_save_building_derives_quarter_when_missing() -> None:
"""Если NSPD не дал quarter_cad_number → выводим из cad_num (первые 3 сегмента)."""
db, captured = _capturing_db()
_save_building(db, _building_payload(quarter=None), "66:41:0610029:83")
_sql, params = captured[0]
assert params["qcad"] == "66:41:0610029"
def test_save_building_geom_no_transform_no_st_multi() -> None:
"""cad_buildings.geom = GEOMETRY(Geometry, 4326) полиморфный → ST_Multi НЕ нужен;
геометрия уже WGS84 (rosreestr2coord) → ST_SetSRID(...,4326) БЕЗ ST_Transform
(#1336: ST_Transform из 3857 → Null Island)."""
db, captured = _capturing_db()
_save_building(db, _building_payload(), "66:41:0610029:83")
sql, _params = captured[0]
executable = _strip_sql_comments(sql)
assert "ST_SetSRID(ST_GeomFromGeoJSON(:g), 4326)" in executable
assert "ST_Transform" not in executable
assert "ST_Multi(" not in executable # полиморфная колонка — coerce не нужен
def test_save_building_skips_feature_with_other_cad_num() -> None:
"""NSPD может вернуть здания всего квартала — берём только запрошенный cad_num."""
db, captured = _capturing_db()
payload = _building_payload(cad_num="66:41:0610029:99") # чужой cad_num
n = _save_building(db, payload, "66:41:0610029:83")
assert n == 0
assert captured == []
def test_save_building_area_safe_numeric() -> None:
"""area NUMERIC: строковое "1234.56" → float через _safe_numeric."""
db, captured = _capturing_db()
_save_building(db, _building_payload(), "66:41:0610029:83")
_sql, params = captured[0]
assert params["area"] == 1234.56
# ── psycopg v3 CAST-trap guard (backend.md) ─────────────────────────────────
def test_save_sql_uses_cast_not_double_colon() -> None:
"""SQL _save_* использует CAST(:x AS jsonb), не :x::jsonb (SQLAlchemy psycopg v3 trap)."""
db, captured = _capturing_db()
_save_parcel(db, _parcel_payload(), "66:41:0610029:83")
_save_quarter(db, _quarter_payload(), "66:41:0610029")
_save_building(db, _building_payload(), "66:41:0610029:83")
for sql, _params in captured:
assert "::jsonb" not in sql, f"SQL содержит ::jsonb — нужен CAST(:x AS jsonb): {sql[:80]}"
assert "CAST(:props AS jsonb)" in sql
# ── cleanup_zombies thresholds (issue #1215: WAF backoff vs zombie race) ────
class _FakeRowsResult:
"""sqlalchemy-like Result, который запоминает (sql, params) и возвращает 0 rows.
cleanup_zombies использует cain `.mappings().all()` — мокаем под него.
"""
def __init__(self, captured: list[tuple[str, dict[str, Any]]], stmt: Any, params: Any) -> None:
captured.append((str(stmt), params or {}))
self._stmt = stmt
self._params = params
def mappings(self) -> _FakeRowsResult:
return self
def all(self) -> list[dict[str, Any]]:
return []
def _capturing_session_factory() -> tuple[MagicMock, list[tuple[str, dict[str, Any]]]]:
"""Fake SessionLocal() — возвращает session, чьи execute-calls фиксируются."""
captured: list[tuple[str, dict[str, Any]]] = []
session = MagicMock()
session.execute = lambda stmt, params=None: _FakeRowsResult(captured, stmt, params)
session.commit = MagicMock()
session.rollback = MagicMock()
session.close = MagicMock()
factory = MagicMock(return_value=session)
return factory, captured
def test_cleanup_zombies_sql_uses_cast_as_interval_not_double_colon(
monkeypatch: Any,
) -> None:
"""psycopg v3: только CAST(:x AS interval), запрещён `:x::interval`."""
factory, captured = _capturing_session_factory()
monkeypatch.setattr(nspd_geo_mod, "SessionLocal", factory)
cleanup_zombies()
assert len(captured) == 1, "cleanup_zombies должен выполнить ровно один UPDATE"
sql, _params = captured[0]
assert "CAST(:running_threshold AS interval)" in sql
assert "CAST(:paused_threshold AS interval)" in sql
# SQLAlchemy + psycopg v3 ломаются на `:name::type` (backend.md)
assert ":running_threshold::" not in sql
assert ":paused_threshold::" not in sql
def test_cleanup_zombies_binds_running_and_paused_thresholds(monkeypatch: Any) -> None:
"""Issue #1215: running порог > max WAF backoff (240s = 4 min), paused — большой cooldown.
Контракт thresholds задокументирован в module-level константах.
"""
factory, captured = _capturing_session_factory()
monkeypatch.setattr(nspd_geo_mod, "SessionLocal", factory)
cleanup_zombies()
_sql, params = captured[0]
assert params["running_threshold"] == _ZOMBIE_RUNNING_THRESHOLD
assert params["paused_threshold"] == _ZOMBIE_PAUSED_THRESHOLD
# Контракт: running threshold выше старых 2 мин (которые ре-enqueue'или живые WAF-jobs)
# и выше max WAF backoff (WAF_BACKOFF_BASE_S * 8 = 240s = 4 мин).
assert _ZOMBIE_RUNNING_THRESHOLD == "6 minutes"
# Контракт: paused threshold большой (WAF-пауза должна реально удержаться).
assert _ZOMBIE_PAUSED_THRESHOLD == "30 minutes"
def test_cleanup_zombies_splits_running_and_paused_in_where_clause(
monkeypatch: Any,
) -> None:
"""SQL разделяет 'running' и 'paused' с разными порогами — не общий IN-фильтр."""
factory, captured = _capturing_session_factory()
monkeypatch.setattr(nspd_geo_mod, "SessionLocal", factory)
cleanup_zombies()
sql, _params = captured[0]
executable = _strip_sql_comments(sql)
# Старая ветка с общим INTERVAL '2 minutes' для обоих статусов удалена
assert "INTERVAL '2 minutes'" not in executable
assert "status IN ('running', 'paused')" not in executable
# Новая ветка: status='running' и status='paused' в отдельных предикатах
assert "status = 'running'" in executable
assert "status = 'paused'" in executable
# Каждый из двух предикатов сравнивает heartbeat_at с NOW() - threshold
assert executable.count("heartbeat_at") >= 2
def test_cleanup_zombies_paused_threshold_strictly_greater_than_running(
monkeypatch: Any,
) -> None:
"""paused cooldown должен быть СТРОГО больше running, иначе пауза аннулируется минутным beat."""
factory, captured = _capturing_session_factory()
monkeypatch.setattr(nspd_geo_mod, "SessionLocal", factory)
cleanup_zombies()
# 6 vs 30 — числовое сравнение префиксов как guard от случайного "6 minutes" == "30 minutes"
running_min = int(_ZOMBIE_RUNNING_THRESHOLD.split()[0])
paused_min = int(_ZOMBIE_PAUSED_THRESHOLD.split()[0])
assert paused_min > running_min, (
"paused threshold должен быть строго больше running, "
"иначе WAF-пауза не удержится дольше running-cooldown"
)
# Sanity: SQL действительно отправлен с двумя разными порогами
_sql, params = captured[0]
assert params["running_threshold"] != params["paused_threshold"]
# ── SoftTimeLimitExceeded handler (issue #1235) ─────────────────────────────
class _JobFetchResult:
"""sqlalchemy-like Result, возвращающий заданный mapping для SELECT FROM nspd_geo_jobs."""
def __init__(self, row: dict[str, Any] | None) -> None:
self._row = row
def mappings(self) -> _JobFetchResult:
return self
def first(self) -> dict[str, Any] | None:
return self._row
def scalar_one(self) -> Any:
return 1
def _build_soft_limit_session(
raise_in: str = "fetch",
) -> tuple[MagicMock, list[tuple[str, dict[str, Any]]]]:
"""Fake SessionLocal() для теста graceful shutdown по SoftTimeLimitExceeded.
Контракт: первый SELECT ... FROM nspd_geo_jobs WHERE job_id=... → job row,
UPDATE _start_job → OK, INSERT nspd_geo_log → OK, второй SELECT (target loop) →
pending target row. Затем fetcher кидает SoftTimeLimitExceeded.
После raise — handler должен (a) выполнить UPDATE heartbeat + counters,
(b) UPDATE nspd_geo_jobs SET status='paused', (c) re-raise.
"""
captured: list[tuple[str, dict[str, Any]]] = []
job_row = {
"job_id": 42,
"rate_ms": 0, # отключаем sleep между итерациями для теста
"targets_done": 0,
"targets_failed": 0,
"targets_skipped": 0,
"requests_count": 0,
"waf_blocked_count": 0,
}
target_row = {
"target_id": 7,
"cad_num": "66:41:0610029:83",
"thematic_id": 1,
"attempts": 0,
}
call_state = {"job_selects": 0, "target_selects": 0}
def _execute(stmt: Any, params: Any = None) -> Any:
sql = str(stmt)
captured.append((sql, params or {}))
if "FROM nspd_geo_jobs" in sql and "SELECT" in sql:
call_state["job_selects"] += 1
return _JobFetchResult(job_row)
if "FROM nspd_geo_targets" in sql and "SELECT" in sql:
call_state["target_selects"] += 1
# Первый SELECT — отдаём pending target. Дальше — None (loop break),
# но до второй итерации не дойдём — fetcher кинет SoftTimeLimitExceeded.
if call_state["target_selects"] == 1:
return _JobFetchResult(target_row)
return _JobFetchResult(None)
if "INSERT INTO nspd_geo_jobs" in sql:
return _JobFetchResult({"job_id": 42}) # для scalar_one
return MagicMock()
session = MagicMock()
session.execute = _execute
session.commit = MagicMock()
session.rollback = MagicMock()
session.close = MagicMock()
return MagicMock(return_value=session), captured
def test_soft_time_limit_exceeded_reraises_not_swallowed(monkeypatch: Any) -> None:
"""Issue #1235 root: `except (NspdLiteError, Exception)` проглатывало SoftTimeLimitExceeded.
После fix — explicit `except SoftTimeLimitExceeded: raise` ДО generic ветки →
исключение пробрасывается наружу process_nspd_geo_job, Celery освобождает воркер.
"""
factory, _captured = _build_soft_limit_session()
monkeypatch.setattr(nspd_geo_mod, "SessionLocal", factory)
def _raise_soft_limit(*_args: Any, **_kwargs: Any) -> Any:
raise SoftTimeLimitExceeded()
monkeypatch.setattr(nspd_geo_mod, "fetch_via_rosreestr2coord", _raise_soft_limit)
with pytest.raises(SoftTimeLimitExceeded):
process_nspd_geo_job(42)
def test_soft_time_limit_exceeded_sets_paused_status(monkeypatch: Any) -> None:
"""Handler коммитит status='paused' перед re-raise — иначе job навечно 'running'
с stale heartbeat → cleanup_zombies ре-enqueue'ит через running-threshold (6 мин),
но мы хотим явный paused + 30 мин cooldown.
"""
factory, captured = _build_soft_limit_session()
monkeypatch.setattr(nspd_geo_mod, "SessionLocal", factory)
monkeypatch.setattr(
nspd_geo_mod,
"fetch_via_rosreestr2coord",
MagicMock(side_effect=SoftTimeLimitExceeded()),
)
with pytest.raises(SoftTimeLimitExceeded):
process_nspd_geo_job(42)
# Должен быть UPDATE с status = :st где :st == 'paused'
paused_updates = [
(sql, params)
for sql, params in captured
if "UPDATE nspd_geo_jobs" in sql and "status = :st" in sql and params.get("st") == "paused"
]
assert paused_updates, (
"SoftTimeLimitExceeded handler должен выставить status='paused' до re-raise; "
f"captured updates: {[(s[:80], p) for s, p in captured if 'UPDATE nspd_geo_jobs' in s]}"
)
def test_soft_time_limit_exceeded_flushes_heartbeat(monkeypatch: Any) -> None:
"""Handler коммитит финальный heartbeat (counters + timestamp) до raise.
Иначе после паузы job выглядит в UI с устаревшими счётчиками — admin не видит
сколько таргетов реально обработано.
"""
factory, captured = _build_soft_limit_session()
monkeypatch.setattr(nspd_geo_mod, "SessionLocal", factory)
monkeypatch.setattr(
nspd_geo_mod,
"fetch_via_rosreestr2coord",
MagicMock(side_effect=SoftTimeLimitExceeded()),
)
with pytest.raises(SoftTimeLimitExceeded):
process_nspd_geo_job(42)
heartbeat_updates = [
(sql, params)
for sql, params in captured
if "heartbeat_at = NOW()" in sql and "targets_done" in sql
]
assert (
heartbeat_updates
), "SoftTimeLimitExceeded handler должен flush'нуть heartbeat с counters перед raise"
def test_soft_time_limit_exceeded_does_not_overwrite_paused_with_failed(
monkeypatch: Any,
) -> None:
"""Outer `except Exception` НЕ должен затереть 'paused' на 'failed' для SoftTimeLimitExceeded.
Issue #1235 corollary: если inner handler выставил 'paused' и raise — outer
fallback не должен мгновенно переписать status='failed' через _finish_job.
"""
factory, captured = _build_soft_limit_session()
monkeypatch.setattr(nspd_geo_mod, "SessionLocal", factory)
monkeypatch.setattr(
nspd_geo_mod,
"fetch_via_rosreestr2coord",
MagicMock(side_effect=SoftTimeLimitExceeded()),
)
with pytest.raises(SoftTimeLimitExceeded):
process_nspd_geo_job(42)
failed_updates = [
params
for sql, params in captured
if "UPDATE nspd_geo_jobs" in sql and "status = :st" in sql and params.get("st") == "failed"
]
assert not failed_updates, (
"outer except не должен затирать 'paused' на 'failed' для SoftTimeLimitExceeded; "
f"got failed updates: {failed_updates}"
)