feat: scripts/check_dbml_vs_orm.py — CLI сверки ORM с DBML (версия, таблицы, колонки, типы, FK, индексы, enum'ы)

Ядро сверки перенесено из tests/dbml_parity.py в scripts/dbml_parity.py (единственная
реализация, R-14); tests/dbml_parity.py и dbml_helpers.py — тонкие обёртки с пропуском
теста без docs. Единый формат строк: `<объект>: ожидалось X / фактически Y`,
`<объект>: отсутствует в ORM|DBML`; тип и nullability — одной строкой в нотации DBML.
Версия из шапки DBML сверяется с DBML_VERSION до сравнения таблиц. Коды: 0/1/2.
Без рабочего .env скрипт берёт api/.env.example. Makefile: dbml-check (в check-full),
mypy проверяет scripts.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev 2026-10-03 19:56:58 +04:00
parent 218ce88a31
commit e064830702
8 changed files with 735 additions and 289 deletions

View File

@ -2,8 +2,10 @@
VENV := api/.venv
BIN := $(CURDIR)/$(VENV)/bin
COMPOSE_FILE ?= ../docs/dev/docker-compose.yml
# DBML относительно api/: и в основном workspace, и в worktree .wt/<срез>/ это соседний docs/
DBML ?= ../../docs/model/data-model-v5.13.dbml
.PHONY: venv lint test test-integration check check-full up migrate seed seed-dev
.PHONY: venv lint test test-integration dbml-check check check-full up migrate seed seed-dev
# Python 3.11 (версия платформы); внутренние пакеты raisa_* вне банка недоступны — пропускаем их
venv:
@ -14,7 +16,7 @@ venv:
lint:
cd api && $(BIN)/ruff check .
cd api && $(BIN)/ruff format --check .
cd api && $(BIN)/mypy src
cd api && $(BIN)/mypy src scripts
test:
cd api && $(BIN)/pytest -m "not integration"
@ -22,10 +24,14 @@ test:
test-integration:
cd api && $(BIN)/pytest -m integration
# Сверка ORM с DBML (spec data-model): пустой вывод и код 0 — совпадение
dbml-check:
cd api && $(BIN)/python scripts/check_dbml_vs_orm.py $(DBML)
check: lint test
# TODO (задача 25): добавить фронт (lint/test/build web/), проверку миграций с нуля и откат, актуальность OpenAPI
check-full: check test-integration
check-full: check dbml-check test-integration
up:
docker compose -f $(COMPOSE_FILE) up -d

4
api/scripts/__init__.py Normal file
View File

@ -0,0 +1,4 @@
"""Служебные скрипты arm-core (запуск из `api/`).
Сверка ORM с DBML — `check_dbml_vs_orm.py` (ядро — `dbml_parity.py`); seed — `python -m scripts.seed`.
"""

View File

@ -0,0 +1,96 @@
"""Сверка ORM-моделей arm-core с DBML-моделью данных (spec data-model, design D-3).
Запуск из `api/`:
python scripts/check_dbml_vs_orm.py ../../docs/model/data-model-v5.13.dbml
Без аргумента DBML ищется в `$ARM_KA_DOCS/model/` или в соседнем `docs/` рядом с `arm-core`.
Вывод: пусто и код 0 — ORM совпадает с DBML; иначе по строке на расхождение и код 1, например:
osv_entry.extra_note: отсутствует в DBML
osv_entry.opening_debit: ожидалось numeric(18,2) NULL / фактически numeric(18,4) NOT NULL
Сначала сверяется версия из шапки DBML с `src.models.DBML_VERSION`: при расхождении — одна строка о версии, код 1,
таблицы не сравниваются. Код 2 — ошибка использования: нет файла, нет версии в шапке, DBML не разбирается
(сообщение — в stderr).
Окружение. Импорт `src.models` тянет `src/__init__` → приложение → `Settings`, которому нужны ключи `OPENBAO__*`.
Рабочий `.env` скрипту не нужен: если `ENV_FILE` не задан и в текущем каталоге нет `.env`, берётся `api/.env.example`
(все обязательные ключи там есть; соединений при импорте не открывается).
"""
import enum
import os
import sys
from collections.abc import Mapping, Sequence
from pathlib import Path
_API_DIR = Path(__file__).resolve().parents[1]
# Запуск файлом (`python scripts/check_dbml_vs_orm.py`): в sys.path попадает только `api/scripts`, а нужен `api/`.
if str(_API_DIR) not in sys.path:
sys.path.insert(0, str(_API_DIR))
if "ENV_FILE" not in os.environ and not Path(".env").is_file():
os.environ["ENV_FILE"] = str(_API_DIR / ".env.example")
from sqlalchemy import MetaData # noqa: E402
from scripts.dbml_parity import default_dbml_path, find_all_problems, parse_dbml, read_dbml_version # noqa: E402
from src.models import DBML_VERSION, Base # noqa: E402
from src.models.enums import ENUMS # noqa: E402
EXIT_OK, EXIT_PROBLEMS, EXIT_USAGE = 0, 1, 2
USAGE = "использование: python scripts/check_dbml_vs_orm.py [путь к .dbml]"
def run_check(
dbml_path: Path,
*,
metadata: MetaData | None = None,
enums: Mapping[str, type[enum.Enum]] | None = None,
orm_version: str = DBML_VERSION,
) -> tuple[int, list[str]]:
"""Сверить ORM с DBML-файлом; вернуть (код возврата, строки отчёта).
По умолчанию сверяются реальные `Base.metadata` и `ENUMS`; тесты подставляют испорченные копии.
При коде 2 единственная строка — сообщение об ошибке использования.
"""
if not dbml_path.is_file():
return EXIT_USAGE, [f"файл DBML не найден: {dbml_path}"]
# версия — до разбора и сравнения таблиц: новая модель без осознанного обновления ORM не сверяется вовсе
dbml_version = read_dbml_version(dbml_path)
if dbml_version is None:
return EXIT_USAGE, [
f"в первой строке {dbml_path} не найдена версия DBML (ожидалось `... v<мажор>.<минор> ...`)"
]
if dbml_version != orm_version:
return EXIT_PROBLEMS, [f"версия DBML {dbml_version} не совпадает с версией ORM {orm_version}: обновите модели"]
try:
dbml = parse_dbml(dbml_path)
except Exception as error: # любая ошибка pydbml = непригодный файл
return EXIT_USAGE, [f"не удалось разобрать DBML {dbml_path}: {error}"]
problems = find_all_problems(
Base.metadata if metadata is None else metadata, ENUMS if enums is None else enums, dbml
)
return (EXIT_PROBLEMS if problems else EXIT_OK), problems
def main(argv: Sequence[str] | None = None) -> int:
"""CLI: расхождения — в stdout, ошибки использования — в stderr."""
args = list(sys.argv[1:] if argv is None else argv)
if args[:1] in (["-h"], ["--help"]):
print(__doc__)
return EXIT_OK
if len(args) > 1:
print(USAGE, file=sys.stderr)
return EXIT_USAGE
code, lines = run_check(Path(args[0]) if args else default_dbml_path())
stream = sys.stderr if code == EXIT_USAGE else sys.stdout
for line in lines:
print(line, file=stream)
return code
if __name__ == "__main__":
sys.exit(main())

394
api/scripts/dbml_parity.py Normal file
View File

@ -0,0 +1,394 @@
"""Сверка ORM-метаданных с DBML (design D-3) — единственная реализация: её вызывают CLI `check_dbml_vs_orm.py` и тесты.
Модуль не импортирует `src` (на вход приходят `MetaData` и реестр enum'ов) и не зависит от pytest.
Что сравнивается:
- множество таблиц (`таблица: отсутствует в ORM|DBML`) и схема таблиц ORM (`dkb`);
- колонки: состав (`таблица.колонка: отсутствует в ORM|DBML`), порядок общих колонок, тип + nullability одной строкой
в нотации DBML (`ожидалось numeric(18,2) NULL / фактически numeric(18,4) NOT NULL`), PK, unique, default;
- FK по колонке-ссылке (`таблица.колонка: FK: ожидалось родитель.колонка / фактически …`);
- индексы по набору колонок: уникальность, partial-условие (note `partial UNIQUE WHERE <expr>`), `NULLS NOT DISTINCT`;
составной `[pk]` в Indexes DBML — это первичный ключ, а не индекс;
- enum'ы: множество имён и упорядоченный список значений.
Формат каждой строки — `<объект>: ожидалось X / фактически Y` или `<объект>: отсутствует в ORM|DBML` (функции
`mismatch` / `missing`), чтобы тесты и человек читали один и тот же текст.
Нетривиальные места. DBML-default — python-литерал pydbml (bool/int/float/str) или `Expression` (обратные кавычки,
напр. `now()`); ORM server_default — `text(...)` или SQL-функция (`func.now()`). Обе стороны приводятся к строке:
bool -> `true`/`false`, число -> как есть, строка -> в одинарных кавычках, выражение -> текст в нижнем регистре.
"""
import enum
import os
import re
from collections.abc import Callable, Collection, Mapping
from pathlib import Path
from typing import Any
from pydbml import PyDBML
from pydbml.classes import Column as DbmlColumn
from pydbml.classes import Enum as DbmlEnum
from pydbml.classes import Expression
from pydbml.classes import Table as DbmlTable
from sqlalchemy import (
JSON,
BigInteger,
Boolean,
Column,
Date,
DateTime,
Integer,
MetaData,
Numeric,
SmallInteger,
String,
Table,
Text,
)
from sqlalchemy import Enum as SAEnum
from sqlalchemy.dialects.postgresql import JSONB, UUID
from sqlalchemy.sql.elements import TextClause
from sqlalchemy.sql.functions import FunctionElement
SCHEMA = "dkb"
DBML_FILE_NAME = "data-model-v5.13.dbml"
ORM = "ORM"
DBML = "DBML"
ABSENT = "нет"
_NUMERIC_RE = re.compile(r"numeric\((\d+),\s*(\d+)\)")
_VERSION_RE = re.compile(r"v(\d+\.\d+)")
_PARTIAL_RE = re.compile(r"partial\s+UNIQUE\s+WHERE\s+(.+?)(?:\s+—.*)?$", re.IGNORECASE | re.DOTALL)
_NULLS_NOT_DISTINCT_RE = re.compile(r"NULLS\s+NOT\s+DISTINCT", re.IGNORECASE)
_SYNONYMS = {"current_timestamp": "now()", "current_timestamp()": "now()"}
#: Типы DBML, которые сверка умеет сопоставлять (плюс `numeric(p,s)` и enum'ы); прочие — явное расхождение.
SUPPORTED_DBML_TYPES = frozenset(
{"uuid", "varchar", "text", "smallint", "int", "bigint", "boolean", "date", "timestamptz", "jsonb"}
)
# --- формат строк отчёта ---------------------------------------------------------------------------------------
def mismatch(obj: str, expected: object, actual: object) -> str:
"""Строка расхождения значения: `<объект>: ожидалось X / фактически Y`."""
return f"{obj}: ожидалось {expected} / фактически {actual}"
def missing(obj: str, where: str) -> str:
"""Строка отсутствия объекта: `<объект>: отсутствует в ORM|DBML`."""
return f"{obj}: отсутствует в {where}"
# --- поиск и чтение DBML ---------------------------------------------------------------------------------------
def default_dbml_path() -> Path:
"""Путь к DBML по умолчанию: каталог docs из `ARM_KA_DOCS`, иначе соседний `docs` рядом с `arm-core`."""
docs = os.environ.get("ARM_KA_DOCS")
docs_dir = Path(docs) if docs else Path(__file__).resolve().parents[3] / "docs"
return docs_dir / "model" / DBML_FILE_NAME
def read_dbml_version(path: Path) -> str | None:
"""Версия из первой строки шапки DBML (`// ДКБ · Модель данных v5.13: ...`) или `None`, если её там нет."""
with path.open(encoding="utf-8") as file:
first_line = file.readline()
found = _VERSION_RE.search(first_line)
return found.group(1) if found else None
def parse_dbml(path: Path) -> PyDBML:
"""Разобрать DBML-файл через pydbml."""
return PyDBML.parse_file(path)
# --- типы ------------------------------------------------------------------------------------------------------
#: Типы без параметров: точный класс SQLAlchemy -> имя в DBML.
_SIMPLE_TYPE_NAMES: dict[type, str] = {
Text: "text",
SmallInteger: "smallint",
Integer: "int",
BigInteger: "bigint",
Boolean: "boolean",
Date: "date",
JSONB: "jsonb",
JSON: "json",
}
#: Типы с параметрами: точный класс -> отрисовка с параметрами (длина, точность, часовой пояс).
_PARAMETRIZED_TYPES: dict[type, Callable[[Any], str]] = {
Numeric: lambda t: f"numeric({t.precision},{t.scale})",
String: lambda t: "varchar" if t.length is None else f"varchar({t.length})",
DateTime: lambda t: "timestamptz" if t.timezone else "timestamp",
}
def render_orm_type(orm_type: object) -> str:
"""SQL-тип ORM-колонки в нотации DBML (`numeric(18,2)`, `varchar`, `timestamptz`, имя enum, ...).
Классы сравниваются точно: `SmallInteger` — подкласс `Integer`, `Text` — подкласс `String`, `Float` — подкласс
`Numeric`; подкласс не должен «проходить» за базовый тип.
"""
if isinstance(orm_type, SAEnum):
return str(orm_type.name) if orm_type.schema == SCHEMA else f"{orm_type.schema}.{orm_type.name}"
if isinstance(orm_type, UUID):
return "uuid" if orm_type.as_uuid else "uuid(as_uuid=False)"
cls = type(orm_type)
if cls in _PARAMETRIZED_TYPES:
return _PARAMETRIZED_TYPES[cls](orm_type)
return _SIMPLE_TYPE_NAMES.get(cls, cls.__name__.lower())
def render_dbml_type(dbml_type: object) -> str | None:
"""Тип DBML в канонической нотации; `None` — тип не поддержан сверкой."""
if isinstance(dbml_type, DbmlEnum):
return str(dbml_type.name)
name = str(dbml_type).strip().lower()
numeric = _NUMERIC_RE.fullmatch(name)
if numeric:
return f"numeric({numeric.group(1)},{numeric.group(2)})"
return name if name in SUPPORTED_DBML_TYPES else None
def type_pair(orm_type: object, dbml_type: object) -> tuple[str, str] | None:
"""(ожидаемый, фактический) тип в нотации DBML; совпадение — равенство строк. `None` — тип DBML не поддержан.
Для enum с верным именем и схемой сравниваются ещё и значения в порядке объявления: при расхождении к имени
приписываются списки значений, чтобы строка отчёта показывала разницу.
"""
expected = render_dbml_type(dbml_type)
if expected is None:
return None
actual = render_orm_type(orm_type)
if isinstance(dbml_type, DbmlEnum) and isinstance(orm_type, SAEnum) and actual == expected:
expected_values = [item.name for item in dbml_type.items]
if list(orm_type.enums) != expected_values:
return f"{expected}{expected_values}", f"{actual}{list(orm_type.enums)}"
return expected, actual
def type_problem(orm_type: object, dbml_type: object) -> str | None:
"""Расхождение только типа (`ожидалось X / фактически Y`), сообщение о неподдержанном типе DBML или `None`."""
pair = type_pair(orm_type, dbml_type)
if pair is None:
return f"тип DBML `{dbml_type}` не поддержан сверкой (добавьте в dbml_parity.SUPPORTED_DBML_TYPES)"
expected, actual = pair
return None if expected == actual else f"ожидалось {expected} / фактически {actual}"
# --- default ---------------------------------------------------------------------------------------------------
def _canon_expr(text: str) -> str:
text = " ".join(text.split())
if text.startswith("'"):
return text # строковый литерал — регистр значим
text = text.lower()
return _SYNONYMS.get(text, text)
def canonical_default_dbml(default: object) -> str:
"""Канонический вид DBML-default: bool/число/строка-литерал/`Expression`."""
if isinstance(default, Expression):
return _canon_expr(str(default.text))
if isinstance(default, bool):
return "true" if default else "false"
if isinstance(default, int | float):
return str(default)
return "'" + str(default).replace("'", "''") + "'"
def canonical_default_orm(server_default: object) -> str:
"""Канонический вид ORM `server_default` (`text(...)`, `func.now()`, строка)."""
arg = getattr(server_default, "arg", server_default)
if isinstance(arg, TextClause):
return _canon_expr(arg.text)
if isinstance(arg, FunctionElement):
return _canon_expr(f"{arg.name}()")
return _canon_expr(str(arg))
# --- колонки ---------------------------------------------------------------------------------------------------
def _nullability(nullable: bool) -> str:
return "NULL" if nullable else "NOT NULL"
def _yes_no(flag: bool) -> str:
return "да" if flag else "нет"
def _column_problems(orm_table: Table, dbml_table: DbmlTable) -> list[str]:
"""Состав, порядок и свойства колонок одной таблицы."""
name = dbml_table.name
dbml_names = [c.name for c in dbml_table.columns]
orm_names = [c.name for c in orm_table.columns]
problems = [missing(f"{name}.{col}", ORM) for col in dbml_names if col not in orm_names]
problems += [missing(f"{name}.{col}", DBML) for col in orm_names if col not in dbml_names]
# порядок сравнивается только по общим колонкам: пропавшая/лишняя колонка уже названа выше
common_dbml = [col for col in dbml_names if col in orm_names]
common_orm = [col for col in orm_names if col in dbml_names]
if common_dbml != common_orm:
problems.append(mismatch(f"{name}: порядок колонок", common_dbml, common_orm))
composite_pk = {c.name for i in dbml_table.indexes if i.pk for c in i.subjects}
for dbml_col in dbml_table.columns:
if dbml_col.name in orm_table.columns:
label = f"{name}.{dbml_col.name}"
problems += _one_column_problems(label, orm_table.columns[dbml_col.name], dbml_col, composite_pk)
return problems
def _one_column_problems(label: str, orm_col: Column[Any], dbml_col: DbmlColumn, composite_pk: set[str]) -> list[str]:
problems: list[str] = []
is_pk = bool(dbml_col.pk) or dbml_col.name in composite_pk
# тип и nullability — одной строкой, как в спеке: «ожидалось numeric(18,2) NULL / фактически numeric(18,4) NOT NULL»
pair = type_pair(orm_col.type, dbml_col.type)
if pair is None:
problems.append(f"{label}: тип DBML `{dbml_col.type}` не поддержан сверкой")
else:
expected_null = _nullability(not (dbml_col.not_null or is_pk))
actual_null = _nullability(bool(orm_col.nullable))
expected, actual = f"{pair[0]} {expected_null}", f"{pair[1]} {actual_null}"
if expected != actual:
problems.append(mismatch(label, expected, actual))
if orm_col.primary_key != is_pk:
problems.append(mismatch(f"{label}: первичный ключ", _yes_no(is_pk), _yes_no(orm_col.primary_key)))
if bool(orm_col.unique) != bool(dbml_col.unique):
problems.append(mismatch(f"{label}: unique", _yes_no(bool(dbml_col.unique)), _yes_no(bool(orm_col.unique))))
expected_default = ABSENT if dbml_col.default is None else canonical_default_dbml(dbml_col.default)
actual_default = ABSENT if orm_col.server_default is None else canonical_default_orm(orm_col.server_default)
if expected_default != actual_default:
problems.append(mismatch(f"{label}: default", expected_default, actual_default))
return problems
# --- FK --------------------------------------------------------------------------------------------------------
def _fk_problems(metadata: MetaData, dbml: PyDBML, names: Collection[str]) -> list[str]:
"""FK по колонке-ссылке: множество целей `родитель.колонка` в DBML против ORM."""
expected: dict[tuple[str, str], set[str]] = {}
for ref in dbml.refs:
# `>`: table1 ссылается на table2 (many-to-one); `<` — наоборот; `-` (one-to-one) читаем как `>`
child, child_cols, parent, parent_cols = ref.table1, ref.col1, ref.table2, ref.col2
if ref.type == "<":
child, child_cols, parent, parent_cols = parent, parent_cols, child, child_cols
if child.name in names:
for child_col, parent_col in zip(child_cols, parent_cols, strict=True):
expected.setdefault((child.name, child_col.name), set()).add(f"{parent.name}.{parent_col.name}")
actual: dict[tuple[str, str], set[str]] = {}
for table in metadata.tables.values():
if table.name in names:
for fk in table.foreign_keys:
target = f"{fk.column.table.name}.{fk.column.name}"
actual.setdefault((table.name, fk.parent.name), set()).add(target)
problems = []
for key in sorted(expected.keys() | actual.keys()):
want, have = expected.get(key, set()), actual.get(key, set())
if want != have:
problems.append(mismatch(f"{key[0]}.{key[1]}: FK", _targets(want), _targets(have)))
return problems
def _targets(targets: set[str]) -> str:
return ", ".join(sorted(targets)) or ABSENT
# --- индексы ---------------------------------------------------------------------------------------------------
def _norm_where(where: object) -> str | None:
if where is None:
return None
text = where.text if isinstance(where, TextClause) else str(where)
return " ".join(text.split()).lower()
def _index_props(unique: bool, where: str | None, nulls_not_distinct: bool) -> str:
"""Свойства индекса текстом: `unique WHERE is_current NULLS NOT DISTINCT` / `не unique`."""
parts = ["unique" if unique else "не unique"]
if where:
parts.append(f"WHERE {where}")
if nulls_not_distinct:
parts.append("NULLS NOT DISTINCT")
return " ".join(parts)
def _index_problems(orm_table: Table, dbml_table: DbmlTable) -> list[str]:
"""Индексы, сгруппированные по набору колонок: состав и свойства (unique, WHERE, NULLS NOT DISTINCT)."""
expected: dict[tuple[str, ...], list[str]] = {}
for index in dbml_table.indexes:
if index.pk:
continue # составной [pk] — первичный ключ, сверяется в колонках
note = str(index.note) if index.note else ""
partial = _PARTIAL_RE.search(note)
props = _index_props(
bool(index.unique) or partial is not None,
_norm_where(partial.group(1)) if partial else None,
# note с `NULLS NOT DISTINCT` => уникальный индекс обязан иметь postgresql_nulls_not_distinct=True
_NULLS_NOT_DISTINCT_RE.search(note) is not None,
)
expected.setdefault(tuple(c.name for c in index.subjects), []).append(props)
actual: dict[tuple[str, ...], list[str]] = {}
for orm_index in orm_table.indexes:
options = orm_index.dialect_options["postgresql"]
props = _index_props(bool(orm_index.unique), _norm_where(options["where"]), bool(options["nulls_not_distinct"]))
actual.setdefault(tuple(c.name for c in orm_index.columns), []).append(props)
problems = []
for columns in sorted(expected.keys() | actual.keys()):
label = f"{dbml_table.name}: индекс ({', '.join(columns)})"
want, have = sorted(expected.get(columns, [])), sorted(actual.get(columns, []))
if not have:
problems.append(missing(label, ORM))
elif not want:
problems.append(missing(label, DBML))
elif want != have:
problems.append(mismatch(label, "; ".join(want), "; ".join(have)))
return problems
# --- точки входа -----------------------------------------------------------------------------------------------
def find_problems(metadata: MetaData, tables: Collection[str], dbml: PyDBML) -> list[str]:
"""Расхождения ORM (`metadata`) с DBML для таблиц `tables`; пустой список — совпадение."""
dbml_tables = {t.name: t for t in dbml.tables}
orm_tables = {t.name: t for t in metadata.tables.values()}
problems: list[str] = []
for name in sorted(tables):
if name not in dbml_tables:
problems.append(missing(name, DBML))
elif name not in orm_tables:
problems.append(missing(name, ORM))
else:
if orm_tables[name].schema != SCHEMA:
problems.append(mismatch(f"{name}: схема", SCHEMA, orm_tables[name].schema))
problems += _column_problems(orm_tables[name], dbml_tables[name])
problems += _index_problems(orm_tables[name], dbml_tables[name])
return problems + _fk_problems(metadata, dbml, {n for n in tables if n in dbml_tables})
def enum_problems(enums: Mapping[str, type[enum.Enum]], dbml: PyDBML) -> list[str]:
"""Enum'ы ORM-реестра (имя DBML → класс) против блоков `Enum {}` DBML: множество имён и значения по порядку."""
dbml_enums = {e.name: [item.name for item in e.items] for e in dbml.enums}
problems: list[str] = []
for name in sorted(dbml_enums.keys() | enums.keys()):
if name not in enums:
problems.append(missing(f"enum {name}", ORM))
elif name not in dbml_enums:
problems.append(missing(f"enum {name}", DBML))
else:
actual = [member.value for member in enums[name]]
if actual != dbml_enums[name]:
problems.append(mismatch(f"enum {name}", dbml_enums[name], actual))
return problems
def find_all_problems(metadata: MetaData, enums: Mapping[str, type[enum.Enum]], dbml: PyDBML) -> list[str]:
"""Полная сверка: все таблицы обеих сторон (количество — следствие сравнения множеств) и все enum'ы."""
names = {t.name for t in dbml.tables} | {t.name for t in metadata.tables.values()}
return find_problems(metadata, names, dbml) + enum_problems(enums, dbml)

View File

@ -1,19 +1,18 @@
"""Поиск и разбор DBML-модели для тестов (переиспользуется тестами чётности ORM и DBML)."""
"""Поиск и разбор DBML-модели для тестов: обёртка над `scripts.dbml_parity` с пропуском теста, если docs недоступен."""
import os
from pathlib import Path
import pytest
from pydbml import PyDBML
DBML_FILE_NAME = "data-model-v5.13.dbml"
from scripts.dbml_parity import DBML_FILE_NAME, default_dbml_path, parse_dbml
__all__ = ["DBML_FILE_NAME", "dbml_path", "load_dbml"]
def dbml_path() -> Path:
"""Путь к DBML: каталог docs из `ARM_KA_DOCS`, иначе соседний `../../docs` относительно `arm-core/api`."""
docs = os.environ.get("ARM_KA_DOCS")
docs_dir = Path(docs) if docs else Path(__file__).resolve().parents[3] / "docs"
return docs_dir / "model" / DBML_FILE_NAME
return default_dbml_path()
def load_dbml() -> PyDBML:
@ -21,4 +20,4 @@ def load_dbml() -> PyDBML:
path = dbml_path()
if not path.is_file():
pytest.skip(f"docs недоступен: нет {path} (задайте ARM_KA_DOCS или положите docs рядом с arm-core)")
return PyDBML.parse_file(path)
return parse_dbml(path)

View File

@ -1,263 +1,20 @@
"""Сверка ORM-метаданных с DBML v5.13 для заданного набора таблиц (без БД); переиспользуется тестами частей 01–09.
"""Тонкая обёртка тестов над `scripts.dbml_parity` (единственная реализация сверки ORM с DBML, рулинг R-14).
Публичное API:
- `find_problems(metadata, tables)` — список расхождений (пустой = ORM совпадает с DBML) по колонкам (имя, порядок,
тип, nullable, PK, unique, server_default), FK и индексам (состав, уникальность, partial-условие WHERE,
`NULLS NOT DISTINCT` по note DBML);
- `type_problem(orm_type, dbml_type)` — расхождение типа одной колонки (или `None`);
- `canonical_default_dbml` / `canonical_default_orm` — приведение default к сопоставимому виду.
Нетривиальные места. DBML-default — python-литерал pydbml (bool/int/float/str) или `Expression` (обратные кавычки,
напр. `now()`); ORM server_default — `text(...)` или SQL-функция (`func.now()`). Обе стороны приводятся к строке:
bool -> `true`/`false`, число -> как есть, строка -> в одинарных кавычках, выражение -> текст в нижнем регистре.
Отличие одно: `find_problems` без явного `dbml` берёт DBML через `load_dbml()` (пропуск теста, если docs недоступен).
"""
import re
from collections.abc import Callable, Collection
from collections.abc import Collection
from pydbml import PyDBML
from pydbml.classes import Column as DbmlColumn
from pydbml.classes import Enum as DbmlEnum
from pydbml.classes import Expression
from pydbml.classes import Table as DbmlTable
from sqlalchemy import (
BigInteger,
Boolean,
Date,
DateTime,
Integer,
MetaData,
Numeric,
SmallInteger,
String,
Table,
Text,
)
from sqlalchemy import Enum as SAEnum
from sqlalchemy.dialects.postgresql import JSONB, UUID
from sqlalchemy.sql.elements import TextClause
from sqlalchemy.sql.functions import FunctionElement
from sqlalchemy import MetaData
from scripts import dbml_parity
from scripts.dbml_parity import canonical_default_dbml, canonical_default_orm, type_problem
from tests.dbml_helpers import load_dbml
SCHEMA = "dkb"
_NUMERIC_RE = re.compile(r"numeric\((\d+),\s*(\d+)\)")
_PARTIAL_RE = re.compile(r"partial\s+UNIQUE\s+WHERE\s+(.+?)(?:\s+—.*)?$", re.IGNORECASE | re.DOTALL)
_NULLS_NOT_DISTINCT_RE = re.compile(r"NULLS\s+NOT\s+DISTINCT", re.IGNORECASE)
_SYNONYMS = {"current_timestamp": "now()", "current_timestamp()": "now()"}
# --- типы -------------------------------------------------------------------------------------------------------
def _is_exact(orm_type: object, cls: type) -> bool:
"""Точное совпадение класса (SmallInteger — подкласс Integer, Text — подкласс String, и т.д.)."""
return type(orm_type) is cls
_SIMPLE_TYPES: dict[str, Callable[[object], bool]] = {
"uuid": lambda t: isinstance(t, UUID) and t.as_uuid is True,
"varchar": lambda t: _is_exact(t, String) and t.length is None, # type: ignore[attr-defined]
"text": lambda t: _is_exact(t, Text),
"smallint": lambda t: _is_exact(t, SmallInteger),
"int": lambda t: _is_exact(t, Integer),
"bigint": lambda t: _is_exact(t, BigInteger),
"boolean": lambda t: _is_exact(t, Boolean),
"date": lambda t: _is_exact(t, Date),
"timestamptz": lambda t: _is_exact(t, DateTime) and t.timezone is True, # type: ignore[attr-defined]
"jsonb": lambda t: isinstance(t, JSONB),
}
def type_problem(orm_type: object, dbml_type: object) -> str | None:
"""Расхождение SQL-типа ORM-колонки с типом DBML; `dbml_type` — строка типа или объект enum pydbml."""
if isinstance(dbml_type, DbmlEnum):
return _enum_problem(orm_type, dbml_type)
name = str(dbml_type).strip().lower()
numeric = _NUMERIC_RE.fullmatch(name)
if numeric:
precision, scale = int(numeric.group(1)), int(numeric.group(2))
if _is_exact(orm_type, Numeric) and (orm_type.precision, orm_type.scale) == (precision, scale): # type: ignore[attr-defined]
return None
return f"ожидался Numeric({precision}, {scale}), фактически {_describe(orm_type)}"
check = _SIMPLE_TYPES.get(name)
if check is None:
return f"тип DBML `{dbml_type}` не поддержан сверкой (добавьте в dbml_parity._SIMPLE_TYPES)"
if check(orm_type):
return None
return f"ожидался тип для DBML `{name}`, фактически {_describe(orm_type)}"
def _enum_problem(orm_type: object, dbml_enum: DbmlEnum) -> str | None:
if not isinstance(orm_type, SAEnum):
return f"ожидался enum {dbml_enum.name}, фактически {_describe(orm_type)}"
expected_values = [item.name for item in dbml_enum.items]
if orm_type.name != dbml_enum.name:
return f"имя enum: ожидалось {dbml_enum.name}, фактически {orm_type.name}"
if orm_type.schema != SCHEMA:
return f"схема enum {dbml_enum.name}: ожидалась {SCHEMA}, фактически {orm_type.schema}"
if list(orm_type.enums) != expected_values:
return f"значения enum {dbml_enum.name}: ожидалось {expected_values}, фактически {list(orm_type.enums)}"
return None
def _describe(orm_type: object) -> str:
extra = ""
if isinstance(orm_type, Numeric):
extra = f"({orm_type.precision}, {orm_type.scale})"
elif isinstance(orm_type, DateTime):
extra = f"(timezone={orm_type.timezone})"
elif isinstance(orm_type, String):
extra = f"(length={orm_type.length})"
elif isinstance(orm_type, SAEnum):
extra = f"({orm_type.name})"
return f"{type(orm_type).__name__}{extra}"
# --- default ----------------------------------------------------------------------------------------------------
def _canon_expr(text: str) -> str:
text = " ".join(text.split())
if text.startswith("'"):
return text # строковый литерал — регистр значим
text = text.lower()
return _SYNONYMS.get(text, text)
def canonical_default_dbml(default: object) -> str:
"""Канонический вид DBML-default: bool/число/строка-литерал/`Expression`."""
if isinstance(default, Expression):
return _canon_expr(str(default.text))
if isinstance(default, bool):
return "true" if default else "false"
if isinstance(default, int | float):
return str(default)
return "'" + str(default).replace("'", "''") + "'"
def canonical_default_orm(server_default: object) -> str:
"""Канонический вид ORM `server_default` (`text(...)`, `func.now()`, строка)."""
arg = getattr(server_default, "arg", server_default)
if isinstance(arg, TextClause):
return _canon_expr(arg.text)
if isinstance(arg, FunctionElement):
return _canon_expr(f"{arg.name}()")
return _canon_expr(str(arg))
# --- колонки, FK, индексы -------------------------------------------------------------------------------------
def _column_problems(orm_table: Table, dbml_table: DbmlTable) -> list[str]:
name = dbml_table.name
problems: list[str] = []
dbml_cols = {c.name: c for c in dbml_table.columns}
orm_names = [c.name for c in orm_table.columns]
if orm_names != list(dbml_cols):
problems.append(f"{name}: колонки/порядок: ожидалось {list(dbml_cols)}, фактически {orm_names}")
composite_pk = {c.name for i in dbml_table.indexes if i.pk for c in i.subjects}
for col_name, dbml_col in dbml_cols.items():
if col_name not in orm_table.columns:
continue
problems.extend(_one_column_problems(f"{name}.{col_name}", orm_table.columns[col_name], dbml_col, composite_pk))
return problems
def _one_column_problems(label: str, orm_col, dbml_col: DbmlColumn, composite_pk: set[str]) -> list[str]: # type: ignore[no-untyped-def]
problems: list[str] = []
is_pk = bool(dbml_col.pk) or dbml_col.name in composite_pk
if orm_col.primary_key != is_pk:
problems.append(f"{label}: primary_key ожидался {is_pk}")
if orm_col.nullable != (not (dbml_col.not_null or is_pk)):
problems.append(f"{label}: nullable фактически {orm_col.nullable}")
if bool(orm_col.unique) != bool(dbml_col.unique):
problems.append(f"{label}: unique ожидался {bool(dbml_col.unique)}")
type_diff = type_problem(orm_col.type, dbml_col.type)
if type_diff:
problems.append(f"{label}: тип: {type_diff}")
if dbml_col.default is None:
if orm_col.server_default is not None:
problems.append(f"{label}: лишний server_default {canonical_default_orm(orm_col.server_default)}")
elif orm_col.server_default is None:
problems.append(f"{label}: нет server_default (ожидался {canonical_default_dbml(dbml_col.default)})")
else:
expected, actual = canonical_default_dbml(dbml_col.default), canonical_default_orm(orm_col.server_default)
if expected != actual:
problems.append(f"{label}: server_default ожидался {expected}, фактически {actual}")
return problems
def _fk_problems(orm_tables: dict[str, Table], dbml: PyDBML, names: Collection[str]) -> list[str]:
expected: set[tuple[str, str, str, str]] = set()
for ref in dbml.refs:
child, child_col, parent, parent_col = ref.table1, ref.col1[0], ref.table2, ref.col2[0]
if ref.type == "<":
child, child_col, parent, parent_col = parent, parent_col, child, child_col
if child.name in names:
expected.add((child.name, child_col.name, parent.name, parent_col.name))
actual = {
(t.name, fk.parent.name, fk.column.table.name, fk.column.name)
for t in orm_tables.values()
for fk in t.foreign_keys
if t.name in names
}
problems = [f"FK отсутствует в ORM: {fk}" for fk in sorted(expected - actual)]
return problems + [f"FK лишний в ORM: {fk}" for fk in sorted(actual - expected)]
def _norm_where(where: object) -> str | None:
if where is None:
return None
text = where.text if isinstance(where, TextClause) else str(where)
return " ".join(text.split()).lower()
def _index_problems(orm_table: Table, dbml_table: DbmlTable) -> list[str]:
expected: list[tuple[tuple[str, ...], bool, str | None, bool]] = []
for i in dbml_table.indexes:
if i.pk:
continue
note = str(i.note) if i.note else ""
partial = _PARTIAL_RE.search(note)
where = _norm_where(partial.group(1)) if partial else None
# note с `NULLS NOT DISTINCT` => уникальный индекс обязан иметь postgresql_nulls_not_distinct=True
nulls_not_distinct = _NULLS_NOT_DISTINCT_RE.search(note) is not None
expected.append(
(tuple(c.name for c in i.subjects), bool(i.unique) or partial is not None, where, nulls_not_distinct)
)
actual = [
(
tuple(c.name for c in i.columns),
bool(i.unique),
_norm_where(i.dialect_options["postgresql"]["where"]),
bool(i.dialect_options["postgresql"]["nulls_not_distinct"]),
)
for i in orm_table.indexes
]
if sorted(actual, key=repr) == sorted(expected, key=repr):
return []
return [
f"{dbml_table.name}: индексы (колонки, unique, WHERE, NULLS NOT DISTINCT): "
f"ожидалось {sorted(expected, key=repr)}, фактически {sorted(actual, key=repr)}"
]
# --- точка входа ------------------------------------------------------------------------------------------------
__all__ = ["canonical_default_dbml", "canonical_default_orm", "find_problems", "type_problem"]
def find_problems(metadata: MetaData, tables: Collection[str], dbml: PyDBML | None = None) -> list[str]:
"""Все расхождения ORM (`metadata`) с DBML для таблиц `tables`; пустой список — совпадение."""
dbml = dbml or load_dbml()
dbml_tables = {t.name: t for t in dbml.tables}
orm_tables = {t.name: t for t in metadata.tables.values()}
problems: list[str] = []
for name in sorted(tables):
if name not in dbml_tables:
problems.append(f"{name}: нет в DBML")
elif name not in orm_tables:
problems.append(f"{name}: нет в ORM-метаданных")
else:
problems.extend(_column_problems(orm_tables[name], dbml_tables[name]))
problems.extend(_index_problems(orm_tables[name], dbml_tables[name]))
return problems + _fk_problems(orm_tables, dbml, tables)
"""Расхождения ORM с DBML для таблиц `tables` (см. `scripts.dbml_parity.find_problems`)."""
return dbml_parity.find_problems(metadata, tables, dbml or load_dbml())

View File

@ -0,0 +1,162 @@
"""CLI сверки ORM с DBML (`scripts/check_dbml_vs_orm.py`): сценарии спеки data-model «Сверка ORM с DBML»,
«Лишняя колонка в ORM», «Расхождение типа или nullability», «Новая версия DBML без обновления ORM».
Мутации ORM делаются на копии метаданных (`run_check(..., metadata=...)`), реальный запуск — через subprocess.
"""
import enum
import os
import subprocess
import sys
from pathlib import Path
import pytest
from sqlalchemy import Column, MetaData, Numeric, String, Table
from scripts.check_dbml_vs_orm import EXIT_USAGE, run_check
from src.models import Base
from src.models.enums import ENUMS
from tests.dbml_helpers import dbml_path
API_DIR = Path(__file__).resolve().parents[1]
SCRIPT = "scripts/check_dbml_vs_orm.py"
@pytest.fixture
def dbml_file() -> Path:
path = dbml_path()
if not path.is_file():
pytest.skip(f"docs недоступен: нет {path} (задайте ARM_KA_DOCS или положите docs рядом с arm-core)")
return path
def clone_metadata() -> MetaData:
"""Копия реальных метаданных ORM, которую можно портить, не затрагивая `Base.metadata`."""
clone = MetaData(schema=Base.metadata.schema, naming_convention=Base.metadata.naming_convention)
for table in Base.metadata.sorted_tables:
table.to_metadata(clone)
return clone
def _run_cli(*args: str) -> subprocess.CompletedProcess[str]:
"""Запуск скрипта как файла из `api/` без ENV_FILE: скрипт обязан обойтись без рабочего `.env`."""
env = {k: v for k, v in os.environ.items() if k not in {"ENV_FILE", "APP_ENV"}}
return subprocess.run(
[sys.executable, SCRIPT, *args], cwd=API_DIR, env=env, capture_output=True, text=True, check=False
)
# --- реальный запуск -------------------------------------------------------------------------------------------
def test_cli_on_real_models_prints_nothing_and_exits_0(dbml_file: Path) -> None:
result = _run_cli(str(dbml_file))
assert (result.returncode, result.stdout) == (0, ""), result.stdout + result.stderr
def test_cli_missing_file_exits_2_with_message(tmp_path: Path) -> None:
result = _run_cli(str(tmp_path / "no-such.dbml"))
assert result.returncode == EXIT_USAGE
assert result.stdout == ""
assert "файл DBML не найден" in result.stderr
def test_run_check_on_real_models_is_clean(dbml_file: Path) -> None:
assert run_check(dbml_file) == (0, [])
# --- сценарий «Лишняя колонка в ORM» ---------------------------------------------------------------------------
def test_extra_orm_column_is_reported_missing_in_dbml(dbml_file: Path) -> None:
metadata = clone_metadata()
metadata.tables["dkb.osv_entry"].append_column(Column("extra_note", String()))
assert run_check(dbml_file, metadata=metadata) == (1, ["osv_entry.extra_note: отсутствует в DBML"])
def test_removed_orm_column_is_reported_missing_in_orm(dbml_file: Path) -> None:
metadata = clone_metadata()
osv_entry = metadata.tables["dkb.osv_entry"]
osv_entry._columns.remove(osv_entry.columns["opening_credit"]) # noqa: SLF001 — намеренная порча копии
assert run_check(dbml_file, metadata=metadata) == (1, ["osv_entry.opening_credit: отсутствует в ORM"])
# --- сценарий «Расхождение типа или nullability» --------------------------------------------------------------
def test_type_and_nullability_mismatch_is_one_line_in_dbml_notation(dbml_file: Path) -> None:
metadata = clone_metadata()
column = metadata.tables["dkb.osv_entry"].columns["opening_debit"]
column.type = Numeric(18, 4)
column.nullable = False
assert run_check(dbml_file, metadata=metadata) == (
1,
["osv_entry.opening_debit: ожидалось numeric(18,2) NULL / фактически numeric(18,4) NOT NULL"],
)
def test_nullability_only_mismatch(dbml_file: Path) -> None:
metadata = clone_metadata()
metadata.tables["dkb.osv_entry"].columns["opening_debit"].nullable = False
code, lines = run_check(dbml_file, metadata=metadata)
assert (code, lines) == (
1,
["osv_entry.opening_debit: ожидалось numeric(18,2) NULL / фактически numeric(18,2) NOT NULL"],
)
# --- таблицы и enum'ы: сравнение множеств ----------------------------------------------------------------------
def test_missing_and_extra_tables_are_reported(dbml_file: Path) -> None:
metadata = clone_metadata()
metadata.remove(metadata.tables["dkb.consolidation_lock"])
code, lines = run_check(dbml_file, metadata=metadata)
assert code == 1
assert "consolidation_lock: отсутствует в ORM" in lines
metadata = clone_metadata()
Table("orphan_table", metadata, Column("id", String(), primary_key=True))
assert run_check(dbml_file, metadata=metadata) == (1, ["orphan_table: отсутствует в DBML"])
def test_enum_values_and_sets_are_compared(dbml_file: Path) -> None:
enums: dict[str, type[enum.Enum]] = dict(ENUMS)
enums["entity_type"] = enum.Enum("EntityType", {"ИП": "ИП", "ЮЛ": "ЮЛ"}) # порядок значений нарушен
del enums["audit_section"]
enums["no_such_enum"] = enum.Enum("NoSuch", {"A": "A"})
code, lines = run_check(dbml_file, enums=enums)
assert code == 1
assert sorted(lines) == sorted(
[
"enum entity_type: ожидалось ['ЮЛ', 'ИП'] / фактически ['ИП', 'ЮЛ']",
"enum audit_section: отсутствует в ORM",
"enum no_such_enum: отсутствует в DBML",
]
)
# --- сценарий «Новая версия DBML без обновления ORM» -----------------------------------------------------------
def test_newer_dbml_version_fails_before_comparing_tables(tmp_path: Path, dbml_file: Path) -> None:
lines = dbml_file.read_text(encoding="utf-8").splitlines()
newer = tmp_path / "data-model-v5.14.dbml"
# шапка 5.14 + заведомо сломанное тело: если бы таблицы сравнивались, были бы другие строки/ошибка разбора
newer.write_text(lines[0].replace("v5.13", "v5.14") + "\nTable broken {{{\n", encoding="utf-8")
assert run_check(newer) == (1, ["версия DBML 5.14 не совпадает с версией ORM 5.13: обновите модели"])
def test_newer_dbml_version_via_cli(tmp_path: Path, dbml_file: Path) -> None:
newer = tmp_path / "data-model-v5.14.dbml"
newer.write_text(dbml_file.read_text(encoding="utf-8").replace("v5.13", "v5.14", 1), encoding="utf-8")
result = _run_cli(str(newer))
assert result.returncode == 1
assert result.stdout == "версия DBML 5.14 не совпадает с версией ORM 5.13: обновите модели\n"
def test_header_without_version_is_usage_error(tmp_path: Path) -> None:
bad = tmp_path / "model.dbml"
bad.write_text("// модель без версии\nTable t {\n id uuid [pk]\n}\n", encoding="utf-8")
code, lines = run_check(bad)
assert code == EXIT_USAGE
assert len(lines) == 1 and "версия" in lines[0]

View File

@ -1,4 +1,5 @@
"""Самопроверка сверки ORM с DBML (`dbml_parity`): компаратор обязан находить заведомо неверные расхождения."""
"""Самопроверка сверки ORM с DBML (`scripts.dbml_parity`): компаратор обязан находить заведомо неверные расхождения
и описывать их строками единого формата `<объект>: ожидалось X / фактически Y` / `<объект>: отсутствует в ORM|DBML`."""
import pytest
from pydbml.classes import Expression
@ -41,22 +42,26 @@ def test_clone_is_clean_baseline() -> None:
@pytest.mark.parametrize(
("table", "column", "wrong_type", "needle"),
("table", "column", "wrong_type", "expected"),
[
("regional_office", "name", Text(), "name: тип"), # String -> Text
("regional_office", "name", String(50), "name: тип"), # varchar с длиной
("app_user", "created_at", DateTime(), "created_at: тип"), # без timezone
("elimination_rule", "rule_no", Integer(), "rule_no: тип"), # smallint -> int
("elimination_rule", "valid_from", DateTime(timezone=True), "valid_from: тип"), # date -> timestamptz
("permission", "action", pg_enum("permission_object"), "action: тип"), # не тот enum
("app_user", "regional_office_id", String(), "regional_office_id: тип"), # uuid -> varchar
("regional_office", "name", Text(), "varchar NOT NULL / фактически text NOT NULL"), # String -> Text
("regional_office", "name", String(50), "varchar NOT NULL / фактически varchar(50) NOT NULL"), # с длиной
("app_user", "created_at", DateTime(), "timestamptz NULL / фактически timestamp NULL"), # без timezone
("elimination_rule", "rule_no", Integer(), "smallint NOT NULL / фактически int NOT NULL"),
("elimination_rule", "valid_from", DateTime(timezone=True), "date NOT NULL / фактически timestamptz NOT NULL"),
( # не тот enum
"permission",
"action",
pg_enum("permission_object"),
"permission_action NOT NULL / фактически permission_object NOT NULL",
),
("app_user", "regional_office_id", String(), "uuid NOT NULL / фактически varchar NOT NULL"), # uuid -> varchar
],
)
def test_wrong_column_type_is_detected(table: str, column: str, wrong_type: object, needle: str) -> None:
def test_wrong_column_type_is_detected(table: str, column: str, wrong_type: object, expected: str) -> None:
metadata = clone_metadata()
metadata.tables[f"dkb.{table}"].columns[column].type = wrong_type # type: ignore[assignment]
problems = find_problems(metadata, TABLES)
assert any(needle in p and p.startswith(f"{table}.") for p in problems), problems
assert find_problems(metadata, TABLES) == [f"{table}.{column}: ожидалось {expected}"]
def test_enum_with_wrong_schema_or_values_is_detected() -> None:
@ -109,10 +114,11 @@ def test_missing_wrong_or_extra_server_default_is_detected() -> None:
cols["is_active"].server_default = None
cols["is_head_office"].server_default = _default("true") # в DBML false
cols["name"].server_default = _default("'x'") # в DBML default нет
problems = find_problems(metadata, TABLES)
assert any("is_active: нет server_default" in p for p in problems), problems
assert any("is_head_office: server_default ожидался false, фактически true" in p for p in problems), problems
assert any("name: лишний server_default" in p for p in problems), problems
assert sorted(find_problems(metadata, TABLES)) == [
"regional_office.is_active: default: ожидалось true / фактически нет",
"regional_office.is_head_office: default: ожидалось false / фактически true",
"regional_office.name: default: ожидалось нет / фактически 'x'",
]
def _default(sql: str) -> object:
@ -124,12 +130,23 @@ def test_nullable_pk_unique_and_order_mismatches_are_detected() -> None:
ro = metadata.tables["dkb.regional_office"]
ro.columns["name"].nullable = True
ro.columns["name"].unique = None
problems = find_problems(metadata, {"regional_office"})
assert any("name: nullable" in p for p in problems), problems
assert any("name: unique" in p for p in problems), problems
assert find_problems(metadata, {"regional_office"}) == [
"regional_office.name: ожидалось varchar NOT NULL / фактически varchar NULL",
"regional_office.name: unique: ожидалось да / фактически нет",
]
ro.columns["id"].primary_key = False
assert "regional_office.id: первичный ключ: ожидалось да / фактически нет" in find_problems(
metadata, {"regional_office"}
)
perm = metadata.tables["dkb.permission"]
perm._columns.remove(perm.columns["name"]) # noqa: SLF001 — намеренная порча копии
assert any("колонки/порядок" in p for p in find_problems(metadata, {"permission"}))
name_col = perm.columns["name"]
perm._columns.remove(name_col) # noqa: SLF001 — намеренная порча копии
assert find_problems(metadata, {"permission"}) == ["permission.name: отсутствует в ORM"]
perm.append_column(name_col) # вернулась в конец: состав совпал, порядок — нет
[problem] = find_problems(metadata, {"permission"})
assert problem.startswith(
"permission: порядок колонок: ожидалось ['id', 'code', 'object_type', 'action', 'name'"
), problem
def test_missing_extra_and_wrong_where_indexes_are_detected() -> None:
@ -137,15 +154,24 @@ def test_missing_extra_and_wrong_where_indexes_are_detected() -> None:
member = metadata.tables["dkb.related_group_member"]
index = next(i for i in member.indexes if i.name == "uq_related_group_member_entity_current")
index.dialect_options["postgresql"]["where"] = text("is_current AND valid_to IS NULL")
assert any("индексы" in p for p in find_problems(metadata, {"related_group_member"}))
label = "related_group_member: индекс (legal_entity_id)"
assert find_problems(metadata, {"related_group_member"}) == [
f"{label}: ожидалось unique WHERE is_current / фактически unique WHERE is_current and valid_to is null"
]
index.dialect_options["postgresql"]["where"] = None # partial превращён в обычный unique
assert any("индексы" in p for p in find_problems(metadata, {"related_group_member"}))
assert find_problems(metadata, {"related_group_member"}) == [
f"{label}: ожидалось unique WHERE is_current / фактически unique"
]
index.dialect_options["postgresql"]["where"] = text("is_current")
assert find_problems(metadata, {"related_group_member"}) == []
# лишний partial там, где DBML его не объявляет
ro = metadata.tables["dkb.regional_office"]
Index("ix_regional_office_extra", ro.columns["name"], postgresql_where=text("is_active"))
assert any("индексы" in p for p in find_problems(metadata, {"regional_office"}))
assert find_problems(metadata, {"regional_office"}) == ["regional_office: индекс (name): отсутствует в DBML"]
# пропавший индекс
index.dialect_options["postgresql"]["where"] = text("is_current")
member.indexes.discard(index)
assert find_problems(metadata, {"related_group_member"}) == [f"{label}: отсутствует в ORM"]
def test_nulls_not_distinct_flag_is_compared_both_ways() -> None:
@ -171,12 +197,14 @@ def test_missing_and_extra_foreign_keys_are_detected() -> None:
fk = next(iter(app_user.foreign_keys))
app_user.foreign_keys.remove(fk)
app_user.constraints.discard(fk.constraint) # type: ignore[arg-type]
assert any("FK отсутствует" in p for p in find_problems(metadata, {"app_user"}))
assert find_problems(metadata, {"app_user"}) == [
f"app_user.{fk.parent.name}: FK: ожидалось {fk.column.table.name}.{fk.column.name} / фактически нет"
]
def test_unknown_table_is_reported() -> None:
assert find_problems(Base.metadata, {"no_such_table"}) == ["no_such_table: нет в DBML"]
assert find_problems(MetaData(schema="dkb"), {"regional_office"}) == ["regional_office: нет в ORM-метаданных"]
assert find_problems(Base.metadata, {"no_such_table"}) == ["no_such_table: отсутствует в DBML"]
assert find_problems(MetaData(schema="dkb"), {"regional_office"}) == ["regional_office: отсутствует в ORM"]
@pytest.mark.parametrize(