"""У rosreestr_deals есть партиция под каждый квартал, который Росреестр уже мог
опубликовать (#2998).

Таблица партиционирована по period_start_date, и партиции создавались списком в
01_schema_rosreestr_deals.sql — «2024 Q3 — 2026 Q1». Дальше этого горизонта таблица
ничего не знала, и никакой механизм новые партиции не создаёт. Q2 2026 вышел 10.07,
poll заметил его 14.08, а загрузка 21.08 упала:

    ERROR: no partition of relation "rosreestr_deals" found for row
    DETAIL: Partition key of the failing row contains (period_start_date) = (2026-04-01).

То есть даже оператор, запустив 02_load_all_quarters.sh по подсказке poll, получил
бы отказ. Миграция 193 добавляет Q2–Q4 2026; этот тест следит, чтобы горизонт не
отставал снова: партиция обязана существовать на ПОСЛЕДНИЙ квартал, который по
календарю уже мог быть опубликован (публикация отстаёт от конца квартала ~на 10
дней: Q2 2026 вышел 10.07), плюс на следующий — чтобы предупреждение приходило за
квартал до отказа, а не в день публикации.

Герметично, как соседние DB-тесты: в своей СХЕМЕ-песочнице поднимается
rosreestr_deals из 01_schema (без индексов), применяется миграция 193, горизонт
читается из pg_inherits по этой схеме. Прод не трогается; в CI идёт на
postgres-сервисе. Без БД — skip с причиной.

Красная сторона воспроизводима, а не предполагается: до миграции 193 горизонт
01-схемы кончается на 2026q1, и головная проверка краснеет по значению
(«нет партиции на квартал 2026-04-01»).
"""

from __future__ import annotations

import os

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

import re
import uuid
from datetime import date, timedelta
from pathlib import Path

import pytest
from sqlalchemy import create_engine, text

_SQL_DIR = Path(__file__).resolve().parents[3] / "data" / "sql"
_SCHEMA_01 = _SQL_DIR / "01_schema_rosreestr_deals.sql"
_MIGRATION = _SQL_DIR / "193_partitions_rosreestr_2026_q2_q4.sql"

# Публикация квартала отстаёт от его конца; берём запас, чтобы не требовать партицию
# раньше, чем данные вообще могут появиться. Q2 2026 (конец 30.06) опубликован 10.07.
_PUBLICATION_LAG_DAYS = 20


def _dsn() -> str:
    raw = os.environ.get("TEST_DATABASE_URL") or os.environ.get(
        "DATABASE_URL", "postgresql+psycopg://gendesign@localhost:15432/gendesign"
    )
    return raw.replace("postgresql://", "postgresql+psycopg://", 1)


