diff --git a/Makefile b/Makefile index bb2840d..34f3984 100644 --- a/Makefile +++ b/Makefile @@ -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 diff --git a/api/scripts/__init__.py b/api/scripts/__init__.py new file mode 100644 index 0000000..dad5dd8 --- /dev/null +++ b/api/scripts/__init__.py @@ -0,0 +1,4 @@ +"""Служебные скрипты arm-core (запуск из `api/`). + +Сверка ORM с DBML — `check_dbml_vs_orm.py` (ядро — `dbml_parity.py`); seed — `python -m scripts.seed`. +""" diff --git a/api/scripts/check_dbml_vs_orm.py b/api/scripts/check_dbml_vs_orm.py new file mode 100644 index 0000000..c8d3837 --- /dev/null +++ b/api/scripts/check_dbml_vs_orm.py @@ -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()) diff --git a/api/scripts/dbml_parity.py b/api/scripts/dbml_parity.py new file mode 100644 index 0000000..f55d400 --- /dev/null +++ b/api/scripts/dbml_parity.py @@ -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 `), `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) diff --git a/api/tests/dbml_helpers.py b/api/tests/dbml_helpers.py index 92216a8..f984333 100644 --- a/api/tests/dbml_helpers.py +++ b/api/tests/dbml_helpers.py @@ -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) diff --git a/api/tests/dbml_parity.py b/api/tests/dbml_parity.py index 02fa7e1..eb19329 100644 --- a/api/tests/dbml_parity.py +++ b/api/tests/dbml_parity.py @@ -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()) diff --git a/api/tests/test_check_dbml_script.py b/api/tests/test_check_dbml_script.py new file mode 100644 index 0000000..3b45f17 --- /dev/null +++ b/api/tests/test_check_dbml_script.py @@ -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] diff --git a/api/tests/test_dbml_parity.py b/api/tests/test_dbml_parity.py index d610a7e..fc4f33a 100644 --- a/api/tests/test_dbml_parity.py +++ b/api/tests/test_dbml_parity.py @@ -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(