lauadminandClaude Opus 5.5
56ac0fc370
Cashiers can edit recipient, TTN, notes, goods and total in the order
modal and save via PATCH /orders/{id}. Edited orders get edited_at and
are no longer overwritten by CRM sync. Editing is blocked once a receipt
exists; changing the TTN resets Nova Poshta tracking fields.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
95 lines
4.4 KiB
Python
95 lines
4.4 KiB
Python
"""Миграции обязаны точно соответствовать моделям.
|
|
|
|
Расхождение между `Base.metadata` и миграциями обнаруживается только на живой
|
|
БД и обычно уже в проде. Проверка статическая: сравниваем таблицы, колонки,
|
|
nullability и индексы, которые создают миграции, с тем, что описано моделями.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.db.models import Base
|
|
|
|
VERSIONS_DIR = Path(__file__).resolve().parents[1] / "alembic" / "versions"
|
|
MIGRATION_FILES = sorted(VERSIONS_DIR.glob("*.py"))
|
|
SOURCES = {path.name: path.read_text(encoding="utf-8") for path in MIGRATION_FILES}
|
|
# Склеенный текст всех миграций — таблица может быть создана в любой из них.
|
|
COMBINED_SOURCE = "\n".join(SOURCES.values())
|
|
|
|
MODEL_TABLES = set(Base.metadata.tables)
|
|
|
|
|
|
def _migration_block(table: str) -> str:
|
|
"""Тело вызова op.create_table для указанной таблицы плюс её поздние op.add_column.
|
|
|
|
Границей create_table служит следующий op.create_table или конец файла миграции.
|
|
"""
|
|
marker = f'op.create_table(\n "{table}",'
|
|
source = next(text for text in SOURCES.values() if marker in text)
|
|
start = source.index(marker)
|
|
next_table = source.find("op.create_table(", start + 1)
|
|
end = next_table if next_table != -1 else len(source)
|
|
# Колонки, добавленные позже через op.add_column, — в любой миграции.
|
|
added = re.findall(rf'op\.add_column\(\s*"{table}",\s*(sa\.Column\(.*?\n)', COMBINED_SOURCE)
|
|
return source[start:end] + "\n" + "\n".join(added)
|
|
|
|
|
|
def test_migrations_create_every_model_table() -> None:
|
|
created = set(re.findall(r'op\.create_table\(\n\s+"(\w+)"', COMBINED_SOURCE))
|
|
assert created == MODEL_TABLES, (
|
|
f"Миграции и модели разошлись. Только в моделях: {MODEL_TABLES - created}; "
|
|
f"только в миграциях: {created - MODEL_TABLES}"
|
|
)
|
|
|
|
|
|
def test_each_migration_downgrade_drops_its_own_tables() -> None:
|
|
"""Каждая миграция откатывает ровно те таблицы, которые сама создаёт."""
|
|
for name, source in SOURCES.items():
|
|
created = set(re.findall(r'op\.create_table\(\n\s+"(\w+)"', source))
|
|
downgrade = source[source.index("def downgrade()") :]
|
|
dropped = set(re.findall(r'op\.drop_table\("(\w+)"\)', downgrade))
|
|
assert dropped == created, f"{name}: downgrade не удаляет {created - dropped}"
|
|
|
|
|
|
@pytest.mark.parametrize("table_name", sorted(MODEL_TABLES))
|
|
def test_columns_match(table_name: str) -> None:
|
|
block = _migration_block(table_name)
|
|
in_migration = set(re.findall(r'sa\.Column\(\n?\s*"(\w+)"', block))
|
|
in_model = {column.name for column in Base.metadata.tables[table_name].columns}
|
|
|
|
assert in_migration == in_model, (
|
|
f"Таблица {table_name}: только в модели {in_model - in_migration}, "
|
|
f"только в миграции {in_migration - in_model}"
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("table_name", sorted(MODEL_TABLES))
|
|
def test_nullability_matches(table_name: str) -> None:
|
|
block = _migration_block(table_name)
|
|
model_table = Base.metadata.tables[table_name]
|
|
|
|
for column in model_table.columns:
|
|
match = re.search(rf'sa\.Column\(\n?\s*"{column.name}".*?\n', block)
|
|
assert match, f"{table_name}.{column.name} отсутствует в миграции"
|
|
|
|
migration_nullable = "nullable=True" in match.group(0)
|
|
assert migration_nullable == column.nullable, (
|
|
f"{table_name}.{column.name}: в модели nullable={column.nullable}, "
|
|
f"в миграции nullable={migration_nullable}"
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("table_name", sorted(MODEL_TABLES))
|
|
def test_indexes_match(table_name: str) -> None:
|
|
model_indexes = {index.name for index in Base.metadata.tables[table_name].indexes}
|
|
migration_indexes = set(
|
|
re.findall(r'op\.create_index\(\s*(?:op\.f\()?"(\w+)"', COMBINED_SOURCE)
|
|
)
|
|
|
|
missing = model_indexes - migration_indexes
|
|
assert not missing, f"Таблица {table_name}: миграция не создаёт индексы {missing}"
|