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
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 (оба файла).
604 lines
24 KiB
Python
604 lines
24 KiB
Python
"""Тесты для 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}"
|
||
)
|