"""Витрина сделок лэндинга попадает в расписание и в монитор свежести (#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()