gendesign/tradein-mvp/backend/tests/test_migration_numbering.py

"""Инварианты нумерации миграций trade-in (issues #2216, #2683).

КОНТРАКТ. ОДНА ФОРМУЛИРОВКА, И ОНА ЗДЕСЬ — больше нигде её дублировать не надо.

  Эталон «что уже закреплено на проде» — origin/main, а не файл-список.
  deploy-tradein.yml прогоняет КАЖДЫЙ data/sql/*.sql из main под ON_ERROR_STOP
  (падение миграции => красный деплой) и трекает применённое по bare filename в
  public._schema_migrations. То есть «файл доехал до main» == «имя закреплено на
  проде», и вести это знание отдельно от git незачем: git и есть журнал.

  Отсюда ровно два инварианта, и ниже проверяются именно они.

  1. Имя, существовавшее в точке ветвления, нельзя переименовать или удалить.
     Прод помнит СТАРОЕ имя; новое считается неприменённым и прогоняется
     повторно — дубль-INSERT / повторный DDL / PK violation под ON_ERROR_STOP,
     то есть либо красный деплой, либо тихо задвоенные данные. Нужно изменить
     уже применённую миграцию — заводи НОВЫЙ файл, старый оставь как есть.

  2. НОВЫЙ файл обязан нести NN-префикс, свободный не только в рабочем дереве,
     но и в origin/main. Порядок применения — `ls | sort`, два файла с одним NN
     дают неопределённый порядок. Шесть исторических дублей (084/108/113/121/
     124/130) не «новые» и не флагаются.

  ДОПИСЫВАТЬ НИЧЕГО НЕ НАДО. Автор миграции кладёт файл со свободным номером — и
  всё. Списка, который можно забыть обновить, здесь больше нет: до #2683 таким
  списком был data/sql/_manifest_applied.txt, и он по построению не мог
  покраснеть — «файл, которого нет в списке» и «новый файл этого PR» были для
  теста одним и тем же, поэтому забытое имя навсегда оставалось зелёным
  (замер 2026-08-07: 15 забытых имён, сьют зелёный).

  ПОЧЕМУ ТОЧКА ВЕТВЛЕНИЯ, А НЕ САМ origin/main. Ветка, отведённая неделю назад,
  не содержит миграций, смерженных после неё. Правило «origin/main ⊆ рабочее
  дерево» красило бы каждую такую ветку без вины автора — и его отключили бы
  через неделю. Сверка с merge-base ловит ровно то, что удалила или
  переименовала ЭТА ветка, а номера при этом сверяются с ПОЛНЫМ origin/main,
  чтобы коллизия с миграцией, смерженной после ветвления, всё-таки нашлась.

  КАК УБЕДИТЬСЯ, ЧТО СТОРОЖ УМЕЕТ КРАСНЕТЬ (не на слово):
    test_collision_rule_flags_a_taken_number ниже проверяет само правило на
    литеральных входах, а сквозной прогон воспроизводится так —
      git worktree add --detach /tmp/wt <sha до нужной миграции>
      touch /tmp/wt/tradein-mvp/backend/data/sql/<занятый-NN>_probe.sql
      (cd /tmp/wt/tradein-mvp/backend && pytest tests/test_migration_numbering.py)
"""

from __future__ import annotations

import os
import re
import subprocess
from collections.abc import Iterable
from pathlib import Path, PurePosixPath

import pytest

_TESTS_DIR = Path(__file__).resolve().parent
_SQL_DIR = _TESTS_DIR.parent / "data" / "sql"
_REPO_ROOT = _TESTS_DIR.parents[2]
# Путь каталога ОТ КОРНЯ РЕПО — им адресуем дерево коммита через git ls-tree.
_SQL_PATHSPEC = "tradein-mvp/backend/data/sql"

_NN_PREFIX = re.compile(r"^(\d+)_")
# origin — штатный remote; forgejo/main оставлен как исторический алиас.
_MAIN_REFS = ("origin/main", "forgejo/main", "main")


def _git(*args: str) -> str | None:
    """stdout git-команды, либо None если git недоступен/команда упала."""
    try:
        done = subprocess.run(
            # Фиксированный argv, без shell — args приходят только из этого модуля.
            ["git", "-C", str(_REPO_ROOT), *args],
            capture_output=True,
            text=True,
            timeout=30,
            check=False,
        )
    except (OSError, subprocess.SubprocessError):
        return None
    return done.stdout if done.returncode == 0 else None


def _sql_names_at(rev: str) -> set[str]:
    """Bare-имена *.sql в data/sql на ревизии rev."""
    out = _git("ls-tree", "-r", "-z", "--name-only", rev, "--", _SQL_PATHSPEC) or ""
    return {PurePosixPath(p).name for p in out.split("\0") if p.endswith(".sql")}


def _sql_names_on_disk() -> set[str]:
    """Bare-имена *.sql в рабочем дереве — включая ещё не закоммиченные."""
    return {p.name for p in _SQL_DIR.glob("*.sql")}


def _prefix(name: str) -> str | None:
    m = _NN_PREFIX.match(name)
    return m.group(1) if m else None


