"""Инварианты миграций БД `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."