7ea1e67f4e
I-1: табличный UniqueConstraint ORM сверяется как уникальный индекс по колонкам (флаг колонки unique=True — по-прежнему в свойствах колонки); любое CHECK в ORM — «отсутствует в DBML». I-2: у совпавших целей FK сравниваются ON DELETE / ON UPDATE (нет в DBML = NO ACTION), DEFERRABLE и INITIALLY. I-3: неизвестный класс типа отрисовывается repr с параметрами (VARCHAR(length=50), Uuid(as_uuid=False)) и никогда не совпадает с именем типа DBML. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
286 lines
14 KiB
Python
286 lines
14 KiB
Python
"""Самопроверка сверки 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),
|
||
(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_current")
|
||
index.dialect_options["postgresql"]["where"] = text("is_current AND valid_to IS NULL")
|
||
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 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.unique)
|
||
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 равен отсутствию действия
|