Light1YT
c646a71001
Adversarial review found the single-newest last_success_work basis fragile: a load cycle = MULTIPLE done-runs (objective per group_name, kn per region + resume), so the newest run alone misrepresents the cycle. Replace with SUM(work_col) FILTER(done) over the source's own fresh_days window. - recent_output = SUM(COALESCE(work_col,0)) FILTER done in fresh_days window (make_interval secs => fresh_days*86400; no :: cast; NULL-counter -> 0) - downgrade when status==ok and recent_output < min_output_rows (strict <) - registry floors kept: kn_flats=50000 (healthy ~376k, broken sum <=3670), objective=1000 (SUM 7d ~946k); comments rewritten to cycle-sum basis + kn_flats zombie-resume KNOWN LIMITATION - window=fresh_days keeps the #1947 aging-FP fix (7-8d run stays in window) - tests: low-output-cycle->failed, above-floor->ok, boundary==floor->ok, weekly-aging->ok, age-stale precedence, registry floors Verified on prod: domrf_kn_flats latest snapshot 2026-06-22=9 flats vs healthy 2026-05-17=376604 (broken 5+ weeks); objective SUM(rows_lots,7d)=946264.
575 lines
25 KiB
Python
575 lines
25 KiB
Python
"""Тесты для GET /api/v1/admin/scrape/freshness (#73 B5-3).
|
||
|
||
Эндпоинт читает 6 источников по 5 run-/data-таблицам (kn / kn_flats / objective /
|
||
nspd / nspd_geo / cadastre — kn и kn_flats делят kn_scrape_runs, разные work_col)
|
||
и для каждого считает свежесть. Здесь DB замокан: на каждый per-source SELECT
|
||
возвращаем заранее заданную строку, чтобы проверить классификацию
|
||
fresh/stale/critical, маппинг status (ok/stale/failed), output-floor downgrade
|
||
(#1945 / redesigned #1947 — recent_output = SUM(work_col) по done-прогонам цикла в
|
||
окне fresh_days < min_output_rows) и агрегат 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",
|
||
"recent_output": 376000, # здоровый цикл flats_count → kn_flats ok
|
||
},
|
||
"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",
|
||
"recent_output": 325000,
|
||
},
|
||
# 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
|
||
# kn / kn_flats / objective / nspd / nspd_geo / cadastre.
|
||
assert len(body["sources"]) == len(admin_scrape._FRESHNESS_SOURCES)
|
||
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",
|
||
"recent_output": 100000, # выше любого floor → flagged-источники ok
|
||
}
|
||
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",
|
||
"recent_output": 100000, # выше любого floor → flagged-источники ok
|
||
}
|
||
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",
|
||
"recent_output": 100000,
|
||
}
|
||
rows = {s.table: dict(base) for s in admin_scrape._FRESHNESS_SOURCES}
|
||
# objective критичный — обнулим успех (last_success=None → нет done-прогона,
|
||
# recent_output=None → floor не применяется, status=failed по отсутствию успеха).
|
||
rows["objective_scrape_runs"] = {
|
||
"last_success_at": None,
|
||
"last_attempt_at": _ago(5.0),
|
||
"upd_24h": 0,
|
||
"upd_7d": 0,
|
||
"last_status": "running",
|
||
"recent_output": None,
|
||
}
|
||
_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"
|
||
|
||
|
||
# NB по мокам output-floor: _make_db ключуется по ИМЕНИ таблицы → kn и kn_flats
|
||
# делят одну kn_scrape_runs-строку, поэтому оба видят ОДИН recent_output (per-work_col
|
||
# дивергенция objects_count vs flats_count есть только в живом SQL, не в моке). Здесь
|
||
# мок проверяет Python-сравнение (recent_output < min_output_rows + все guard'ы);
|
||
# семантику оконной SQL-SUM подтвердили на проде (objective SUM(rows_lots,7d)=946264,
|
||
# kn_flats SUM(flats_count,8d)=9 — gendesign-postgres-1, #1947).
|
||
|
||
|
||
def test_output_floor_downgrades_low_output_cycle() -> None:
|
||
"""issue #1945 / #1947: kn_flats прогнался успешно И недавно (был бы fresh→ok), но
|
||
recent_output=SUM(flats_count) по циклу=9 < min_output_rows=50000 → failed (alertable).
|
||
|
||
Прод-сценарий: цикл дал суммарно flats_count=9 (latest done = resume, считает хвост)
|
||
при здоровых ~376k. kn БЕЗ флага (alert_on_zero_output=False) → branch не входит,
|
||
остаётся ok даже читая ту же низкую recent_output (no false positive).
|
||
"""
|
||
# Сломанный цикл: свежий (1d), но recent_output (SUM flats_count за 8d)=9.
|
||
broken_kn = {
|
||
"last_success_at": _ago(1.0), # < fresh_days=8 → был бы fresh
|
||
"last_attempt_at": _ago(1.0),
|
||
"upd_24h": 9,
|
||
"upd_7d": 9,
|
||
"last_status": "done",
|
||
"recent_output": 9, # SUM(flats_count) по done-прогонам цикла ≪ 50000
|
||
}
|
||
# Остальные источники свежие+здоровые, чтобы изолировать поведение kn/kn_flats.
|
||
healthy = {
|
||
"last_success_at": _ago(0.5),
|
||
"last_attempt_at": _ago(0.5),
|
||
"upd_24h": 100,
|
||
"upd_7d": 1000,
|
||
"last_status": "done",
|
||
"recent_output": 100000, # выше любого floor
|
||
}
|
||
rows: dict[str, dict[str, Any]] = {}
|
||
for s in admin_scrape._FRESHNESS_SOURCES:
|
||
if s.timestamp_col is not None:
|
||
rows[s.table] = {**healthy, "last_status": None, "recent_output": None}
|
||
elif s.table == "kn_scrape_runs":
|
||
rows[s.table] = dict(broken_kn)
|
||
else:
|
||
rows[s.table] = dict(healthy)
|
||
_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()
|
||
by_src = {s["source"]: s for s in body["sources"]}
|
||
|
||
# kn_flats: otherwise-fresh, но recent_output=9 < 50000 → DOWNGRADE до failed.
|
||
assert by_src["kn_flats"]["freshness"] == "fresh" # по возрасту был бы fresh
|
||
assert by_src["kn_flats"]["status"] == "failed" # output-floor downgrade
|
||
assert by_src["kn_flats"]["recent_output"] == 9
|
||
assert by_src["kn_flats"]["reason"] is not None
|
||
assert "flats_count" in by_src["kn_flats"]["reason"]
|
||
assert "9" in by_src["kn_flats"]["reason"]
|
||
|
||
# kn — alert_on_zero_output=False гейтит ВЕСЬ branch → ok, даже читая recent_output=9
|
||
# из той же mocked-строки (per-work_col дивергенция есть только в живом SQL).
|
||
assert by_src["kn"]["status"] == "ok"
|
||
assert by_src["kn"]["reason"] is None
|
||
|
||
# critical kn_flats упал → overall_status=failed.
|
||
assert by_src["kn_flats"]["critical"] is True
|
||
assert body["overall_status"] == "failed"
|
||
|
||
|
||
def test_output_floor_no_false_positive_when_above_floor() -> None:
|
||
"""alert_on_zero_output источник с recent_output ВЫШЕ floor → остаётся ok.
|
||
|
||
Регрессия: output-floor НЕ должен бить по здоровому источнику, чей цикл извлёк
|
||
нормальный объём (kn_flats ~376k > 50000; objective ~946k > 1000).
|
||
"""
|
||
healthy = {
|
||
"last_success_at": _ago(1.0),
|
||
"last_attempt_at": _ago(1.0),
|
||
"upd_24h": 12,
|
||
"upd_7d": 1234,
|
||
"last_status": "done",
|
||
"recent_output": 376000, # здоровый цикл ≫ всех floor'ов
|
||
}
|
||
rows: dict[str, dict[str, Any]] = {}
|
||
for s in admin_scrape._FRESHNESS_SOURCES:
|
||
rows[s.table] = (
|
||
{**healthy, "last_status": None, "recent_output": None}
|
||
if s.timestamp_col is not None
|
||
else dict(healthy)
|
||
)
|
||
_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"]}
|
||
# Все alert_on_zero_output-источники с выходом выше floor → ok, reason=None.
|
||
for s in admin_scrape._FRESHNESS_SOURCES:
|
||
if s.alert_on_zero_output:
|
||
assert by_src[s.source]["status"] == "ok", s.source
|
||
assert by_src[s.source]["reason"] is None, s.source
|
||
|
||
|
||
def test_output_floor_boundary_equal_floor_stays_ok() -> None:
|
||
"""Граница: recent_output == min_output_rows → остаётся ok (проверка строгое '<').
|
||
|
||
kn_flats recent_output ровно 50000 (= floor) → НЕ даунгрейдится (порог = минимально-
|
||
приемлемый объём, на нём ещё ok). Регрессия на off-by-one в сравнении.
|
||
"""
|
||
kn_flats = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == "kn_flats")
|
||
at_floor = {
|
||
"last_success_at": _ago(1.0),
|
||
"last_attempt_at": _ago(1.0),
|
||
"upd_24h": 0,
|
||
"upd_7d": kn_flats.min_output_rows,
|
||
"last_status": "done",
|
||
"recent_output": kn_flats.min_output_rows, # ровно на пороге → ok
|
||
}
|
||
healthy = {
|
||
"last_success_at": _ago(0.5),
|
||
"last_attempt_at": _ago(0.5),
|
||
"upd_24h": 100,
|
||
"upd_7d": 1000,
|
||
"last_status": "done",
|
||
"recent_output": 1000000,
|
||
}
|
||
rows: dict[str, dict[str, Any]] = {}
|
||
for s in admin_scrape._FRESHNESS_SOURCES:
|
||
if s.timestamp_col is not None:
|
||
rows[s.table] = {**healthy, "last_status": None, "recent_output": None}
|
||
elif s.table == "kn_scrape_runs":
|
||
rows[s.table] = dict(at_floor)
|
||
else:
|
||
rows[s.table] = dict(healthy)
|
||
_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["kn_flats"]["recent_output"] == kn_flats.min_output_rows
|
||
assert by_src["kn_flats"]["status"] == "ok" # ровно на пороге → НЕ failed
|
||
assert by_src["kn_flats"]["reason"] is None
|
||
|
||
|
||
def test_output_floor_no_false_positive_on_weekly_window_aging() -> None:
|
||
"""#1947 fix: здоровый недельный прогон возрастом 7–8d остаётся в окне fresh_days=8d
|
||
→ recent_output здоров (376k) → НЕ ложно-флагуется.
|
||
|
||
Регрессия на самый первый #1947-баг: оконная SUM была хардкод 7d, поэтому прогон
|
||
7–8d (>7d назад) давал SUM=0 при status=ok → ложный downgrade ~раз/неделю. Теперь
|
||
окно = fresh_days (8d) → прогон 7.5d ВНУТРИ окна, recent_output=376k > floor → ok.
|
||
upd_7d=0 (мок старого 7d-счётчика) НЕ влияет — floor смотрит recent_output.
|
||
"""
|
||
weekly_aged = {
|
||
"last_success_at": _ago(7.5), # < fresh_days=8 → всё ещё fresh, и внутри окна
|
||
"last_attempt_at": _ago(7.5),
|
||
"upd_24h": 0,
|
||
"upd_7d": 0, # старый 7d-счётчик: прогон >7d назад → 0 (раньше ложно флагал)
|
||
"last_status": "done",
|
||
"recent_output": 376000, # окно fresh_days=8d ловит прогон 7.5d → цикл здоров
|
||
}
|
||
healthy = {
|
||
"last_success_at": _ago(0.5),
|
||
"last_attempt_at": _ago(0.5),
|
||
"upd_24h": 50,
|
||
"upd_7d": 500,
|
||
"last_status": "done",
|
||
"recent_output": 100000,
|
||
}
|
||
rows: dict[str, dict[str, Any]] = {}
|
||
for s in admin_scrape._FRESHNESS_SOURCES:
|
||
if s.timestamp_col is not None:
|
||
rows[s.table] = {**healthy, "last_status": None, "recent_output": None}
|
||
elif s.table == "kn_scrape_runs":
|
||
rows[s.table] = dict(weekly_aged)
|
||
else:
|
||
rows[s.table] = dict(healthy)
|
||
_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"]}
|
||
# kn_flats: fresh по возрасту (7.5<8), upd_7d=0 НЕ важен — floor смотрит
|
||
# recent_output (376k > 50000, окно fresh_days покрыло прогон) → ok, без downgrade.
|
||
assert by_src["kn_flats"]["freshness"] == "fresh"
|
||
assert by_src["kn_flats"]["status"] == "ok"
|
||
assert by_src["kn_flats"]["objects_updated_7d"] == 0
|
||
assert by_src["kn_flats"]["recent_output"] == 376000
|
||
assert by_src["kn_flats"]["reason"] is None
|
||
|
||
|
||
def test_output_floor_not_applied_when_already_stale() -> None:
|
||
"""Если источник уже stale/failed по возрасту — output-floor НЕ передоунгрейдит
|
||
(age уже покрывает), и не двойнит. objective: старый успех (10d → stale) +
|
||
recent_output=0 → остаётся stale (НЕ перепрыгивает в failed через floor)."""
|
||
base_fresh = {
|
||
"last_success_at": _ago(0.5),
|
||
"last_attempt_at": _ago(0.5),
|
||
"upd_24h": 5,
|
||
"upd_7d": 50,
|
||
"last_status": "done",
|
||
"recent_output": 500000,
|
||
}
|
||
rows: dict[str, dict[str, Any]] = {}
|
||
for s in admin_scrape._FRESHNESS_SOURCES:
|
||
rows[s.table] = (
|
||
{**base_fresh, "last_status": None, "recent_output": None}
|
||
if s.timestamp_col is not None
|
||
else dict(base_fresh)
|
||
)
|
||
# objective: stale по возрасту (10d: fresh<7, stale<14) + нулевой выход.
|
||
rows["objective_scrape_runs"] = {
|
||
"last_success_at": _ago(10.0),
|
||
"last_attempt_at": _ago(10.0),
|
||
"upd_24h": 0,
|
||
"upd_7d": 0,
|
||
"last_status": "done",
|
||
"recent_output": 0,
|
||
}
|
||
_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"]}
|
||
# output-floor применяется ТОЛЬКО к otherwise-ok; stale остаётся stale, reason=None.
|
||
assert by_src["objective"]["status"] == "stale"
|
||
assert by_src["objective"]["reason"] is None
|
||
|
||
|
||
def test_kn_flats_registry_config() -> None:
|
||
"""kn_flats: та же run-таблица что у kn, но work_col=flats_count + alert_on_zero_output
|
||
+ min_output_rows=50000, критичный, недельные пороги. kn (objects_count) — без флага."""
|
||
kn_flats = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == "kn_flats")
|
||
assert kn_flats.table == "kn_scrape_runs"
|
||
assert kn_flats.work_col == "flats_count"
|
||
assert kn_flats.timestamp_col is None # run-ledger режим
|
||
assert kn_flats.alert_on_zero_output is True
|
||
# healthy цикл ~376k ≫ floor ≫ broken-sum ≤3670 (прод #1947).
|
||
assert kn_flats.min_output_rows == 50000
|
||
assert kn_flats.critical is True
|
||
assert kn_flats.fresh_days == 8.0
|
||
assert kn_flats.stale_days == 14.0
|
||
|
||
kn = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == "kn")
|
||
assert kn.work_col == "objects_count"
|
||
assert kn.alert_on_zero_output is False # существующий источник unchanged
|
||
assert kn.min_output_rows == 1 # default — не проверяется без флага
|
||
|
||
objective = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == "objective")
|
||
assert objective.alert_on_zero_output is True
|
||
# healthy SUM(rows_lots,7d)~946k по 4 группам; floor 1000 с огромным запасом.
|
||
assert objective.min_output_rows == 1000
|
||
|
||
# Legitimate-low источники НЕ должны иметь флаг (избегаем ложных срабатываний);
|
||
# min_output_rows остаётся default=1 и не проверяется.
|
||
for name in ("nspd", "nspd_geo", "cadastre"):
|
||
src = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == name)
|
||
assert src.alert_on_zero_output is False, name
|
||
assert src.min_output_rows == 1, name
|
||
|
||
|
||
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"
|