arm-core/api/tests/test_dbml_parity.py

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