diff --git a/api/tests/dbml_parity.py b/api/tests/dbml_parity.py new file mode 100644 index 0000000..144a8e4 --- /dev/null +++ b/api/tests/dbml_parity.py @@ -0,0 +1,249 @@ +"""Сверка ORM-метаданных с DBML v5.13 для заданного набора таблиц (без БД); переиспользуется тестами частей 01–09. + +Публичное API: +- `find_problems(metadata, tables)` — список расхождений (пустой = ORM совпадает с DBML) по колонкам (имя, порядок, + тип, nullable, PK, unique, server_default), FK и индексам (состав, уникальность, partial-условие WHERE); +- `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`, число -> как есть, строка -> в одинарных кавычках, выражение -> текст в нижнем регистре. +""" + +import re +from collections.abc import Callable, 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 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) +_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]] = [] + for i in dbml_table.indexes: + if i.pk: + continue + partial = _PARTIAL_RE.search(str(i.note)) if i.note else None + where = _norm_where(partial.group(1)) if partial else None + expected.append((tuple(c.name for c in i.subjects), bool(i.unique) or partial is not None, where)) + actual = [ + (tuple(c.name for c in i.columns), bool(i.unique), _norm_where(i.dialect_options["postgresql"]["where"])) + for i in orm_table.indexes + ] + if sorted(actual, key=repr) == sorted(expected, key=repr): + return [] + return [f"{dbml_table.name}: индексы (колонки, unique, WHERE): ожидалось {sorted(expected, key=repr)}, " + f"фактически {sorted(actual, key=repr)}"] # fmt: skip + + +# --- точка входа ------------------------------------------------------------------------------------------------ + + +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) diff --git a/api/tests/test_dbml_parity.py b/api/tests/test_dbml_parity.py new file mode 100644 index 0000000..1f7fff3 --- /dev/null +++ b/api/tests/test_dbml_parity.py @@ -0,0 +1,187 @@ +"""Самопроверка сверки ORM с DBML (`dbml_parity`): компаратор обязан находить заведомо неверные расхождения.""" + +import pytest +from pydbml.classes import Expression +from sqlalchemy import ( + BigInteger, + Boolean, + Date, + DateTime, + DefaultClause, + Index, + Integer, + MetaData, + Numeric, + SmallInteger, + String, + Text, + func, + text, +) +from sqlalchemy.dialects.postgresql import JSON, JSONB, UUID + +from src.models import Base +from src.models.enums import pg_enum +from tests.dbml_helpers import load_dbml +from tests.dbml_parity import canonical_default_dbml, canonical_default_orm, find_problems, type_problem + +TABLES = {"regional_office", "permission", "related_group_member", "app_user", "elimination_rule", "role_permission"} + + +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 test_clone_is_clean_baseline() -> None: + assert find_problems(clone_metadata(), TABLES) == [] + + +@pytest.mark.parametrize( + ("table", "column", "wrong_type", "needle"), + [ + ("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 + ], +) +def test_wrong_column_type_is_detected(table: str, column: str, wrong_type: object, needle: 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 + + +def test_enum_with_wrong_schema_or_values_is_detected() -> None: + wrong_schema = pg_enum("permission_action") + wrong_schema.schema = "public" + assert type_problem(wrong_schema, _dbml_enum("permission_action")) is not None + wrong_values = pg_enum("permission_action") + wrong_values.enums = list(reversed(wrong_values.enums)) + assert type_problem(wrong_values, _dbml_enum("permission_action")) is not None + assert type_problem(pg_enum("permission_action"), _dbml_enum("permission_action")) is None + + +def _dbml_enum(name: str) -> object: + return next(e for e in load_dbml().enums if e.name == name) + + +@pytest.mark.parametrize( + ("orm_type", "dbml_type", "ok"), + [ + (Numeric(18, 2), "numeric(18,2)", True), + (Numeric(18, 3), "numeric(18,2)", False), + (Numeric(9, 4), "numeric(7,4)", False), + (Integer(), "numeric(18,2)", False), + (JSONB(), "jsonb", True), + (JSON(), "jsonb", False), + (BigInteger(), "bigint", True), + (Integer(), "bigint", False), + (SmallInteger(), "smallint", True), + (Boolean(), "boolean", True), + (Date(), "date", True), + (UUID(as_uuid=True), "uuid", True), + (UUID(as_uuid=False), "uuid", False), + (Text(), "text", True), + (String(), "text", False), + (String(), "varchar", True), + (DateTime(timezone=True), "timestamptz", True), + ], +) +def test_type_mapper(orm_type: object, dbml_type: str, ok: bool) -> None: + assert (type_problem(orm_type, dbml_type) is None) is ok + + +def test_unknown_dbml_type_is_reported_not_ignored() -> None: + assert "не поддержан" in (type_problem(String(), "geometry") or "") + + +def test_missing_wrong_or_extra_server_default_is_detected() -> None: + metadata = clone_metadata() + cols = metadata.tables["dkb.regional_office"].columns + 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 + + +def _default(sql: str) -> object: + return DefaultClause(text(sql)) + + +def test_nullable_pk_unique_and_order_mismatches_are_detected() -> None: + metadata = clone_metadata() + 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 + perm = metadata.tables["dkb.permission"] + perm._columns.remove(perm.columns["name"]) # noqa: SLF001 — намеренная порча копии + assert any("колонки/порядок" in p for p in find_problems(metadata, {"permission"})) + + +def test_missing_extra_and_wrong_where_indexes_are_detected() -> None: + metadata = clone_metadata() + 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"})) + index.dialect_options["postgresql"]["where"] = None # partial превращён в обычный unique + assert any("индексы" in p for p in find_problems(metadata, {"related_group_member"})) + 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"})) + + +def test_missing_and_extra_foreign_keys_are_detected() -> None: + metadata = clone_metadata() + app_user = metadata.tables["dkb.app_user"] + 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"})) + + +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-метаданных"] + + +@pytest.mark.parametrize( + ("dbml_default", "orm_default"), + [ + (False, "false"), + (True, "TRUE"), + (1, "1"), + ("RUB", "'RUB'"), + ("it's", "'it''s'"), + ], +) +def test_default_canonical_forms_agree(dbml_default: object, orm_default: str) -> None: + assert canonical_default_dbml(dbml_default) == canonical_default_orm(DefaultClause(text(orm_default))) + + +def test_now_default_forms_agree() -> None: + expected = canonical_default_dbml(Expression("now()")) + assert expected == canonical_default_orm(DefaultClause(func.now())) == "now()" + assert ( + expected + == canonical_default_orm(DefaultClause(text("now()"))) + == canonical_default_orm(DefaultClause(text("CURRENT_TIMESTAMP"))) + ) + assert expected != canonical_default_dbml(False) diff --git a/api/tests/test_models_part_01_03.py b/api/tests/test_models_part_01_03.py index db205b3..c1723f5 100644 --- a/api/tests/test_models_part_01_03.py +++ b/api/tests/test_models_part_01_03.py @@ -4,7 +4,6 @@ import datetime import uuid import pytest -from pydbml.classes import Table as DbmlTable from sqlalchemy import Table, inspect, text from sqlalchemy.dialects import postgresql from sqlalchemy.exc import IntegrityError @@ -12,7 +11,7 @@ from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine from sqlalchemy.schema import CreateIndex, CreateTable from src.models import Base -from tests.dbml_helpers import load_dbml +from tests.dbml_parity import find_problems PART_01_03_TABLES = { # directories @@ -43,10 +42,6 @@ def orm_tables() -> dict[str, Table]: return {t.name: t for t in Base.metadata.sorted_tables} -def dbml_tables() -> dict[str, DbmlTable]: - return {t.name: t for t in load_dbml().tables if t.name in PART_01_03_TABLES} - - def test_part_01_03_tables_registered_in_metadata() -> None: assert set(orm_tables()) == PART_01_03_TABLES assert len(Base.metadata.sorted_tables) == EXPECTED_TABLE_COUNT # на этом этапе (задачи 7–8 добавят остальные) @@ -77,56 +72,9 @@ def test_constraint_names_are_explicit_and_fit_postgres_limit() -> None: assert len(str(name)) <= PG_IDENTIFIER_LIMIT, f"{table.name}: {name} длиннее {PG_IDENTIFIER_LIMIT}" -def test_columns_match_dbml() -> None: - """Имена колонок, nullability, PK и FK-цели — как в DBML.""" - orm, dbml = orm_tables(), dbml_tables() - refs = [r for r in load_dbml().refs if r.table1.name in PART_01_03_TABLES] - expected_fks = {(r.table1.name, r.col1[0].name, r.table2.name, r.col2[0].name) for r in refs} - actual_fks = { - (t.name, fk.parent.name, fk.column.table.name, fk.column.name) for t in orm.values() for fk in t.foreign_keys - } - assert actual_fks == expected_fks - for name, dbml_table in dbml.items(): - orm_table = orm[name] - dbml_cols = {c.name: c for c in dbml_table.columns} - assert set(orm_table.columns.keys()) == set(dbml_cols), name - 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(): - orm_col = orm_table.columns[col_name] - is_pk = dbml_col.pk or col_name in composite_pk - assert orm_col.primary_key == is_pk, f"{name}.{col_name}: pk" - assert orm_col.nullable == (not (dbml_col.not_null or is_pk)), f"{name}.{col_name}: nullable" - assert bool(orm_col.unique) == bool(dbml_col.unique), f"{name}.{col_name}: unique" - if dbml_col.default is None: - assert orm_col.server_default is None, f"{name}.{col_name}: лишний server_default" - else: - assert orm_col.server_default is not None, f"{name}.{col_name}: нет server_default" - assert str(orm_col.server_default.arg).lower() == str(dbml_col.default).lower() - - -def test_column_order_follows_dbml() -> None: - """Порядок колонок как в DBML (в частности `id` первой) — миграция 0001 читается рядом с DBML.""" - orm = orm_tables() - for name, dbml_table in dbml_tables().items(): - assert [c.name for c in orm[name].columns] == [c.name for c in dbml_table.columns], name - - -def test_indexes_match_dbml() -> None: - """Состав индексов (кроме PK): колонки, уникальность, partial-условие.""" - orm = orm_tables() - for name, dbml_table in dbml_tables().items(): - expected = sorted( - (tuple(c.name for c in i.subjects), i.unique or "partial UNIQUE" in str(i.note)) - for i in dbml_table.indexes - if not i.pk - ) - actual = sorted((tuple(c.name for c in i.columns), bool(i.unique)) for i in orm[name].indexes) - assert actual == expected, name - partial = {i.name: str(i.dialect_options["postgresql"]["where"]) for i in orm["related_group_member"].indexes} - assert partial == { - "uq_related_group_member_group_entity_current": "is_current", - "uq_related_group_member_entity_current": "is_current", - } +def test_orm_matches_dbml() -> None: + """Колонки (имя, порядок, тип, nullable, PK, unique, default), FK и индексы (вкл. partial WHERE) — как в DBML.""" + assert find_problems(Base.metadata, PART_01_03_TABLES) == [] @pytest.mark.integration