f526a43ef9
Bot review on PR #187 caught: 1. cadastre_jobs has no extra_config column → Fix H crashed on KeyError 2. Missing tests for Fix H + Fix I 3. _parse_nspd_date didn't validate date ranges Changes: - Pivot Fix H: read skip_fresh_hours from raw_targets JSONB (existing column), no schema migration needed - _parse_nspd_date: validate via date.fromisoformat() — rejects 2023-13-45 - Add 3 tests for Fix H in backend/tests/workers/test_scrape_cadastre.py - Add 2 tests for Fix I heartbeat callbacks in test_cadastre_bulk.py - Add 1 test for parse_nspd_date range validation (was: only format) 49 tests passing.
132 lines
4.9 KiB
Python
132 lines
4.9 KiB
Python
"""Тесты для backend/app/workers/tasks/scrape_cadastre.py.
|
||
|
||
Покрывает Fix H — per-quarter freshness skip в enqueue_cadastre_harvest.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import datetime as _dt
|
||
from typing import Any
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
|
||
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)
|