271 lines
11 KiB
Python
271 lines
11 KiB
Python
"""Тесты для backend/app/workers/tasks/scrape_cadastre.py.
|
||
|
||
Покрывает:
|
||
- Fix H — per-quarter freshness skip в enqueue_cadastre_harvest.
|
||
- Bug #1236 — генерик-фейл инкрементит targets_failed только на финальной
|
||
попытке + регистрирует cleanup_cadastre_zombies в beat_schedule.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import datetime as _dt
|
||
from types import SimpleNamespace
|
||
from typing import Any
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import pytest
|
||
|
||
|
||
def _make_mock_job(quarters: list[str], raw_targets_extra: dict[str, Any] | None = None) -> dict:
|
||
"""Mock cadastre_jobs row."""
|
||
raw_targets = {"quarters": quarters}
|
||
if raw_targets_extra:
|
||
raw_targets.update(raw_targets_extra)
|
||
return {
|
||
"job_id": 99,
|
||
"status": "pending",
|
||
"raw_targets": raw_targets,
|
||
}
|
||
|
||
|
||
def _make_mock_db(job: dict, fresh_quarters: list[str] | None = None) -> MagicMock:
|
||
"""Mock DB session.
|
||
|
||
1. db.execute("SELECT * FROM cadastre_jobs ...").mappings().first() → job
|
||
2. db.execute("SELECT cad_number FROM cad_quarter_stats ...").scalars().all() → fresh
|
||
3. db.execute(UPDATE) → no-op
|
||
"""
|
||
db = MagicMock()
|
||
fresh_quarters = fresh_quarters or []
|
||
|
||
def execute_router(stmt: Any, params: Any = None) -> MagicMock:
|
||
sql = str(stmt).upper()
|
||
result = MagicMock()
|
||
if "FROM CADASTRE_JOBS" in sql and "WHERE JOB_ID" in sql and "SELECT *" in str(stmt):
|
||
result.mappings.return_value.first.return_value = job
|
||
elif "FROM CAD_QUARTER_STATS" in sql:
|
||
result.scalars.return_value.all.return_value = fresh_quarters
|
||
else:
|
||
# UPDATE / other queries
|
||
result.scalars.return_value.all.return_value = []
|
||
result.mappings.return_value.first.return_value = None
|
||
return result
|
||
|
||
db.execute = execute_router
|
||
return db
|
||
|
||
|
||
def test_enqueue_skips_fresh_quarters_with_default_threshold() -> None:
|
||
"""Fix H: фреш кварталы (cad_quarter_stats.fetched_at < 1h) пропускаются."""
|
||
from app.workers.tasks import scrape_cadastre
|
||
|
||
quarters = ["66:41:0001001", "66:41:0001002", "66:41:0001003"]
|
||
fresh = ["66:41:0001001", "66:41:0001002"] # 2 свежих, 1 stale
|
||
|
||
job = _make_mock_job(quarters)
|
||
db = _make_mock_db(job, fresh_quarters=fresh)
|
||
|
||
with (
|
||
patch.object(scrape_cadastre, "SessionLocal", return_value=db),
|
||
patch.object(scrape_cadastre.bulk_harvest_quarter_task, "apply_async") as mock_apply,
|
||
):
|
||
result = scrape_cadastre.enqueue_cadastre_harvest.run(job_id=99)
|
||
|
||
# Только 1 stale quarter должен попасть в apply_async
|
||
assert mock_apply.call_count == 1
|
||
assert mock_apply.call_args[1]["args"] == ["66:41:0001003", 99]
|
||
assert result["enqueued"] == 1
|
||
assert result["skipped_fresh"] == 2
|
||
|
||
|
||
def test_enqueue_no_skip_when_skip_fresh_hours_zero() -> None:
|
||
"""skip_fresh_hours=0 → отключает freshness check, все quarters enqueued."""
|
||
from app.workers.tasks import scrape_cadastre
|
||
|
||
quarters = ["66:41:0001001", "66:41:0001002"]
|
||
|
||
job = _make_mock_job(quarters, raw_targets_extra={"skip_fresh_hours": 0})
|
||
db = _make_mock_db(job, fresh_quarters=["66:41:0001001"]) # один свежий
|
||
|
||
with (
|
||
patch.object(scrape_cadastre, "SessionLocal", return_value=db),
|
||
patch.object(scrape_cadastre.bulk_harvest_quarter_task, "apply_async") as mock_apply,
|
||
):
|
||
result = scrape_cadastre.enqueue_cadastre_harvest.run(job_id=99)
|
||
|
||
# Все 2 quarters должны быть enqueued (skip отключен)
|
||
assert mock_apply.call_count == 2
|
||
assert result["enqueued"] == 2
|
||
assert result["skipped_fresh"] == 0
|
||
|
||
|
||
def test_enqueue_freshness_threshold_passed_in_sql() -> None:
|
||
"""skip_fresh_hours=2 → threshold = NOW() - 2h в SQL query."""
|
||
from app.workers.tasks import scrape_cadastre
|
||
|
||
quarters = ["66:41:0001001"]
|
||
captured_params: list[dict[str, Any]] = []
|
||
|
||
job = _make_mock_job(quarters, raw_targets_extra={"skip_fresh_hours": 2})
|
||
|
||
def execute_router(stmt: Any, params: Any = None) -> MagicMock:
|
||
sql = str(stmt).upper()
|
||
result = MagicMock()
|
||
if "FROM CADASTRE_JOBS" in sql and "SELECT *" in str(stmt):
|
||
result.mappings.return_value.first.return_value = job
|
||
elif "FROM CAD_QUARTER_STATS" in sql:
|
||
if params:
|
||
captured_params.append(params)
|
||
result.scalars.return_value.all.return_value = []
|
||
else:
|
||
result.scalars.return_value.all.return_value = []
|
||
return result
|
||
|
||
db = MagicMock()
|
||
db.execute = execute_router
|
||
|
||
with (
|
||
patch.object(scrape_cadastre, "SessionLocal", return_value=db),
|
||
patch.object(scrape_cadastre.bulk_harvest_quarter_task, "apply_async"),
|
||
):
|
||
scrape_cadastre.enqueue_cadastre_harvest.run(job_id=99)
|
||
|
||
assert len(captured_params) == 1
|
||
threshold: _dt.datetime = captured_params[0]["t"]
|
||
now = _dt.datetime.now(_dt.UTC)
|
||
# Threshold должен быть приблизительно 2h назад (±5s tolerance)
|
||
delta = now - threshold
|
||
assert _dt.timedelta(hours=1, minutes=59) < delta < _dt.timedelta(hours=2, minutes=1)
|
||
|
||
|
||
# ── Bug #1236: permanent non-WAF fail counter ──────────────────────────────
|
||
|
||
|
||
def _make_counter_db() -> tuple[MagicMock, list[dict[str, Any]]]:
|
||
"""Mock db, capturing UPDATE cadastre_jobs param dicts in chronological order."""
|
||
captured: list[dict[str, Any]] = []
|
||
db = MagicMock()
|
||
|
||
def execute_router(stmt: Any, params: Any = None) -> MagicMock:
|
||
sql = str(stmt).upper()
|
||
result = MagicMock()
|
||
if "UPDATE CADASTRE_JOBS" in sql:
|
||
captured.append({"sql": sql, "params": params or {}})
|
||
result.rowcount = 1
|
||
elif "SELECT TARGETS_TOTAL" in sql:
|
||
# _maybe_finish_job: эмулируем done+failed < total → finish не дёрнется,
|
||
# тест проверяет ТОЛЬКО факт вызова + targets_failed_inc.
|
||
result.mappings.return_value.first.return_value = {
|
||
"targets_total": 10,
|
||
"targets_done": 0,
|
||
"targets_failed": 1,
|
||
"status": "running",
|
||
}
|
||
else:
|
||
result.mappings.return_value.first.return_value = None
|
||
result.scalars.return_value.all.return_value = []
|
||
return result
|
||
|
||
db.execute = execute_router
|
||
return db, captured
|
||
|
||
|
||
def test_bulk_harvest_quarter_final_fail_increments_targets_failed() -> None:
|
||
"""Bug #1236: на финальной (retries == max_retries) попытке non-WAF фейл
|
||
инкрементит targets_failed_inc=1 и вызывает _maybe_finish_job."""
|
||
from app.workers.tasks import scrape_cadastre
|
||
|
||
db, captured = _make_counter_db()
|
||
# Эмулируем `self` Celery task: финальная попытка autoretry.
|
||
self_mock = SimpleNamespace(
|
||
max_retries=3,
|
||
request=SimpleNamespace(retries=3),
|
||
)
|
||
|
||
finish_called: list[int] = []
|
||
|
||
def fake_finish(_db: Any, job_id: int) -> None:
|
||
finish_called.append(job_id)
|
||
|
||
with (
|
||
patch.object(scrape_cadastre, "SessionLocal", return_value=db),
|
||
patch.object(scrape_cadastre, "_maybe_finish_job", side_effect=fake_finish),
|
||
patch.object(
|
||
scrape_cadastre,
|
||
"_is_job_cancelled",
|
||
return_value=False,
|
||
),
|
||
patch(
|
||
"app.services.cadastre.bulk_harvest.harvest_quarter",
|
||
side_effect=RuntimeError("boom"),
|
||
),
|
||
):
|
||
with pytest.raises(RuntimeError):
|
||
# Вызываем underlying function через .run (не apply_async).
|
||
scrape_cadastre.bulk_harvest_quarter_task.__wrapped__.__func__(
|
||
self_mock, "66:41:0001001", 42
|
||
)
|
||
|
||
# В except-ветке ровно один UPDATE с targets_failed_inc=1 и last_error.
|
||
fail_updates = [
|
||
c for c in captured if c["params"].get("failed_inc") == 1 and "last_error" in c["params"]
|
||
]
|
||
assert len(fail_updates) == 1, f"expected single fail-update, got: {captured}"
|
||
assert fail_updates[0]["params"]["id"] == 42
|
||
assert finish_called == [42], "_maybe_finish_job must be called on final attempt"
|
||
|
||
|
||
def test_bulk_harvest_quarter_intermediate_fail_does_not_increment() -> None:
|
||
"""Bug #1236: промежуточная (retries < max_retries) попытка НЕ трогает
|
||
targets_failed_inc и НЕ вызывает _maybe_finish_job — Celery re-enqueue'ит."""
|
||
from app.workers.tasks import scrape_cadastre
|
||
|
||
db, captured = _make_counter_db()
|
||
self_mock = SimpleNamespace(
|
||
max_retries=3,
|
||
request=SimpleNamespace(retries=1), # ещё две попытки впереди
|
||
)
|
||
|
||
finish_called: list[int] = []
|
||
|
||
with (
|
||
patch.object(scrape_cadastre, "SessionLocal", return_value=db),
|
||
patch.object(
|
||
scrape_cadastre,
|
||
"_maybe_finish_job",
|
||
side_effect=lambda _db, jid: finish_called.append(jid),
|
||
),
|
||
patch.object(scrape_cadastre, "_is_job_cancelled", return_value=False),
|
||
patch(
|
||
"app.services.cadastre.bulk_harvest.harvest_quarter",
|
||
side_effect=RuntimeError("transient"),
|
||
),
|
||
):
|
||
with pytest.raises(RuntimeError):
|
||
scrape_cadastre.bulk_harvest_quarter_task.__wrapped__.__func__(
|
||
self_mock, "66:41:0001002", 43
|
||
)
|
||
|
||
fail_updates = [c for c in captured if c["params"].get("failed_inc")]
|
||
assert fail_updates == [], "intermediate retry must NOT bump targets_failed"
|
||
assert finish_called == [], "intermediate retry must NOT call _maybe_finish_job"
|
||
|
||
|
||
# ── Bug #1236: beat schedule registers cadastre-zombie-cleanup ─────────────
|
||
|
||
|
||
def test_beat_schedule_includes_cadastre_zombie_cleanup() -> None:
|
||
"""Bug #1236: cleanup_cadastre_zombies должен быть зарегистрирован в beat,
|
||
иначе застрявший в 'running' job никто не переведёт в 'paused'."""
|
||
from app.workers import beat_schedule
|
||
|
||
# build_beat_schedule читает DB → мокаем _build_beat_schedule_from_db,
|
||
# чтобы тест не зависел от живого Postgres.
|
||
with patch.object(beat_schedule, "_build_beat_schedule_from_db", return_value={}):
|
||
schedule = beat_schedule.build_beat_schedule()
|
||
|
||
assert "cadastre-zombie-cleanup" in schedule, (
|
||
f"cadastre-zombie-cleanup отсутствует в beat: {sorted(schedule.keys())}"
|
||
)
|
||
entry = schedule["cadastre-zombie-cleanup"]
|
||
assert entry["task"] == "tasks.cadastre.cleanup_zombies"
|