test: чётность ORM частей 01–03 с DBML, create_all на временной БД, фикстура scratch_db_url
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
fb2b9ef9fc
commit
56c98a5d05
@ -0,0 +1,3 @@
|
||||
"""Общие фикстуры тестов."""
|
||||
|
||||
pytest_plugins = ["tests.pg_helpers"]
|
||||
54
api/tests/pg_helpers.py
Normal file
54
api/tests/pg_helpers.py
Normal file
@ -0,0 +1,54 @@
|
||||
"""Помощники интеграционных тестов с PostgreSQL: временная БД внутри того же сервера, что и `OPENBAO__DATABASE`.
|
||||
|
||||
Фикстура `scratch_db_url` создаёт пустую БД `armcore_scratch_<hex>`, отдаёт её SQLAlchemy-URL (драйвер asyncpg) и
|
||||
в конце удаляет. Рабочую БД (`armka`) тесты не трогают. Подключается плагином в `conftest.py`.
|
||||
"""
|
||||
|
||||
import uuid
|
||||
from collections.abc import AsyncIterator
|
||||
|
||||
import asyncpg
|
||||
import pytest_asyncio
|
||||
from sqlalchemy.engine import URL, make_url
|
||||
|
||||
from src.core.config import settings
|
||||
|
||||
|
||||
def base_url() -> URL:
|
||||
"""SQLAlchemy-URL рабочей БД из настроек (`.env.test`)."""
|
||||
return make_url(settings.OPENBAO.DATABASE.DATABASE_URL)
|
||||
|
||||
|
||||
def _asyncpg_dsn(url: URL) -> str:
|
||||
return url.set(drivername="postgresql").render_as_string(hide_password=False)
|
||||
|
||||
|
||||
async def create_scratch_db(name: str) -> URL:
|
||||
"""Создать пустую БД `name` на сервере рабочей БД; вернуть SQLAlchemy-URL на неё."""
|
||||
url = base_url()
|
||||
conn = await asyncpg.connect(_asyncpg_dsn(url))
|
||||
try:
|
||||
await conn.execute(f'CREATE DATABASE "{name}"')
|
||||
finally:
|
||||
await conn.close()
|
||||
return url.set(database=name)
|
||||
|
||||
|
||||
async def drop_scratch_db(name: str) -> None:
|
||||
"""Удалить БД `name` (закрывая оставшиеся подключения)."""
|
||||
conn = await asyncpg.connect(_asyncpg_dsn(base_url()))
|
||||
try:
|
||||
await conn.execute(f'DROP DATABASE IF EXISTS "{name}" WITH (FORCE)')
|
||||
finally:
|
||||
await conn.close()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def scratch_db_url() -> AsyncIterator[str]:
|
||||
"""URL пустой временной БД (asyncpg); БД удаляется после теста."""
|
||||
name = f"armcore_scratch_{uuid.uuid4().hex[:12]}"
|
||||
url = await create_scratch_db(name)
|
||||
try:
|
||||
yield url.render_as_string(hide_password=False)
|
||||
finally:
|
||||
await drop_scratch_db(name)
|
||||
210
api/tests/test_models_part_01_03.py
Normal file
210
api/tests/test_models_part_01_03.py
Normal file
@ -0,0 +1,210 @@
|
||||
"""ORM частей 01–03 DBML v5.13 (справочники, доступ, учёт): чётность с DBML без БД и create_all на PostgreSQL."""
|
||||
|
||||
import datetime
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from pydbml.classes import Table as DbmlTable
|
||||
from sqlalchemy import Table, inspect, text
|
||||
from sqlalchemy.dialects import postgresql
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy.ext.asyncio import create_async_engine
|
||||
from sqlalchemy.schema import CreateIndex, CreateTable
|
||||
|
||||
from src.models import Base
|
||||
from tests.dbml_helpers import load_dbml
|
||||
|
||||
PART_01_03_TABLES = {
|
||||
# directories
|
||||
"regional_office",
|
||||
"company_group",
|
||||
"company_group_history",
|
||||
"legal_entity",
|
||||
"legal_entity_history",
|
||||
"legal_entity_alias",
|
||||
"related_group",
|
||||
"related_group_member",
|
||||
# access
|
||||
"app_user",
|
||||
"role",
|
||||
"permission",
|
||||
"role_permission",
|
||||
"user_role",
|
||||
# accounting
|
||||
"chart_of_account",
|
||||
"elimination_rule",
|
||||
"elimination_rule_account",
|
||||
}
|
||||
PG_IDENTIFIER_LIMIT = 63
|
||||
EXPECTED_TABLE_COUNT = 16
|
||||
|
||||
|
||||
def orm_tables() -> dict[str, Table]:
|
||||
return {t.name: t for t in Base.metadata.sorted_tables}
|
||||
|
||||
|
||||
def dbml_tables() -> dict[str, DbmlTable]:
|
||||
return {t.name: t for t in load_dbml().tables if t.name in PART_01_03_TABLES}
|
||||
|
||||
|
||||
def test_part_01_03_tables_registered_in_metadata() -> None:
|
||||
assert set(orm_tables()) == PART_01_03_TABLES
|
||||
assert len(Base.metadata.sorted_tables) == EXPECTED_TABLE_COUNT # на этом этапе (задачи 7–8 добавят остальные)
|
||||
assert all(t.schema == "dkb" for t in Base.metadata.sorted_tables)
|
||||
|
||||
|
||||
def test_every_table_and_index_compiles_for_postgresql() -> None:
|
||||
dialect = postgresql.dialect()
|
||||
for table in Base.metadata.sorted_tables:
|
||||
assert str(CreateTable(table).compile(dialect=dialect)).strip()
|
||||
for index in table.indexes:
|
||||
assert str(CreateIndex(index).compile(dialect=dialect)).strip()
|
||||
|
||||
|
||||
def test_foreign_keys_resolve_to_registered_tables() -> None:
|
||||
for table in Base.metadata.sorted_tables:
|
||||
for fk in table.foreign_keys:
|
||||
assert fk.column.table.key in Base.metadata.tables, f"{table.name}.{fk.parent.name}"
|
||||
assert fk.column.table.schema == "dkb"
|
||||
|
||||
|
||||
def test_constraint_names_are_explicit_and_fit_postgres_limit() -> None:
|
||||
"""Все имена ограничений заданы (конвенцией/явно), без хеш-усечения, и не длиннее 63 символов."""
|
||||
for table in Base.metadata.sorted_tables:
|
||||
names = [c.name for c in table.constraints] + [i.name for i in table.indexes]
|
||||
for name in names:
|
||||
assert name, f"{table.name}: безымянное ограничение"
|
||||
assert len(str(name)) <= PG_IDENTIFIER_LIMIT, f"{table.name}: {name} длиннее {PG_IDENTIFIER_LIMIT}"
|
||||
|
||||
|
||||
def test_columns_match_dbml() -> None:
|
||||
"""Имена колонок, nullability, PK и FK-цели — как в DBML."""
|
||||
orm, dbml = orm_tables(), dbml_tables()
|
||||
refs = [r for r in load_dbml().refs if r.table1.name in PART_01_03_TABLES]
|
||||
expected_fks = {(r.table1.name, r.col1[0].name, r.table2.name, r.col2[0].name) for r in refs}
|
||||
actual_fks = {
|
||||
(t.name, fk.parent.name, fk.column.table.name, fk.column.name) for t in orm.values() for fk in t.foreign_keys
|
||||
}
|
||||
assert actual_fks == expected_fks
|
||||
for name, dbml_table in dbml.items():
|
||||
orm_table = orm[name]
|
||||
dbml_cols = {c.name: c for c in dbml_table.columns}
|
||||
assert set(orm_table.columns.keys()) == set(dbml_cols), name
|
||||
composite_pk = {c.name for i in dbml_table.indexes if i.pk for c in i.subjects}
|
||||
for col_name, dbml_col in dbml_cols.items():
|
||||
orm_col = orm_table.columns[col_name]
|
||||
is_pk = dbml_col.pk or col_name in composite_pk
|
||||
assert orm_col.primary_key == is_pk, f"{name}.{col_name}: pk"
|
||||
assert orm_col.nullable == (not (dbml_col.not_null or is_pk)), f"{name}.{col_name}: nullable"
|
||||
assert bool(orm_col.unique) == bool(dbml_col.unique), f"{name}.{col_name}: unique"
|
||||
if dbml_col.default is None:
|
||||
assert orm_col.server_default is None, f"{name}.{col_name}: лишний server_default"
|
||||
else:
|
||||
assert orm_col.server_default is not None, f"{name}.{col_name}: нет server_default"
|
||||
assert str(orm_col.server_default.arg).lower() == str(dbml_col.default).lower()
|
||||
|
||||
|
||||
def test_column_order_follows_dbml() -> None:
|
||||
"""Порядок колонок как в DBML (в частности `id` первой) — миграция 0001 читается рядом с DBML."""
|
||||
orm = orm_tables()
|
||||
for name, dbml_table in dbml_tables().items():
|
||||
assert [c.name for c in orm[name].columns] == [c.name for c in dbml_table.columns], name
|
||||
|
||||
|
||||
def test_indexes_match_dbml() -> None:
|
||||
"""Состав индексов (кроме PK): колонки, уникальность, partial-условие."""
|
||||
orm = orm_tables()
|
||||
for name, dbml_table in dbml_tables().items():
|
||||
expected = sorted(
|
||||
(tuple(c.name for c in i.subjects), i.unique or "partial UNIQUE" in str(i.note))
|
||||
for i in dbml_table.indexes
|
||||
if not i.pk
|
||||
)
|
||||
actual = sorted((tuple(c.name for c in i.columns), bool(i.unique)) for i in orm[name].indexes)
|
||||
assert actual == expected, name
|
||||
partial = {i.name: str(i.dialect_options["postgresql"]["where"]) for i in orm["related_group_member"].indexes}
|
||||
assert partial == {
|
||||
"uq_related_group_member_group_entity_current": "is_current",
|
||||
"uq_related_group_member_entity_current": "is_current",
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_create_all_on_scratch_db(scratch_db_url: str) -> None:
|
||||
"""`create_all` на пустой БД: 16 таблиц в схеме dkb, enum-типы, partial unique работают."""
|
||||
engine = create_async_engine(scratch_db_url)
|
||||
try:
|
||||
async with engine.begin() as conn:
|
||||
await conn.execute(text("CREATE SCHEMA IF NOT EXISTS dkb"))
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
|
||||
async with engine.connect() as conn:
|
||||
tables = await conn.run_sync(lambda c: set(inspect(c).get_table_names(schema="dkb")))
|
||||
assert tables == PART_01_03_TABLES
|
||||
enums = {
|
||||
row[0]
|
||||
for row in await conn.execute(
|
||||
text(
|
||||
"SELECT t.typname FROM pg_type t JOIN pg_namespace n ON n.oid = t.typnamespace "
|
||||
"WHERE n.nspname = 'dkb' AND t.typtype = 'e'"
|
||||
)
|
||||
)
|
||||
}
|
||||
assert enums == {
|
||||
"entity_type", "related_group_kind", "account_kind", "account_row_type", "business_segment",
|
||||
"statement_type", "le_history_event", "permission_object", "permission_action",
|
||||
} # fmt: skip
|
||||
indexdefs = {
|
||||
row[0]: row[1]
|
||||
for row in await conn.execute(
|
||||
text(
|
||||
"SELECT indexname, indexdef FROM pg_indexes WHERE schemaname = 'dkb' "
|
||||
"AND tablename = 'related_group_member'"
|
||||
)
|
||||
)
|
||||
}
|
||||
for name in ("uq_related_group_member_group_entity_current", "uq_related_group_member_entity_current"):
|
||||
assert "UNIQUE" in indexdefs[name]
|
||||
assert "WHERE is_current" in indexdefs[name]
|
||||
pk = await conn.run_sync(lambda c: inspect(c).get_pk_constraint("role_permission", schema="dkb"))
|
||||
assert pk["name"] == "pk_role_permission"
|
||||
assert pk["constrained_columns"] == ["role_id", "permission_id"]
|
||||
|
||||
await _check_related_group_member_uniqueness(engine)
|
||||
finally:
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
async def _check_related_group_member_uniqueness(engine) -> None: # type: ignore[no-untyped-def]
|
||||
"""ЮЛ — только в одной ГСК/ГСЗ среди текущих; закрытая версия не мешает (ТЗ 6.1, ANL-030)."""
|
||||
now = datetime.datetime.now(datetime.UTC)
|
||||
ids = {k: uuid.uuid4() for k in ("ro", "cg", "le", "g1", "g2", "m1", "m2", "m3")}
|
||||
async with engine.begin() as conn:
|
||||
await conn.execute(text("INSERT INTO dkb.regional_office (id, name) VALUES (:ro, 'ГО')"), ids)
|
||||
await conn.execute(
|
||||
text("INSERT INTO dkb.company_group (id, code, name, valid_from) VALUES (:cg, 'G1', 'ГК', :now)"),
|
||||
{**ids, "now": now},
|
||||
)
|
||||
await conn.execute(
|
||||
text("INSERT INTO dkb.legal_entity (id, inn, name, valid_from) VALUES (:le, '1', 'ЮЛ', :now)"),
|
||||
{**ids, "now": now},
|
||||
)
|
||||
for g, code in (("g1", "GSK-1"), ("g2", "GSK-2")):
|
||||
await conn.execute(
|
||||
text(
|
||||
"INSERT INTO dkb.related_group (id, code, company_group_id, kind, name) "
|
||||
"VALUES (:g, :code, :cg, 'gsk', 'ГСК')"
|
||||
),
|
||||
{"g": ids[g], "code": code, "cg": ids["cg"]},
|
||||
)
|
||||
insert = text(
|
||||
"INSERT INTO dkb.related_group_member (id, related_group_id, legal_entity_id, valid_from, is_current) "
|
||||
"VALUES (:id, :g, :le, :now, :cur)"
|
||||
)
|
||||
async with engine.begin() as conn:
|
||||
await conn.execute(insert, {"id": ids["m1"], "g": ids["g1"], "le": ids["le"], "now": now, "cur": True})
|
||||
# закрытая версия в другой группе — допустима
|
||||
await conn.execute(insert, {"id": ids["m2"], "g": ids["g2"], "le": ids["le"], "now": now, "cur": False})
|
||||
with pytest.raises(IntegrityError):
|
||||
async with engine.begin() as conn: # второе текущее членство того же ЮЛ — нарушение
|
||||
await conn.execute(insert, {"id": ids["m3"], "g": ids["g2"], "le": ids["le"], "now": now, "cur": True})
|
||||
Loading…
x
Reference in New Issue
Block a user