gendesign/backend/tests/sql/test_auth_sql_migrations.py

"""Инварианты миграций БД `auth` (data/sql/auth/*.sql) + её bootstrap (ops/db-bootstrap/*.sql).

Снимка «уже применённого» здесь нет и не нужно: у соседнего стека такой файл-список был
(tradein data/sql/_manifest_applied.txt) и его удалили в #2683 — он отставал от каталога
и по построению не мог покраснеть. Аналог гейта для tradein теперь берёт эталон из git:
tradein-mvp/backend/tests/test_migration_numbering.py. Здесь же проверяются инварианты,
выполнимые БЕЗ всякого эталона: они верны на новом каталоге с первого дня и ловят
регрессии, которые иначе всплывают только на проде во время деплоя.

Тест не требует БД — только чтение файлов.
"""

from __future__ import annotations

import re
from pathlib import Path

_REPO_ROOT = Path(__file__).resolve().parents[3]
_AUTH_SQL_DIR = _REPO_ROOT / "data" / "sql" / "auth"
_BOOTSTRAP_SQL_DIR = _REPO_ROOT / "ops" / "db-bootstrap"
_DEPLOY_WORKFLOW = _REPO_ROOT / ".forgejo" / "workflows" / "deploy.yml"

_FILENAME_RE = re.compile(r"^(\d{3})_[a-z0-9_]+\.sql$")

# Признаки утёкшего пароля в git. bcrypt-хеши ($2a$/$2b$/$2y$) запрещены наравне с
# plaintext: хеш из репозитория брутфорсится офлайн и переживает ротацию пароля,
# оставаясь в истории коммитов. Конвенция репо — сид вставляет password_hash = NULL,
# значения проставляются на проде (прецедент: tradein м.193).
_SECRET_PATTERNS = (
    re.compile(r"\$2[aby]\$\d{2}\$"),  # bcrypt hash
    re.compile(r"PASSWORD\s+'", re.IGNORECASE),  # CREATE/ALTER ROLE ... PASSWORD 'literal'
)


def _auth_sql_files() -> list[Path]:
    """Файлы, к которым применимы конвенции миграций (имя NNN_*, обёртка BEGIN/COMMIT)."""
    return sorted(_AUTH_SQL_DIR.glob("*.sql"))


def _secret_scanned_files() -> list[Path]:
    """Файлы, по которым гоняется поиск паролей/хешей — ШИРЕ, чем список миграций.

    ⚠️  НЕ «унифицируй» этот список с _auth_sql_files(): разделение намеренное.

      * data/sql/auth/*.sql — миграции: обязаны иметь имя NNN_snake_case.sql и обёртку
        BEGIN;/COMMIT; (см. test_filenames_and_unique_prefix, test_migrations_are_transactional).
      * ops/db-bootstrap/*.sql — bootstrap: НЕ миграции, поэтому намеренно без NNN-префикса
        (порядок задан явными шагами deploy.yml, не сортировкой) и намеренно без транзакции
        (CREATE DATABASE запрещён внутри транзакционного блока). Прогонять по ним проверки
        имён/BEGIN-COMMIT — значит сломать тест на корректных файлах.

    А вот запрет на пароли применим к обоим каталогам, и именно bootstrap здесь важнее:
    единственное место в репозитории с конструкцией `ALTER ROLE ... PASSWORD` — это
    ops/db-bootstrap/set_*_password.sql, то есть ровно тот файл, куда проще всего однажды
    «временно» вписать литерал вместо чтения из env. Другого контроля на это нет:
    в .pre-commit-config.yaml из секрет-сканеров только detect-private-key (bcrypt не ловит),
    а репо-wide grep невозможен — caddy/users.caddy.snippet легально содержит bcrypt-хеши
    действующих логинов.
    """
    return _auth_sql_files() + sorted(_BOOTSTRAP_SQL_DIR.glob("*.sql"))


def test_scanned_dirs_are_not_empty() -> None:
    """Sanity: пути до каталогов не разъехались (иначе все проверки ниже — пустые).

    Red => каталог переименован/перенесён, а тест этого не заметил бы: `glob` по
    несуществующему пути возвращает [], и все циклы ниже стали бы no-op'ами, оставаясь
    зелёными. Особенно опасно для проверки паролей — «зелено, потому что ничего не проверено».
    """
    assert _auth_sql_files(), f"Не найдено *.sql в {_AUTH_SQL_DIR}"
    assert sorted(_BOOTSTRAP_SQL_DIR.glob("*.sql")), f"Не найдено *.sql в {_BOOTSTRAP_SQL_DIR}"


