b4522addc0
- consolidation_version.elimination_rules_version NOT NULL: миграция 0010 проставляет существующим версиям номер правил, действовавших на дату создания (по Москве), иначе последних открытых; без правил — отказ; создание версии без действующих правил — 409 ELIMINATION_RULES_MISSING; ветки «правила не зафиксированы» убраны (rules, preset, is_version_fixed всегда true); - GET …/elimination-totals: все тройки «правило ББ × упорядоченная пара периметра» с ΣA, ΣB, расхождением, суммой элиминации, статусом AUTO/DISCREPANCY/ONE_SIDED/MANUAL и osv_entry_ids; сводная ОСВ строится один раз; сбор строк сторон вынесен в _collect_rule_sides — общий с упаковкой тройки. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
163 lines
7.3 KiB
Python
163 lines
7.3 KiB
Python
"""CLI сверки ORM с DBML (`scripts/check_dbml_vs_orm.py`): сценарии спеки data-model «Сверка ORM с DBML»,
|
||
«Лишняя колонка в ORM», «Расхождение типа или nullability», «Новая версия DBML без обновления ORM».
|
||
|
||
Мутации ORM делаются на копии метаданных (`run_check(..., metadata=...)`), реальный запуск — через subprocess.
|
||
"""
|
||
|
||
import enum
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from sqlalchemy import Column, MetaData, Numeric, String, Table
|
||
|
||
from scripts.check_dbml_vs_orm import EXIT_USAGE, run_check
|
||
from src.models import Base
|
||
from src.models.enums import ENUMS
|
||
from tests.dbml_helpers import dbml_path
|
||
|
||
API_DIR = Path(__file__).resolve().parents[1]
|
||
SCRIPT = "scripts/check_dbml_vs_orm.py"
|
||
|
||
|
||
@pytest.fixture
|
||
def dbml_file() -> Path:
|
||
path = dbml_path()
|
||
if not path.is_file():
|
||
pytest.skip(f"docs недоступен: нет {path} (задайте ARM_KA_DOCS или положите docs рядом с arm-core)")
|
||
return path
|
||
|
||
|
||
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 _run_cli(*args: str) -> subprocess.CompletedProcess[str]:
|
||
"""Запуск скрипта как файла из `api/` без ENV_FILE: скрипт обязан обойтись без рабочего `.env`."""
|
||
env = {k: v for k, v in os.environ.items() if k not in {"ENV_FILE", "APP_ENV"}}
|
||
return subprocess.run(
|
||
[sys.executable, SCRIPT, *args], cwd=API_DIR, env=env, capture_output=True, text=True, check=False
|
||
)
|
||
|
||
|
||
# --- реальный запуск -------------------------------------------------------------------------------------------
|
||
|
||
|
||
def test_cli_on_real_models_prints_nothing_and_exits_0(dbml_file: Path) -> None:
|
||
result = _run_cli(str(dbml_file))
|
||
assert (result.returncode, result.stdout) == (0, ""), result.stdout + result.stderr
|
||
|
||
|
||
def test_cli_missing_file_exits_2_with_message(tmp_path: Path) -> None:
|
||
result = _run_cli(str(tmp_path / "no-such.dbml"))
|
||
assert result.returncode == EXIT_USAGE
|
||
assert result.stdout == ""
|
||
assert "файл DBML не найден" in result.stderr
|
||
|
||
|
||
def test_run_check_on_real_models_is_clean(dbml_file: Path) -> None:
|
||
assert run_check(dbml_file) == (0, [])
|
||
|
||
|
||
# --- сценарий «Лишняя колонка в ORM» ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_extra_orm_column_is_reported_missing_in_dbml(dbml_file: Path) -> None:
|
||
metadata = clone_metadata()
|
||
metadata.tables["dkb.osv_entry"].append_column(Column("extra_note", String()))
|
||
assert run_check(dbml_file, metadata=metadata) == (1, ["osv_entry.extra_note: отсутствует в DBML"])
|
||
|
||
|
||
def test_removed_orm_column_is_reported_missing_in_orm(dbml_file: Path) -> None:
|
||
metadata = clone_metadata()
|
||
osv_entry = metadata.tables["dkb.osv_entry"]
|
||
osv_entry._columns.remove(osv_entry.columns["opening_credit"]) # noqa: SLF001 — намеренная порча копии
|
||
assert run_check(dbml_file, metadata=metadata) == (1, ["osv_entry.opening_credit: отсутствует в ORM"])
|
||
|
||
|
||
# --- сценарий «Расхождение типа или nullability» --------------------------------------------------------------
|
||
|
||
|
||
def test_type_and_nullability_mismatch_is_one_line_in_dbml_notation(dbml_file: Path) -> None:
|
||
metadata = clone_metadata()
|
||
column = metadata.tables["dkb.osv_entry"].columns["opening_debit"]
|
||
column.type = Numeric(18, 4)
|
||
column.nullable = False
|
||
assert run_check(dbml_file, metadata=metadata) == (
|
||
1,
|
||
["osv_entry.opening_debit: ожидалось numeric(18,2) NULL / фактически numeric(18,4) NOT NULL"],
|
||
)
|
||
|
||
|
||
def test_nullability_only_mismatch(dbml_file: Path) -> None:
|
||
metadata = clone_metadata()
|
||
metadata.tables["dkb.osv_entry"].columns["opening_debit"].nullable = False
|
||
code, lines = run_check(dbml_file, metadata=metadata)
|
||
assert (code, lines) == (
|
||
1,
|
||
["osv_entry.opening_debit: ожидалось numeric(18,2) NULL / фактически numeric(18,2) NOT NULL"],
|
||
)
|
||
|
||
|
||
# --- таблицы и enum'ы: сравнение множеств ----------------------------------------------------------------------
|
||
|
||
|
||
def test_missing_and_extra_tables_are_reported(dbml_file: Path) -> None:
|
||
metadata = clone_metadata()
|
||
metadata.remove(metadata.tables["dkb.consolidation_lock"])
|
||
code, lines = run_check(dbml_file, metadata=metadata)
|
||
assert code == 1
|
||
assert "consolidation_lock: отсутствует в ORM" in lines
|
||
metadata = clone_metadata()
|
||
Table("orphan_table", metadata, Column("id", String(), primary_key=True))
|
||
assert run_check(dbml_file, metadata=metadata) == (1, ["orphan_table: отсутствует в DBML"])
|
||
|
||
|
||
def test_enum_values_and_sets_are_compared(dbml_file: Path) -> None:
|
||
enums: dict[str, type[enum.Enum]] = dict(ENUMS)
|
||
enums["entity_type"] = enum.Enum("EntityType", {"ИП": "ИП", "ЮЛ": "ЮЛ"}) # порядок значений нарушен
|
||
del enums["audit_section"]
|
||
enums["no_such_enum"] = enum.Enum("NoSuch", {"A": "A"})
|
||
code, lines = run_check(dbml_file, enums=enums)
|
||
assert code == 1
|
||
assert sorted(lines) == sorted(
|
||
[
|
||
"enum entity_type: ожидалось ['ЮЛ', 'ИП'] / фактически ['ИП', 'ЮЛ']",
|
||
"enum audit_section: отсутствует в ORM",
|
||
"enum no_such_enum: отсутствует в DBML",
|
||
]
|
||
)
|
||
|
||
|
||
# --- сценарий «Новая версия DBML без обновления ORM» -----------------------------------------------------------
|
||
|
||
|
||
def test_newer_dbml_version_fails_before_comparing_tables(tmp_path: Path, dbml_file: Path) -> None:
|
||
lines = dbml_file.read_text(encoding="utf-8").splitlines()
|
||
newer = tmp_path / "data-model-v5.24.dbml"
|
||
# шапка 5.23 + заведомо сломанное тело: если бы таблицы сравнивались, были бы другие строки/ошибка разбора
|
||
newer.write_text(lines[0].replace("v5.23", "v5.24") + "\nTable broken {{{\n", encoding="utf-8")
|
||
assert run_check(newer) == (1, ["версия DBML 5.24 не совпадает с версией ORM 5.23: обновите модели"])
|
||
|
||
|
||
def test_newer_dbml_version_via_cli(tmp_path: Path, dbml_file: Path) -> None:
|
||
newer = tmp_path / "data-model-v5.24.dbml"
|
||
newer.write_text(dbml_file.read_text(encoding="utf-8").replace("v5.23", "v5.24", 1), encoding="utf-8")
|
||
result = _run_cli(str(newer))
|
||
assert result.returncode == 1
|
||
assert result.stdout == "версия DBML 5.24 не совпадает с версией ORM 5.23: обновите модели\n"
|
||
|
||
|
||
def test_header_without_version_is_usage_error(tmp_path: Path) -> None:
|
||
bad = tmp_path / "model.dbml"
|
||
bad.write_text("// модель без версии\nTable t {\n id uuid [pk]\n}\n", encoding="utf-8")
|
||
code, lines = run_check(bad)
|
||
assert code == EXIT_USAGE
|
||
assert len(lines) == 1 and "версия" in lines[0]
|