"""Самопроверка сверки 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)