def test_filenames_and_unique_prefix() -> None:
    """Имя вида NNN_snake_case.sql, префикс NNN уникален.

    Red => прод применяет файлы в порядке `ls | sort`; два файла с одним NNN дают
    неоднозначный порядок (например, роль/гранты раньше таблиц). Присвой следующий
    свободный номер.
    """
    seen: dict[str, str] = {}
    bad_names: list[str] = []
    collisions: list[str] = []
    for path in _auth_sql_files():
        m = _FILENAME_RE.match(path.name)
        if m is None:
            bad_names.append(path.name)
            continue
        prefix = m.group(1)
        if prefix in seen:
            collisions.append(f"{path.name} (префикс {prefix} уже у {seen[prefix]})")
        else:
            seen[prefix] = path.name

    assert not bad_names, (
        f"Имена не соответствуют NNN_snake_case.sql: {bad_names}. "
        "Порядок применения на проде определяется сортировкой имён."
    )
    assert not collisions, "Дублирующийся NNN-префикс: " + "; ".join(collisions)


def test_migrations_are_transactional() -> None:
    """Каждая миграция обёрнута в BEGIN; ... COMMIT; (.claude/rules/sql.md).

    Red => частично применённая миграция оставит БД auth в промежуточном состоянии:
    деплой падает на ON_ERROR_STOP, а уже выполненный DDL не откатывается.
    """
    broken: list[str] = []
    for path in _auth_sql_files():
        text = path.read_text(encoding="utf-8")
        statements = [
            line.strip()
            for line in text.splitlines()
            if line.strip() and not line.strip().startswith("--")
        ]
        if not statements or statements[0] != "BEGIN;" or statements[-1] != "COMMIT;":
            broken.append(path.name)
    assert (
        not broken
    ), f"Миграции без обёртки BEGIN;/COMMIT;: {broken} (.claude/rules/sql.md → Structure)."


def test_no_concurrent_index_in_migrations() -> None:
    """Ни одной CREATE/DROP INDEX CONCURRENTLY в data/sql/auth/*.sql.

    Red => миграция гарантированно падает на проде: CONCURRENTLY нельзя выполнять внутри
    транзакционного блока (Postgres: 25001 «CREATE INDEX CONCURRENTLY cannot run inside a
    transaction block»), а обёртка BEGIN;/COMMIT; здесь обязательна для всех файлов
    (test_migrations_are_transactional). Две проверки по отдельности зелёные, а вместе
    невыполнимые — поэтому запрет нужен явный: комбинация ловится только здесь.
    Нужен CONCURRENTLY на большой таблице — это отдельный ручной прогон вне auto-apply,
    а не файл в этом каталоге.
    """
    hits: list[str] = []
    for path in _auth_sql_files():
        text = path.read_text(encoding="utf-8")
        for line_no, line in enumerate(text.splitlines(), start=1):
            if line.lstrip().startswith("--"):
                continue  # комментарий может объяснять запрет, не нарушая его
            if re.search(r"\bCONCURRENTLY\b", line, re.IGNORECASE):
                hits.append(f"{path.name}:{line_no}: {line.strip()}")
    assert not hits, "CONCURRENTLY внутри BEGIN/COMMIT — упадёт на деплое: " + "; ".join(hits)


def test_no_password_material_in_auth_sql() -> None:
    """Ни в data/sql/auth, ни в ops/db-bootstrap нет plaintext-паролей и bcrypt-хешей.

    Покрытие шире каталога миграций сознательно — обоснование в _secret_scanned_files().

    Red => пароль/хеш попал в git. Убери значение: сид вставляет password_hash = NULL,
    пароль роли ставится из env через ops/db-bootstrap/set_auth_app_password.sql
    (значение приезжает из .env.runtime на VPS и в репозитории не существует).
    """
    hits: list[str] = []
    for path in _secret_scanned_files():
        rel = path.relative_to(_REPO_ROOT).as_posix()
        text = path.read_text(encoding="utf-8")
        for line_no, line in enumerate(text.splitlines(), start=1):
            if line.lstrip().startswith("--"):
                continue  # комментарии описывают запрет, а не нарушают его
            for pattern in _SECRET_PATTERNS:
                if pattern.search(line):
                    hits.append(f"{rel}:{line_no}: {line.strip()}")
    assert not hits, "Похоже на пароль/хеш в SQL: " + "; ".join(hits)


def test_deploy_workflow_applies_auth_migrations() -> None:
    """deploy.yml реально прогоняет data/sql/auth/*.sql.

    Каталог обособлен намеренно: основной цикл миграций использует `ls -1 data/sql/*.sql`
    и в подкаталоги НЕ рекурсирует (чтобы файлы auth физически не могли примениться в БД
    gendesign). Обратная сторона — без отдельного цикла в deploy.yml эти файлы не
    применяются вообще и никто этого не заметит. Red => wiring удалён или переименован.
    """
    workflow = _DEPLOY_WORKFLOW.read_text(encoding="utf-8")
    assert "data/sql/auth/*.sql" in workflow, (
        f"В {_DEPLOY_WORKFLOW.name} нет цикла по data/sql/auth/*.sql — миграции БД auth "
        "не применяются на деплое."
    )
    assert (
        "ops/db-bootstrap/create_auth_db.sql" in workflow
    ), f"В {_DEPLOY_WORKFLOW.name} нет bootstrap-шага создания БД auth."