bot-backend
71d60ff13e
Обработчик landing_showcase_deals безусловно ставил прогону done. У счётчиков витрины (considered/eligible/written) нет результатного ключа кита, поэтому сводка просроченных судит её только по статусу: прогон с written=0 обнулял часы свежести так же, как удачный, а страница тем временем теряла таблицу. Теперь written=0 — mark_failed с причиной и logger.error. Тест идёт путём сводки: обработчик -> freshness_rows -> stale_sources; пустой последний прогон при старом непустом даёт витрину в тревоге, непустой — нет. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
406 lines
23 KiB
Python
406 lines
23 KiB
Python
"""Витрина сделок лэндинга попадает в расписание и в монитор свежести (#3469).
|
||
|
||
ЧТО БЫЛО СЛОМАНО. `landing_showcase_deals` считает витрину публичного лэндинга
|
||
(«МЕРА сказала X — продали за Y»), но в `scrape_schedules` строки для неё не было
|
||
вовсе (`WHERE source LIKE '%showcase%'` — 0 строк на проде 12.09.2026), а в
|
||
реестре `product_handlers` — обработчика. То есть планировщик про задачу не знал
|
||
ни с какой стороны, пересчёт был ручным, и страница показывала прогон
|
||
тринадцатисуточной давности. Заметить это было неоткуда: под таблицей печатались
|
||
счётчики прогона, но не его дата, а сводка просроченных источников
|
||
(`emit_stale_digest`, #2670) ходит по ВКЛЮЧЁННЫМ РАСПИСАНИЯМ — источника, которого
|
||
в таблице нет, для неё не существует.
|
||
|
||
ЧТО ПРОВЕРЯЕТСЯ ЗДЕСЬ, И ПОЧЕМУ ИМЕННО ЭТО.
|
||
|
||
1. Обработчик резолвится ТЕМ ЖЕ `resolve_handler`, которым его ищет боевой
|
||
`_dispatch`. Одной строки расписания мало: без обработчика планировщик
|
||
нашёл бы задачу и не смог её запустить.
|
||
2. Миграция 303 сеет строку, и сеет её ВКЛЮЧЁННОЙ с явным `interval_days` —
|
||
сводка считает порог просрочки из этого же числа.
|
||
3. Сводка краснеет, когда витрина не пересчитывалась дольше ТРЁХ тактов
|
||
(приёмка #3469), и молчит на двух. Число тактов здесь — литерал, а такт
|
||
читается из миграции: ожидание, взятое из той же настройки, которую
|
||
проверяешь, уезжает вместе с ней — см. комментарий при _ACCEPTANCE_CYCLES.
|
||
4. Живой Postgres (само-скип): миграция реально вставляет строку в таблицу,
|
||
повторное применение её не задваивает, и настоящий запрос сводки
|
||
`_STALE_SOURCES_SQL` видит эту строку и отдаёт витрину просроченной.
|
||
|
||
ЖИВОЙ ТЕСТ НЕ УДАЛЯЕТ И НЕ ПРАВИТ НИЧЕГО ЧУЖОГО: он применяет ту же идемпотентную
|
||
миграцию, что применяет деплой (ON CONFLICT DO NOTHING). В CI он ИДЁТ —
|
||
ci-tradein.yml поднимает свой Postgres и кладёт DSN в DATABASE_URL; на машине без
|
||
базы само-скипается (запись в tests/skip_allowlist.txt).
|
||
|
||
ГОНЯТЬ ЕГО НАДО НА ПОЛНОЙ СХЕМЕ, А НЕ НА ПУСТОЙ БАЗЕ. На чистой базе он был
|
||
зелёным и при этом падал в CI: повтор `015_scrape_runs.sql` (его комментарий к
|
||
колонке, снесённой миграцией 214) на полной схеме валится, а на пустой — нет.
|
||
Поэтому зависимости применяются только когда таблицы ещё нет, а проверять надо
|
||
тем же путём, каким гоняет CI:
|
||
|
||
docker exec tradein-postgres psql -U tradein -d postgres -c 'CREATE DATABASE t3469full'
|
||
docker exec -i tradein-postgres psql -U tradein -d t3469full -c \\
|
||
'CREATE EXTENSION postgis; CREATE EXTENSION pg_trgm; CREATE ROLE gendesign_reader;'
|
||
for f in $(ls -1 data/sql/*.sql | sort); do docker exec -i tradein-postgres \\
|
||
psql -U tradein -d t3469full -v ON_ERROR_STOP=on -q < "$f"; done
|
||
DATABASE_URL="postgresql+psycopg://tradein:tradein@127.0.0.1:5433/t3469full" \\
|
||
uv run python -m pytest tests/test_3469_showcase_schedule.py -q
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import re
|
||
from datetime import UTC, datetime, timedelta
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
from typing import Any
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import pytest
|
||
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||
|
||
from scraper_kit.orchestration import scheduler as sched
|
||
|
||
from app.services.product_handlers import _job_landing_showcase_deals, build_product_handlers
|
||
|
||
SOURCE = "landing_showcase_deals"
|
||
|
||
_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
|
||
_MIGRATION = _SQL_DIR / "303_scrape_schedules_seed_landing_showcase_deals.sql"
|
||
# Таблицы, без которых строку расписания некуда класть (FK scrape_schedules →
|
||
# scrape_runs), — живой тест применяет их в том же порядке, что и деплой.
|
||
_DEPS = [
|
||
_SQL_DIR / "015_scrape_runs.sql",
|
||
# counters jsonb — по нему сводка судит, принёс ли прогон данные.
|
||
_SQL_DIR / "051_scrape_runs_extend.sql",
|
||
_SQL_DIR / "052_scrape_schedules.sql",
|
||
]
|
||
|
||
NOW = datetime(2026, 9, 12, 8, 0, tzinfo=UTC)
|
||
|
||
|
||
def _migration_sql() -> str:
|
||
return _MIGRATION.read_text("utf-8")
|
||
|
||
|
||
def _seeded_interval_days() -> int:
|
||
"""Такт из САМОЙ миграции — порог сводки считается из него, не из литерала."""
|
||
m = re.search(r'"interval_days"\s*:\s*(\d+)', _migration_sql())
|
||
assert m is not None, "в default_params миграции 303 нет interval_days"
|
||
return int(m.group(1))
|
||
|
||
|
||
# ── 1. Планировщик видит задачу ──────────────────────────────────────────────
|
||
|
||
|
||
def test_handler_resolves_for_scheduler() -> None:
|
||
"""`resolve_handler` находит витрину — тем же вызовом, что и боевой _dispatch.
|
||
|
||
Ломать так: убрать ключ из реестра в product_handlers — тест покраснеет, а
|
||
планировщик на проде заклеймил бы прогон и не нашёл, чем его выполнить.
|
||
"""
|
||
registry = build_product_handlers(ctx=None) # type: ignore[arg-type]
|
||
handler = sched.resolve_handler(SOURCE, registry)
|
||
assert handler is not None, f"{SOURCE} не резолвится реестром — задача невидима"
|
||
# СРАВНИВАЕМ САМ JOB, А НЕ `log_name`: имя — второй литерал конструктора
|
||
# Handler, и правильный ключ с чужим телом (`_job_landing_stats` под ключом
|
||
# витрины) проходил проверку по имени насквозь. Резолв ведёт к пересчёту
|
||
# витрины или не ведёт — это свойство функции, а не подписи в логе.
|
||
assert handler.job is _job_landing_showcase_deals, (
|
||
f"под ключом {SOURCE} стоит чужой job: {handler.job.__name__}"
|
||
)
|
||
assert handler.log_name == SOURCE
|
||
|
||
|
||
# ── 2. Миграция сеет строку ──────────────────────────────────────────────────
|
||
|
||
|
||
def test_migration_303_exists() -> None:
|
||
assert _MIGRATION.is_file(), f"missing migration: {_MIGRATION}"
|
||
|
||
|
||
def test_migration_303_seeds_source_enabled() -> None:
|
||
sql = _migration_sql()
|
||
assert f"'{SOURCE}'" in sql
|
||
assert "INSERT INTO scrape_schedules" in sql
|
||
# enabled=true — иначе сводка просроченных источников строку не увидит
|
||
# (_STALE_SOURCES_SQL: WHERE sch.enabled), и монитор молчал бы как раньше.
|
||
assert re.search(rf"'{SOURCE}',\s*\n\s*true", sql), "расписание засеяно выключенным"
|
||
|
||
|
||
def test_migration_303_is_idempotent_and_transactional() -> None:
|
||
sql = _migration_sql()
|
||
assert "ON CONFLICT (source) DO NOTHING" in sql
|
||
assert "BEGIN;" in sql
|
||
assert "COMMIT;" in sql
|
||
|
||
|
||
def test_migration_303_no_psycopg_trap() -> None:
|
||
assert not re.search(r":\w+::", _migration_sql())
|
||
|
||
|
||
def test_migration_303_interval_days_is_daily() -> None:
|
||
"""Такт суточный: витрина устаревает от КОДА (деплой), а не от квартальных сделок."""
|
||
assert _seeded_interval_days() == 1
|
||
|
||
|
||
# ── 3. Сводка свежести краснеет на молчащей витрине ──────────────────────────
|
||
|
||
|
||
def _row(age_days: float, *, status: str | None = "done") -> Any:
|
||
"""Строка `_STALE_SOURCES_SQL`: прогон витрины `age_days` суток назад.
|
||
|
||
`status=None` (LEFT JOIN не нашёл прогонов) — витрину не пересчитывали ни разу
|
||
с момента появления расписания; тогда возраст считается от created_at строки.
|
||
"""
|
||
finished = None if status is None else NOW - timedelta(days=age_days)
|
||
return SimpleNamespace(
|
||
source=SOURCE,
|
||
interval_days=str(_seeded_interval_days()),
|
||
created_at=NOW - timedelta(days=age_days),
|
||
finished_at=finished,
|
||
status=status,
|
||
counters={"considered": 200, "eligible": 161, "written": 20},
|
||
)
|
||
|
||
|
||
def _stale_now(rows: list[Any]) -> list[sched.StaleSource]:
|
||
return sched.stale_sources(sched.freshness_rows(rows), NOW)
|
||
|
||
|
||
# Приёмка issue #3469 дословно: «отсутствие прогона дольше 3× такта даёт тревогу».
|
||
# ЧИСЛО ЗДЕСЬ ЛИТЕРАЛ, А НЕ `sched.STALE_DIGEST_INTERVAL_FACTOR`. Взятое из той же
|
||
# настройки, которую проверяем, ожидание уезжает вместе с ней: при факторе 3650
|
||
# ЭТИ ЖЕ тесты оставались зелёными (проверено руками), то есть проверяли ровно
|
||
# ничего. Такт (`interval_days`) при этом читается из миграции — правило «3×»
|
||
# и задано в тактах, а не в сутках.
|
||
_ACCEPTANCE_CYCLES = 3
|
||
|
||
|
||
def test_digest_flags_showcase_after_three_cycles() -> None:
|
||
"""Нет пересчёта дольше 3× такта → витрина в сводке просроченных."""
|
||
lag = _ACCEPTANCE_CYCLES * _seeded_interval_days() + 0.5
|
||
stale = _stale_now([_row(lag)])
|
||
assert [s.source for s in stale] == [SOURCE], (
|
||
f"витрина молчит {lag} суток при такте {_seeded_interval_days()} и не в тревоге"
|
||
)
|
||
assert stale[0].interval_days == _seeded_interval_days()
|
||
|
||
|
||
def test_digest_silent_within_cycle() -> None:
|
||
"""Контроль: два такта — ещё норма, иначе тревога кричала бы всегда."""
|
||
lag = 2 * _seeded_interval_days()
|
||
assert _stale_now([_row(lag)]) == []
|
||
|
||
|
||
def test_digest_flags_showcase_that_never_ran() -> None:
|
||
"""Расписание есть, прогонов нет — самый частый вид молчания (#3469 и был им)."""
|
||
lag = _ACCEPTANCE_CYCLES * _seeded_interval_days() + 1
|
||
stale = _stale_now([_row(lag, status=None)])
|
||
assert [s.source for s in stale] == [SOURCE]
|
||
assert stale[0].never_ok is True
|
||
|
||
|
||
def test_showcase_counters_do_not_fake_freshness() -> None:
|
||
"""Свежесть даёт ПРОГОН, а не его счётчики.
|
||
|
||
`run_brought_data` судит по результатным ключам kit'а, а у витрины их нет
|
||
(`considered`/`eligible`/`written` — свой словарь). Значит мерой остаётся
|
||
успешный статус: прогон, свалившийся в failed, свежести не даёт.
|
||
"""
|
||
assert sched.run_brought_data("done", {"considered": 200, "written": 20}) is True
|
||
assert sched.run_brought_data("failed", {"considered": 200, "written": 0}) is False
|
||
|
||
|
||
# ── 3а. Пустой прогон витрины не гасит часы свежести (#3511) ────────────────
|
||
|
||
|
||
class _RecordedRuns:
|
||
"""`ctx.runs`, который кладёт финал прогона строкой `_STALE_SOURCES_SQL`.
|
||
|
||
Тогда тест идёт ТЕМ ЖЕ путём, что и сводка на проде: обработчик ставит статус,
|
||
`freshness_rows` судит, принёс ли прогон данные, `stale_sources` — просрочку.
|
||
Проверка «вызван mark_failed» такой путь не проходит: она зелёная и тогда,
|
||
когда статус верный, а мера свежести его не видит.
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
self.rows: list[Any] = []
|
||
self.finished_at = NOW
|
||
|
||
def _row(self, status: str, counters: dict[str, Any]) -> None:
|
||
self.rows.append(
|
||
SimpleNamespace(
|
||
source=SOURCE,
|
||
interval_days=str(_seeded_interval_days()),
|
||
created_at=NOW - timedelta(days=30),
|
||
finished_at=self.finished_at,
|
||
status=status,
|
||
counters=counters,
|
||
)
|
||
)
|
||
|
||
def mark_done(self, db: Any, run_id: int, counters: dict[str, Any]) -> None:
|
||
self._row("done", counters)
|
||
|
||
def mark_failed(self, db: Any, run_id: int, error: str, counters: dict[str, Any]) -> None:
|
||
self._row("failed", counters)
|
||
|
||
|
||
async def _run_handler(runs: _RecordedRuns, counters: dict[str, int], age_days: float) -> None:
|
||
runs.finished_at = NOW - timedelta(days=age_days)
|
||
with patch(
|
||
"app.tasks.landing_showcase_deals.refresh_landing_showcase_deals",
|
||
return_value=counters,
|
||
):
|
||
await _job_landing_showcase_deals(MagicMock(), 1, {}, SimpleNamespace(runs=runs))
|
||
|
||
|
||
_FULL = {"considered": 200, "priced": 180, "eligible": 160, "written": 20}
|
||
_EMPTY = {"considered": 200, "priced": 180, "eligible": 160, "written": 0}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_empty_showcase_run_does_not_reset_freshness() -> None:
|
||
"""Последний непустой прогон старше 3× такта, вчерашний пустой → витрина в тревоге.
|
||
|
||
До #3511 пустой прогон завершался `done`, и `run_brought_data` (судит по
|
||
статусу: результатного ключа кита у витрины нет) засчитывал его свежестью —
|
||
опустевший блок лэндинга молчал бы сколько угодно.
|
||
"""
|
||
runs = _RecordedRuns()
|
||
await _run_handler(runs, _FULL, age_days=_ACCEPTANCE_CYCLES * _seeded_interval_days() + 1)
|
||
await _run_handler(runs, _EMPTY, age_days=1)
|
||
|
||
stale = _stale_now(runs.rows)
|
||
assert [s.source for s in stale] == [SOURCE], (
|
||
f"пустой прогон витрины засчитан свежестью: статусы {[r.status for r in runs.rows]}"
|
||
)
|
||
assert stale[0].age_days > _ACCEPTANCE_CYCLES * _seeded_interval_days()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_nonempty_showcase_run_keeps_freshness() -> None:
|
||
"""Контроль: тот же сценарий с непустым последним прогоном — тревоги нет.
|
||
|
||
Зелёный с обеих сторон правки; без него правка могла бы валить любой прогон.
|
||
"""
|
||
runs = _RecordedRuns()
|
||
await _run_handler(runs, _FULL, age_days=_ACCEPTANCE_CYCLES * _seeded_interval_days() + 1)
|
||
await _run_handler(runs, _FULL, age_days=1)
|
||
|
||
assert _stale_now(runs.rows) == []
|
||
assert [r.status for r in runs.rows] == ["done", "done"]
|
||
|
||
|
||
# ── 4. Живая БД: строка реально ложится в таблицу ────────────────────────────
|
||
|
||
|
||
def _live_session() -> Any | None:
|
||
"""Session на живой Postgres — та же проба, что у соседних живых тестов.
|
||
|
||
`TEST_DATABASE_URL` имеет приоритет; `localhost:5432/test` — заглушка модулей,
|
||
её не считаем базой. В CI сюда приезжает DSN поднятого в job'е контейнера
|
||
(ci-tradein.yml), поэтому проверка там ИДЁТ, а не тихо скипается.
|
||
"""
|
||
try:
|
||
from sqlalchemy import create_engine, text
|
||
from sqlalchemy.orm import sessionmaker
|
||
|
||
dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "")
|
||
if not dsn or "localhost:5432/test" in dsn:
|
||
return None
|
||
engine = create_engine(dsn, future=True)
|
||
with engine.connect() as conn:
|
||
conn.execute(text("SELECT 1"))
|
||
return sessionmaker(bind=engine, future=True)()
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _apply(db: Any, path: Path) -> None:
|
||
"""Прогнать файл миграции целиком, одним куском — как `psql -f` на деплое.
|
||
|
||
Через ДРАЙВЕРНОЕ соединение, а не `exec_driver_sql`: последний отдаёт текст
|
||
psycopg вместе с пустым набором параметров, и тот начинает искать в нём
|
||
плейсхолдеры — любой процент в комментарии миграции («полоса −5..+20 %»)
|
||
роняет запуск ошибкой про `%`. psql такого разбора не делает, так что это
|
||
артефакт теста, а не свойство файла.
|
||
"""
|
||
db.connection().connection.driver_connection.execute(path.read_text("utf-8"))
|
||
db.commit()
|
||
|
||
|
||
@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
|
||
def test_live_migration_puts_showcase_into_schedules_and_digest() -> None:
|
||
"""Миграция кладёт строку в scrape_schedules, и сводка видит витрину просроченной.
|
||
|
||
Значение, а не текст файла: применяем 303 на живой базе (дважды — дублей быть
|
||
не должно), читаем строку обратно и прогоняем настоящий `_STALE_SOURCES_SQL` —
|
||
тот же запрос, которым сводка судит на проде.
|
||
|
||
Ничего не удаляем: обе миграции идемпотентны (CREATE TABLE IF NOT EXISTS /
|
||
ON CONFLICT DO NOTHING), то есть повтор здесь — ровно то же действие, что и
|
||
повторный деплой.
|
||
"""
|
||
from sqlalchemy import text
|
||
|
||
db = _live_session()
|
||
assert db is not None
|
||
try:
|
||
# Зависимости — ТОЛЬКО на пустой базе. На базе, прошедшей всю цепочку
|
||
# (CI и прод), повтор 015 падает: `CREATE TABLE IF NOT EXISTS` — no-op,
|
||
# а `COMMENT ON COLUMN scrape_runs.returning_count` внизу того же файла
|
||
# обращается к колонке, которую снесла 214. Файл идемпотентен
|
||
# относительно себя, но не относительно схемы, прошедшей 214, — и
|
||
# прогон на чистой базе этого не видит по построению.
|
||
if db.execute(text("SELECT to_regclass('public.scrape_schedules')")).scalar() is None:
|
||
for dep in _DEPS:
|
||
_apply(db, dep)
|
||
_apply(db, _MIGRATION)
|
||
# ИДЕМПОТЕНТНОСТЬ МЕРЯЕТСЯ ПО СОСТОЯНИЮ СТРОКИ, А НЕ ПО ЧИСЛУ СТРОК.
|
||
# «DELETE + INSERT» тоже оставляет ровно одну строку, но на КАЖДОМ
|
||
# деплое стирает last_run_at/next_run_at и взводит расписание заново —
|
||
# счёт строк такую замену не отличает, а created_at отличает.
|
||
first_created_at = db.execute(
|
||
text("SELECT created_at FROM scrape_schedules WHERE source = :s"), {"s": SOURCE}
|
||
).scalar()
|
||
_apply(db, _MIGRATION)
|
||
|
||
rows = db.execute(
|
||
text(
|
||
"SELECT enabled, window_start_hour, window_end_hour, created_at, "
|
||
" (next_run_at > now()) AS next_run_ahead, "
|
||
" default_params->>'interval_days' AS interval_days "
|
||
"FROM scrape_schedules WHERE source = :s"
|
||
),
|
||
{"s": SOURCE},
|
||
).fetchall()
|
||
assert len(rows) == 1, f"ожидалась одна строка расписания, получено {len(rows)}"
|
||
row = rows[0]
|
||
assert row.created_at == first_created_at, (
|
||
"повторное применение пересоздало строку расписания — на каждом деплое "
|
||
"это стирало бы состояние прогонов (last_run_at/next_run_at)"
|
||
)
|
||
assert row.enabled is True
|
||
assert (row.window_start_hour, row.window_end_hour) == (6, 7)
|
||
assert int(row.interval_days) == _seeded_interval_days()
|
||
# next_run_at в БУДУЩЕМ: сев расписания не должен выстреливать прогоном
|
||
# в момент деплоя (образец — 162/275). Стоит в приёмке, значит и здесь.
|
||
assert row.next_run_ahead is True, "next_run_at в прошлом — прогон стартует на деплое"
|
||
|
||
# Сводка: прогонов у витрины нет, возраст считается от created_at строки.
|
||
digest_rows = list(db.execute(sched._STALE_SOURCES_SQL).fetchall())
|
||
assert SOURCE in {r.source for r in digest_rows}, "сводка не видит витрину"
|
||
|
||
overdue_at = row.created_at + timedelta(
|
||
days=_ACCEPTANCE_CYCLES * _seeded_interval_days() + 0.5
|
||
)
|
||
stale = sched.stale_sources(sched.freshness_rows(digest_rows), overdue_at)
|
||
assert SOURCE in {s.source for s in stale}, "витрина без прогонов не попала в тревогу"
|
||
|
||
fresh_at = row.created_at + timedelta(hours=1)
|
||
fresh = sched.stale_sources(sched.freshness_rows(digest_rows), fresh_at)
|
||
assert SOURCE not in {s.source for s in fresh}, "тревога сразу после сева — ложная"
|
||
finally:
|
||
db.close()
|