def _collisions(new_names: Iterable[str], universe: Iterable[str]) -> list[str]:
    """Для каждого НОВОГО имени — чужие имена с тем же NN-префиксом."""
    by_prefix: dict[str, set[str]] = {}
    for name in universe:
        p = _prefix(name)
        if p is not None:
            by_prefix.setdefault(p, set()).add(name)

    found: list[str] = []
    for name in sorted(new_names):
        p = _prefix(name)
        if p is None:
            continue
        others = sorted(by_prefix.get(p, set()) - {name})
        if others:
            found.append(f"{name} — номер {p} уже занят: {', '.join(others)}")
    return found


def _baseline() -> tuple[set[str], set[str]]:
    """(имена в точке ветвления, имена в main). Без git-эталона проверять нечего."""
    main_ref = next(
        (r for r in _MAIN_REFS if _git("rev-parse", "--verify", "--quiet", f"{r}^{{commit}}")),
        None,
    )
    merge_base = None
    if main_ref is not None:
        out = _git("merge-base", main_ref, "HEAD")
        merge_base = out.strip() if out else None

    if main_ref is None or merge_base is None:
        why = (
            f"нет git-эталона миграций: ни один из {_MAIN_REFS} не резолвится либо у него "
            f"нет общего предка с HEAD (repo={_REPO_ROOT}). Лечится "
            "`git fetch --no-tags origin +refs/heads/main:refs/remotes/origin/main` "
            "и НЕ shallow-клоном (нужен общий предок)."
        )
        # В CI это не «нечего проверять», а сломанный гейт: пропуск здесь и есть
        # тот зелёный, который ничего не проверяет. Поэтому красим.
        if os.environ.get("CI") or os.environ.get("GITHUB_ACTIONS"):
            pytest.fail(why)
        pytest.skip(why)

    base_names = _sql_names_at(merge_base)
    main_names = _sql_names_at(main_ref)
    # Анти-вакуум: пустой эталон сделал бы обе проверки зелёными всегда.
    # Ровно так ломается сторож, если _SQL_PATHSPEC разъедется с раскладкой репо.
    assert base_names, (
        f"эталон пуст: git ls-tree {merge_base} -- {_SQL_PATHSPEC} не вернул ни одного "
        "*.sql. Проверка номеров была бы вакуумно-зелёной — почини путь."
    )
    assert main_names, f"в {main_ref} не найдено *.sql по пути {_SQL_PATHSPEC} — то же самое."
    return base_names, main_names


def test_applied_migration_is_not_renamed_or_deleted() -> None:
    """Файл, существовавший в точке ветвления, обязан существовать и сейчас.

    Red => эта ветка переименовала или удалила миграцию, которую прод уже
    применил и помнит по СТАРОМУ имени. Верни исходное имя; нужно поправить
    поведение — заводи новый файл с новым номером.
    """
    on_disk = _sql_names_on_disk()
    assert on_disk, f"не найдено *.sql в {_SQL_DIR}"
    base_names, _ = _baseline()

    gone = sorted(base_names - on_disk)
    assert not gone, (
        f"миграции пропали из {_SQL_PATHSPEC}/ (переименованы или удалены): {gone}. "
        "Прод трекает их по bare-filename в _schema_migrations — под новым именем "
        "миграция прогонится повторно. Верни имена как были."
    )


def test_new_migration_takes_a_free_number() -> None:
    """Новый файл не переиспользует NN, занятый в origin/main или в этой ветке.

    Red => номер уже занят. Возьми следующий свободный, сверяясь с origin/main:

        git fetch origin main
        git ls-tree -r --name-only origin/main -- tradein-mvp/backend/data/sql | tail

    `-r` обязателен: без него ls-tree печатает сам каталог, а не файлы (в этом
    виде рецепт и ходил по issue #2683 — и молча возвращал одну строку).
    Локального `ls` недостаточно: он не видит миграций, смерженных после
    ветвления — ровно так разъехались 212 в #2682 и 234 в #2754.
    """
    on_disk = _sql_names_on_disk()
    base_names, main_names = _baseline()

    new_names = on_disk - base_names
    problems = _collisions(new_names, main_names | on_disk)
    assert not problems, "коллизия номеров миграций: " + "; ".join(problems)


def test_collision_rule_flags_a_taken_number() -> None:
    """Проверка самого правила — сторож обязан уметь краснеть (#2683 п.6).

    Ожидания здесь — ЛИТЕРАЛЫ, а не производные от содержимого data/sql: тест,
    который берёт ожидание из охраняемой настройки, зелен при любой настройке.
    """
    # Реальный случай #2754: ветка отвелась до того, как в main приехал 234_scrape.
    assert _collisions(
        ["234_trade_in_estimates_retain_until.sql"],
        {
            "233_payments.sql",
            "234_scrape_runs_ban_kind_unknown.sql",
            "234_trade_in_estimates_retain_until.sql",
        },
    ) == [
        "234_trade_in_estimates_retain_until.sql — номер 234 уже занят: "
        "234_scrape_runs_ban_kind_unknown.sql"
    ]
    # Два новых файла с одним номером внутри одной ветки — оба названы.
    assert len(_collisions(["300_a.sql", "300_b.sql"], {"300_a.sql", "300_b.sql"})) == 2
    # Свободный номер — тишина; исторические дубли не новые и не флагаются.
    assert _collisions(["300_a.sql"], {"084_x.sql", "084_y.sql", "300_a.sql"}) == []