"""Самопроверка сверки ORM с DBML (`scripts.dbml_parity`): компаратор обязан находить заведомо неверные расхождения и описывать их строками единого формата `<объект>: ожидалось X / фактически Y` / `<объект>: отсутствует в ORM|DBML`.""" import pytest from pydbml.classes import Expression from sqlalchemy import ( VARCHAR, BigInteger, Boolean, CheckConstraint, Date, DateTime, DefaultClause, Index, Integer, MetaData, Numeric, SmallInteger, String, Text, UniqueConstraint, Uuid, func, text, ) from sqlalchemy.dialects.postgresql import JSON, JSONB, UUID from sqlalchemy.dialects.postgresql import VARCHAR as PG_VARCHAR 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", "expected"), [ ("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, expected: str) -> None: metadata = clone_metadata() metadata.tables[f"dkb.{table}"].columns[column].type = wrong_type # type: ignore[assignment] assert find_problems(metadata, TABLES) == [f"{table}.{column}: ожидалось {expected}"] 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), (JSONB(none_as_null=True), "jsonb", True), # SQL NULL вместо JSON null — DDL тот же (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), # неизвестные/производные классы не должны терять параметры и «проходить» за тип DBML (VARCHAR(50), "varchar", False), (PG_VARCHAR(10), "varchar", False), (Uuid(as_uuid=False), "uuid", False), ], ) 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_orm_type_class_keeps_parameters_in_report() -> None: assert type_problem(VARCHAR(50), "varchar") == "ожидалось varchar / фактически VARCHAR(length=50)" assert type_problem(Uuid(as_uuid=False), "uuid") == "ожидалось uuid / фактически Uuid(as_uuid=False)" 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 нет 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: 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 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"] 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: 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_kind_current") index.dialect_options["postgresql"]["where"] = text("is_current AND valid_to IS NULL") label = "related_group_member: индекс (legal_entity_id, kind)" 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 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 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: """DBML-note `NULLS NOT DISTINCT` требует флаг в ORM-индексе; лишний флаг без note — тоже расхождение.""" metadata = clone_metadata() document = metadata.tables["dkb.document"] index = next(i for i in document.indexes if i.name == "uq_document_file_type_account") assert find_problems(metadata, {"document"}) == [] index.dialect_options["postgresql"]["nulls_not_distinct"] = None # индекс без флага problems = find_problems(metadata, {"document"}) assert any("NULLS NOT DISTINCT" in p for p in problems), problems index.dialect_options["postgresql"]["nulls_not_distinct"] = True assert find_problems(metadata, {"document"}) == [] # лишний флаг на индексе, у которого в DBML нет такого note other = next(i for i in metadata.tables["dkb.s3_object"].indexes if i.unique) other.dialect_options["postgresql"]["nulls_not_distinct"] = True assert any("NULLS NOT DISTINCT" in p for p in find_problems(metadata, {"s3_object"})) 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 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"] @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) def test_table_level_unique_constraint_is_compared_as_unique_index() -> None: """Табличный UniqueConstraint ORM, которого нет в DBML, — расхождение (флаг колонки `unique=True` — не он).""" metadata = clone_metadata() metadata.tables["dkb.regional_office"].append_constraint(UniqueConstraint("name", "is_active")) assert find_problems(metadata, {"regional_office"}) == [ "regional_office: индекс (name, is_active): отсутствует в DBML" ] metadata.tables["dkb.permission"].append_constraint(UniqueConstraint("name")) assert find_problems(metadata, {"permission"}) == ["permission: индекс (name): отсутствует в DBML"] def test_check_constraint_is_reported_missing_in_dbml() -> None: metadata = clone_metadata() metadata.tables["dkb.elimination_rule"].append_constraint(CheckConstraint("rule_no > 0", name="rule_no_positive")) assert find_problems(metadata, {"elimination_rule"}) == ["elimination_rule: CHECK rule_no > 0: отсутствует в DBML"] def test_fk_actions_deferrable_and_initially_are_compared() -> None: """В DBML у ссылок нет действий (= NO ACTION) и DEFERRABLE: лишний CASCADE в ORM обязан всплыть.""" label = "app_user.regional_office_id: FK" metadata = clone_metadata() fk = next(f for f in metadata.tables["dkb.app_user"].foreign_keys if f.parent.name == "regional_office_id") fk.ondelete = "CASCADE" # только на ForeignKey, ограничение не тронуто assert find_problems(metadata, {"app_user"}) == [f"{label} ON DELETE: ожидалось NO ACTION / фактически CASCADE"] fk.ondelete = None constraint = fk.constraint assert constraint is not None constraint.onupdate = "set null" constraint.deferrable = True constraint.initially = "deferred" assert find_problems(metadata, {"app_user"}) == [ f"{label} ON UPDATE: ожидалось NO ACTION / фактически SET NULL", f"{label} DEFERRABLE: ожидалось нет / фактически да", f"{label} INITIALLY: ожидалось нет / фактически DEFERRED", ] constraint.onupdate, constraint.deferrable, constraint.initially = "NO ACTION", None, None assert find_problems(metadata, {"app_user"}) == [] # явный NO ACTION равен отсутствию действия