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

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


# ── 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()