def _quarter_start(d: date) -> date:
    return date(d.year, 3 * ((d.month - 1) // 3) + 1, 1)


def _next_quarter(q: date) -> date:
    return date(q.year + (q.month == 10), 1 if q.month == 10 else q.month + 3, 1)


def _latest_publishable_quarter(today: date) -> date:
    """Начало последнего квартала, чей дамп по календарю уже мог выйти."""
    q = _quarter_start(today)
    prev_q = _quarter_start(q - timedelta(days=1))
    if (today - q).days >= _PUBLICATION_LAG_DAYS:
        return prev_q
    return _quarter_start(prev_q - timedelta(days=1))


def _schema_01_without_indexes() -> str:
    """01_schema без CREATE INDEX: индексы тесту не нужны, а часть их — по колонкам,
    которых в песочнице проверять нечего. Партиции и сама таблица — целиком."""
    sql = _SCHEMA_01.read_text(encoding="utf-8")
    return re.sub(r"CREATE INDEX[^;]*;", "", sql, flags=re.S)


def _partition_starts(conn, schema: str) -> set[date]:
    rows = conn.execute(
        text(
            """
            SELECT pg_get_expr(c.relpartbound, c.oid) AS bound
              FROM pg_inherits i
              JOIN pg_class c ON c.oid = i.inhrelid
              JOIN pg_class p ON p.oid = i.inhparent
              JOIN pg_namespace n ON n.oid = p.relnamespace
             WHERE p.relname = 'rosreestr_deals' AND n.nspname = :schema
            """
        ),
        {"schema": schema},
    ).scalars()
    out: set[date] = set()
    for b in rows:
        frm = b.split("FROM ('", 1)[1].split("'", 1)[0]
        out.add(date.fromisoformat(frm))
    return out


@pytest.fixture(scope="module")
def sandbox():
    """Схема-песочница с rosreestr_deals из 01_schema. Удаляется в конце модуля."""
    try:
        eng = create_engine(_dsn(), future=True)
        conn = eng.connect()
        conn.execute(text("SELECT 1"))
        conn.rollback()  # SELECT 1 авто-открыл транзакцию — закрыть, иначе begin() падает
    except Exception as e:  # pragma: no cover - среда без БД
        pytest.skip(f"нет Postgres для проверки партиций: {e}")
    schema = f"t2998_{uuid.uuid4().hex[:10]}"
    with conn.begin():
        conn.execute(text(f"CREATE SCHEMA {schema}"))
        conn.execute(text(f"SET LOCAL search_path TO {schema}"))
        conn.execute(text(_schema_01_without_indexes()))
    try:
        yield conn, schema
    finally:
        conn.rollback()
        with conn.begin():
            conn.execute(text(f"DROP SCHEMA {schema} CASCADE"))
        conn.close()


def _apply_193(conn, schema: str) -> None:
    conn.rollback()  # предыдущие SELECT авто-открыли транзакцию
    with conn.begin():
        conn.execute(text(f"SET LOCAL search_path TO {schema}"))
        conn.execute(text(_MIGRATION.read_text(encoding="utf-8")))


def test_schema_01_alone_is_red_for_the_publishable_quarter(sandbox) -> None:
    """Красная сторона: ДО миграции 193 партиции на публикуемый квартал нет.

    Это ровно состояние origin/main на 21.08.2026, когда загрузка Q2 упала.
    Проверка по значению: горизонт 01-схемы кончается на 2026q1.
    """
    conn, schema = sandbox
    have = _partition_starts(conn, schema)
    assert have, "песочница пуста — 01_schema не применилась"
    assert max(have) == date(2026, 1, 1), (
        f"горизонт 01-схемы ожидался 2026q1, есть {sorted(have)[-2:]}"
    )
    need = date(2026, 4, 1)  # публикуемый квартал на дату инцидента
    assert need not in have, "красная сторона не состоялась: 01-схема уже знает Q2 2026"


def test_migration_193_is_idempotent_and_closes_the_gap(sandbox) -> None:
    """Миграция применяется дважды без ошибки и добавляет Q2–Q4 2026."""
    conn, schema = sandbox
    _apply_193(conn, schema)
    _apply_193(conn, schema)
    have = _partition_starts(conn, schema)
    assert {date(2026, 4, 1), date(2026, 7, 1), date(2026, 10, 1)} <= have, sorted(have)[-4:]


def test_partition_exists_for_every_publishable_quarter(sandbox) -> None:
    """Головной сторож горизонта: на каждый уже-публикуемый квартал есть партиция.

    Сегодня это Q2 2026 (после 193 — есть). Когда календарь дойдёт до квартала,
    которого миграции ещё не создали, тест покраснеет по значению — и это его работа.
    """
    conn, schema = sandbox
    _apply_193(conn, schema)
    have = _partition_starts(conn, schema)
    need = _latest_publishable_quarter(date.today())
    assert need in have, (
        f"нет партиции на квартал {need} — загрузка опубликованного дампа упадёт с "
        f"«no partition of relation rosreestr_deals found for row»; есть: {sorted(have)[-3:]}"
    )


def test_partition_exists_one_quarter_ahead(sandbox) -> None:
    """Контроль горизонта: партиция на СЛЕДУЮЩИЙ квартал тоже есть — предупреждение
    должно приходить за квартал до отказа, а не в день публикации."""
    conn, schema = sandbox
    _apply_193(conn, schema)
    have = _partition_starts(conn, schema)
    need = _next_quarter(_latest_publishable_quarter(date.today()))
    assert need in have, (
        f"нет партиции на следующий квартал {need} — запаса нет, следующая публикация "
        f"снова упрётся в отсутствие партиции; есть: {sorted(have)[-3:]}"
    )


def test_calendar_helper_matches_known_publication() -> None:
    """Контроль калибровки (без БД): 21.08.2026 → Q2 2026; 05.07 → ещё Q1; 25.07 → Q2."""
    assert _latest_publishable_quarter(date(2026, 8, 21)) == date(2026, 4, 1)
    assert _latest_publishable_quarter(date(2026, 7, 5)) == date(2026, 1, 1)
    assert _latest_publishable_quarter(date(2026, 7, 25)) == date(2026, 4, 1)