bot-backend
67d41df880
nspd freshness watched a defunct ledger (nspd_scrape_runs, WAF-banned manual April runs) → false "no successful runs". Add data-table mode (timestamp_col) + repoint nspd to live nspd_quarter_dumps.fetched_at_utc (fresh<14/stale<30). Run-ledger sources untouched. Refs #73
292 lines
11 KiB
Python
292 lines
11 KiB
Python
"""Тесты для GET /api/v1/admin/scrape/freshness (#73 B5-3).
|
||
|
||
Эндпоинт читает 5 run-таблиц (kn / objective / nspd / nspd_geo / cadastre) и
|
||
для каждой считает свежесть. Здесь DB замокан: на каждый per-source SELECT
|
||
возвращаем заранее заданную строку, чтобы проверить классификацию
|
||
fresh/stale/critical, маппинг status (ok/stale/failed) и агрегат overall_status
|
||
по критичным источникам.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import UTC, datetime, timedelta
|
||
from typing import Any
|
||
from unittest.mock import MagicMock
|
||
|
||
from fastapi.testclient import TestClient
|
||
|
||
from app.api.v1 import admin_scrape
|
||
from app.core.db import get_db
|
||
from app.main import app
|
||
|
||
ENDPOINT = "/api/v1/admin/scrape/freshness"
|
||
|
||
|
||
def _ago(days: float) -> datetime:
|
||
return datetime.now(UTC) - timedelta(days=days)
|
||
|
||
|
||
def _make_db(per_source: dict[str, dict[str, Any]]) -> MagicMock:
|
||
"""Db-mock: каждый execute()'s SELECT относится к одной run-таблице; матчим по
|
||
имени таблицы в SQL и возвращаем соответствующую заранее заданную строку.
|
||
|
||
per_source: ключ — имя run-таблицы, значение — mapping-строка (как .first()).
|
||
"""
|
||
db = MagicMock()
|
||
|
||
def execute(stmt: Any, params: Any = None) -> Any:
|
||
sql = str(stmt)
|
||
matched: dict[str, Any] | None = None
|
||
# Матчим по FROM <table> — самый длинный матч (nspd_geo_jobs vs nspd_*).
|
||
for table in sorted(per_source, key=len, reverse=True):
|
||
if f"FROM {table}" in sql:
|
||
matched = per_source[table]
|
||
break
|
||
result = MagicMock()
|
||
result.mappings.return_value.first.return_value = matched
|
||
return result
|
||
|
||
db.execute = execute
|
||
return db
|
||
|
||
|
||
def _override_db(db: MagicMock) -> None:
|
||
def _factory():
|
||
yield db
|
||
|
||
app.dependency_overrides[get_db] = _factory
|
||
|
||
|
||
def test_freshness_mixed_statuses() -> None:
|
||
"""kn свежий (ok), objective stale, nspd (data-table) свежий (ok), nspd_geo critical,
|
||
cadastre fresh. overall_status = stale (по критичному objective→stale)."""
|
||
rows = {
|
||
"kn_scrape_runs": {
|
||
"last_success_at": _ago(1.0), # < fresh_days=2 → fresh → ok
|
||
"last_attempt_at": _ago(1.0),
|
||
"upd_24h": 1500,
|
||
"upd_7d": 1500,
|
||
"last_status": "done",
|
||
},
|
||
"objective_scrape_runs": {
|
||
"last_success_at": _ago(10.0), # fresh<7, stale<14 → stale
|
||
"last_attempt_at": _ago(10.0),
|
||
"upd_24h": 0,
|
||
"upd_7d": 325000,
|
||
"last_status": "done",
|
||
},
|
||
# nspd теперь data-table режим (nspd_quarter_dumps.fetched_at_utc):
|
||
# last_success_at = MAX(timestamp_col), last_status = NULL (нет run-ledger).
|
||
"nspd_quarter_dumps": {
|
||
"last_success_at": _ago(4.0), # < fresh_days=14 → fresh → ok
|
||
"last_attempt_at": _ago(4.0),
|
||
"upd_24h": 5,
|
||
"upd_7d": 56,
|
||
"last_status": None,
|
||
},
|
||
"nspd_geo_jobs": {
|
||
"last_success_at": _ago(40.0), # >= stale_days=30 → critical → failed
|
||
"last_attempt_at": _ago(40.0),
|
||
"upd_24h": 0,
|
||
"upd_7d": 0,
|
||
"last_status": "done",
|
||
},
|
||
"cadastre_jobs": {
|
||
"last_success_at": _ago(3.0), # < fresh_days=14 → fresh → ok
|
||
"last_attempt_at": _ago(3.0),
|
||
"upd_24h": 5,
|
||
"upd_7d": 5,
|
||
"last_status": "done",
|
||
},
|
||
}
|
||
_override_db(_make_db(rows))
|
||
try:
|
||
resp = TestClient(app).get(ENDPOINT)
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert resp.status_code == 200, resp.text
|
||
body = resp.json()
|
||
assert "generated_at" in body
|
||
assert len(body["sources"]) == 5
|
||
by_src = {s["source"]: s for s in body["sources"]}
|
||
|
||
assert by_src["kn"]["status"] == "ok"
|
||
assert by_src["kn"]["freshness"] == "fresh"
|
||
assert by_src["kn"]["objects_updated_7d"] == 1500
|
||
assert by_src["kn"]["critical"] is True
|
||
|
||
assert by_src["objective"]["status"] == "stale"
|
||
assert by_src["objective"]["freshness"] == "stale"
|
||
|
||
# nspd — data-table режим: свежий dump → ok, last_status=NULL не ломает маппинг.
|
||
assert by_src["nspd"]["status"] == "ok"
|
||
assert by_src["nspd"]["freshness"] == "fresh"
|
||
assert by_src["nspd"]["last_status"] is None
|
||
assert by_src["nspd"]["objects_updated_7d"] == 56
|
||
|
||
assert by_src["nspd_geo"]["status"] == "failed"
|
||
assert by_src["nspd_geo"]["freshness"] == "critical"
|
||
|
||
assert by_src["cadastre"]["status"] == "ok"
|
||
|
||
# overall_status — худший по критичным (kn=ok, objective=stale) → stale.
|
||
# nspd_geo критичным НЕ помечен, поэтому его failed в overall не попадает.
|
||
assert by_src["nspd_geo"]["critical"] is False
|
||
assert body["overall_status"] == "stale"
|
||
|
||
|
||
def _healthy_run_rows_except_nspd(nspd_row: dict[str, Any]) -> dict[str, dict[str, Any]]:
|
||
"""Все run-ledger источники свежие, nspd — отдельной data-table строкой.
|
||
|
||
Используется для изоляции поведения nspd-источника в data-table режиме, не
|
||
затрагивая остальные источники.
|
||
"""
|
||
fresh_run = {
|
||
"last_success_at": _ago(0.5),
|
||
"last_attempt_at": _ago(0.5),
|
||
"upd_24h": 10,
|
||
"upd_7d": 100,
|
||
"last_status": "done",
|
||
}
|
||
rows: dict[str, dict[str, Any]] = {}
|
||
for s in admin_scrape._FRESHNESS_SOURCES:
|
||
rows[s.table] = nspd_row if s.source == "nspd" else dict(fresh_run)
|
||
return rows
|
||
|
||
|
||
def test_nspd_uses_timestamp_col_mode_fresh() -> None:
|
||
"""nspd теперь читает data-таблицу nspd_quarter_dumps по timestamp_col=fetched_at_utc.
|
||
|
||
Свежий dump (~4d < fresh_days=14) + last_status=NULL → freshness=fresh, status=ok
|
||
(НЕ failed). Регрессия на repoint defunct nspd_scrape_runs → живой harvest-dump.
|
||
"""
|
||
nspd = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == "nspd")
|
||
# Подтверждаем что источник в data-table режиме и указывает на живую таблицу.
|
||
assert nspd.timestamp_col == "fetched_at_utc"
|
||
assert nspd.table == "nspd_quarter_dumps"
|
||
assert nspd.critical is False # остаётся некритичным
|
||
|
||
rows = _healthy_run_rows_except_nspd(
|
||
{
|
||
"last_success_at": _ago(4.0), # < fresh_days=14 → fresh
|
||
"last_attempt_at": _ago(4.0),
|
||
"upd_24h": 5,
|
||
"upd_7d": 56,
|
||
"last_status": None, # data-table → NULL, маппинг не должен падать
|
||
}
|
||
)
|
||
_override_db(_make_db(rows))
|
||
try:
|
||
resp = TestClient(app).get(ENDPOINT)
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert resp.status_code == 200, resp.text
|
||
by_src = {s["source"]: s for s in resp.json()["sources"]}
|
||
assert by_src["nspd"]["freshness"] == "fresh"
|
||
assert by_src["nspd"]["status"] == "ok"
|
||
assert by_src["nspd"]["last_status"] is None
|
||
assert by_src["nspd"]["objects_updated_7d"] == 56
|
||
assert by_src["nspd"]["age_days"] is not None and by_src["nspd"]["age_days"] < 14.0
|
||
|
||
|
||
def test_nspd_timestamp_col_stale_dump_is_critical() -> None:
|
||
"""Старый dump (40d >= stale_days=30) в data-table режиме → critical → status=failed."""
|
||
rows = _healthy_run_rows_except_nspd(
|
||
{
|
||
"last_success_at": _ago(40.0), # >= stale_days=30 → critical
|
||
"last_attempt_at": _ago(40.0),
|
||
"upd_24h": 0,
|
||
"upd_7d": 0,
|
||
"last_status": None,
|
||
}
|
||
)
|
||
_override_db(_make_db(rows))
|
||
try:
|
||
resp = TestClient(app).get(ENDPOINT)
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert resp.status_code == 200, resp.text
|
||
by_src = {s["source"]: s for s in resp.json()["sources"]}
|
||
assert by_src["nspd"]["freshness"] == "critical"
|
||
assert by_src["nspd"]["status"] == "failed"
|
||
|
||
|
||
def test_freshness_all_healthy() -> None:
|
||
"""Все источники свежие → overall_status=ok."""
|
||
fresh_row = {
|
||
"last_success_at": _ago(0.5),
|
||
"last_attempt_at": _ago(0.5),
|
||
"upd_24h": 10,
|
||
"upd_7d": 100,
|
||
"last_status": "done",
|
||
}
|
||
rows = {
|
||
s.table: dict(fresh_row) for s in admin_scrape._FRESHNESS_SOURCES
|
||
}
|
||
_override_db(_make_db(rows))
|
||
try:
|
||
resp = TestClient(app).get(ENDPOINT)
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert resp.status_code == 200, resp.text
|
||
body = resp.json()
|
||
assert body["overall_status"] == "ok"
|
||
assert all(s["status"] == "ok" for s in body["sources"])
|
||
assert all(s["age_days"] is not None for s in body["sources"])
|
||
|
||
|
||
def test_freshness_critical_source_failed_drives_overall() -> None:
|
||
"""Критичный источник (objective) без успехов → overall_status=failed."""
|
||
base = {
|
||
"last_success_at": _ago(0.5),
|
||
"last_attempt_at": _ago(0.5),
|
||
"upd_24h": 1,
|
||
"upd_7d": 1,
|
||
"last_status": "done",
|
||
}
|
||
rows = {s.table: dict(base) for s in admin_scrape._FRESHNESS_SOURCES}
|
||
# objective критичный — обнулим успех.
|
||
rows["objective_scrape_runs"] = {
|
||
"last_success_at": None,
|
||
"last_attempt_at": _ago(5.0),
|
||
"upd_24h": 0,
|
||
"upd_7d": 0,
|
||
"last_status": "running",
|
||
}
|
||
_override_db(_make_db(rows))
|
||
try:
|
||
resp = TestClient(app).get(ENDPOINT)
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert resp.status_code == 200, resp.text
|
||
body = resp.json()
|
||
assert body["overall_status"] == "failed"
|
||
|
||
|
||
def test_classify_freshness_boundaries() -> None:
|
||
"""Юнит на пороги классификатора: границы fresh/stale/critical, unknown."""
|
||
f = admin_scrape._classify_freshness
|
||
assert f(None, 2.0, 5.0) == "unknown"
|
||
assert f(1.9, 2.0, 5.0) == "fresh"
|
||
assert f(2.0, 2.0, 5.0) == "stale" # ровно fresh_days → уже stale
|
||
assert f(4.9, 2.0, 5.0) == "stale"
|
||
assert f(5.0, 2.0, 5.0) == "critical" # ровно stale_days → critical
|
||
assert f(100.0, 2.0, 5.0) == "critical"
|
||
|
||
|
||
def test_kn_thresholds_match_weekly_cadence() -> None:
|
||
"""kn шедулится еженедельно (Mon cron) → пороги должны покрывать полную неделю,
|
||
иначе ложный 'stale' каждую среду-воскресенье. Регрессия на калибровку."""
|
||
kn = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == "kn")
|
||
f = admin_scrape._classify_freshness
|
||
# В течение недели после понедельничного прогона kn остаётся свежим.
|
||
assert f(3.9, kn.fresh_days, kn.stale_days) == "fresh" # пятница
|
||
assert f(6.5, kn.fresh_days, kn.stale_days) == "fresh" # воскресенье
|
||
# Пропуск одного понедельника → stale; двух → critical.
|
||
assert f(9.0, kn.fresh_days, kn.stale_days) == "stale"
|
||
assert f(15.0, kn.fresh_days, kn.stale_days) == "critical"
|