1369 lines
63 KiB
Python
1369 lines
63 KiB
Python
"""Миграции 0001 (DBML v5.13), 0002 (v5.14), 0003 (v5.15), 0004 (v5.16), 0005 (v5.17), 0006 (v5.18), 0007 (v5.19)
|
||
на PostgreSQL 16:
|
||
полный цикл, откат на шаг, пустой autogenerate, схема = схеме ORM.
|
||
|
||
Spec data-model, требование «Миграции применяются с нуля и откатываются» (сценарии «Полный цикл» и «Автогенерация
|
||
ничего не находит»), design D-2. Alembic запускается отдельным процессом (`python -m alembic -x dburl=...`), как из
|
||
CLI: `env.py` вызывает `asyncio.run(...)`, а внутри event loop pytest-asyncio это невозможно; заодно процесс не
|
||
перенастраивает логирование тестов через `fileConfig`. Каждая проверка идёт на своей временной БД (`scratch_db_url`).
|
||
"""
|
||
|
||
import datetime
|
||
import uuid
|
||
from typing import Any
|
||
|
||
import pytest
|
||
from alembic.autogenerate import compare_metadata
|
||
from alembic.runtime.migration import MigrationContext
|
||
from sqlalchemy import Connection, make_url, text
|
||
from sqlalchemy.exc import IntegrityError, ProgrammingError
|
||
from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine, AsyncSession, create_async_engine
|
||
|
||
from src.models import Base
|
||
from src.models.enums import ENUMS, OsvDataType
|
||
from tests.pg_helpers import alembic_async as alembic
|
||
from tests.pg_helpers import base_url, create_scratch_db, drop_scratch_db, run_alembic, service_url
|
||
from tests.test_models_part_01_03 import _check_related_group_member_uniqueness
|
||
from tests.test_models_part_04_06 import (
|
||
_add_osv_entry,
|
||
_check_nu_entry_inserts,
|
||
_check_nulls_not_distinct,
|
||
_make_file_with_job,
|
||
)
|
||
from tests.test_models_part_07_09 import (
|
||
ALL_TABLES,
|
||
EXPECTED_ENUM_COUNT,
|
||
EXPECTED_TABLE_COUNT,
|
||
_check_active_lock_is_unique,
|
||
)
|
||
|
||
V5_19 = "0007_dbml_v5_19"
|
||
V5_20 = "0008_dbml_v5_20"
|
||
HEAD = "0009_dbml_v5_21"
|
||
V5_18 = "0006_dbml_v5_18"
|
||
V5_17 = "0005_dbml_v5_17"
|
||
V5_16 = "0004_dbml_v5_16"
|
||
V5_15 = "0003_dbml_v5_15"
|
||
V5_14 = "0002_dbml_v5_14"
|
||
PREVIOUS = "0001_dbml_v5_13"
|
||
VERSION_TABLE = "alembic_version"
|
||
FK_CYCLE = "fk_stored_file_current_recognition_job_id_recognition_job"
|
||
# v5.15 п.4 (I-1, OQ-117): индексы для проверок FK при удалении файла/документа и обхода дерева строк ОСВ
|
||
V5_15_INDEXES = {
|
||
"ix_osv_entry_parent_entry_id": ("osv_entry", "parent_entry_id"),
|
||
"ix_osv_entry_recognition_job_id": ("osv_entry", "recognition_job_id"),
|
||
"ix_stored_file_current_recognition_job_id": ("stored_file", "current_recognition_job_id"),
|
||
}
|
||
|
||
|
||
async def _scalars(conn: AsyncConnection, sql: str) -> list[Any]:
|
||
return list((await conn.execute(text(sql))).scalars())
|
||
|
||
|
||
async def tables_in_dkb(conn: AsyncConnection) -> set[str]:
|
||
return set(
|
||
await _scalars(
|
||
conn,
|
||
"SELECT table_name FROM information_schema.tables WHERE table_schema = 'dkb'",
|
||
)
|
||
)
|
||
|
||
|
||
async def enums_in_dkb(conn: AsyncConnection) -> set[str]:
|
||
return set(
|
||
await _scalars(
|
||
conn,
|
||
"SELECT t.typname FROM pg_type t JOIN pg_namespace n ON n.oid = t.typnamespace "
|
||
"WHERE n.nspname = 'dkb' AND t.typtype = 'e'",
|
||
)
|
||
)
|
||
|
||
|
||
async def schema_snapshot(conn: AsyncConnection) -> dict[str, Any]:
|
||
"""Каталог схемы dkb без служебной таблицы alembic: колонки, ограничения, индексы, enum'ы (с порядком значений)."""
|
||
columns = (
|
||
await conn.execute(
|
||
text(
|
||
"SELECT table_name, ordinal_position, column_name, data_type, udt_schema, udt_name, is_nullable, "
|
||
"column_default, numeric_precision, numeric_scale FROM information_schema.columns "
|
||
"WHERE table_schema = 'dkb' AND table_name <> :vt ORDER BY table_name, ordinal_position"
|
||
),
|
||
{"vt": VERSION_TABLE},
|
||
)
|
||
).all()
|
||
constraints = (
|
||
await conn.execute(
|
||
text(
|
||
"SELECT c.conname, c.conrelid::regclass::text, pg_get_constraintdef(c.oid) FROM pg_constraint c "
|
||
"WHERE c.connamespace = 'dkb'::regnamespace AND c.conrelid::regclass::text <> 'dkb.alembic_version' "
|
||
"AND c.contype <> 'n'"
|
||
)
|
||
)
|
||
).all()
|
||
indexes = (
|
||
await conn.execute(
|
||
text("SELECT indexname, indexdef FROM pg_indexes WHERE schemaname = 'dkb' AND tablename <> :vt"),
|
||
{"vt": VERSION_TABLE},
|
||
)
|
||
).all()
|
||
enums = (
|
||
await conn.execute(
|
||
text(
|
||
"SELECT t.typname, array_agg(e.enumlabel ORDER BY e.enumsortorder) FROM pg_type t "
|
||
"JOIN pg_namespace n ON n.oid = t.typnamespace JOIN pg_enum e ON e.enumtypid = t.oid "
|
||
"WHERE n.nspname = 'dkb' GROUP BY t.typname"
|
||
)
|
||
)
|
||
).all()
|
||
return {
|
||
"columns": [tuple(row) for row in columns],
|
||
"constraints": {row[0]: (row[1], row[2]) for row in constraints},
|
||
"indexes": {row[0]: row[1] for row in indexes},
|
||
"enums": {row[0]: list(row[1]) for row in enums},
|
||
}
|
||
|
||
|
||
async def _assert_full_model(conn: AsyncConnection) -> None:
|
||
tables = await tables_in_dkb(conn)
|
||
assert tables - {VERSION_TABLE} == ALL_TABLES
|
||
assert len(tables - {VERSION_TABLE}) == EXPECTED_TABLE_COUNT
|
||
enums = await enums_in_dkb(conn)
|
||
assert enums == set(ENUMS)
|
||
assert len(enums) == EXPECTED_ENUM_COUNT
|
||
assert await _scalars(conn, "SELECT version_num FROM dkb.alembic_version") == [HEAD]
|
||
|
||
|
||
async def _insert_user(conn: AsyncConnection, office_id: uuid.UUID, email: str | None) -> uuid.UUID:
|
||
"""Сырой INSERT пользователя (на ревизии 0002 ORM v5.15 не подходит); NOT NULL — `full_name`, филиал."""
|
||
user_id = uuid.uuid4()
|
||
await conn.execute(
|
||
text("INSERT INTO dkb.app_user (id, email, full_name, regional_office_id) VALUES (:id, :email, 'Тест', :ro)"),
|
||
{"id": user_id, "email": email, "ro": office_id},
|
||
)
|
||
return user_id
|
||
|
||
|
||
async def _insert_office(conn: AsyncConnection) -> uuid.UUID:
|
||
office_id = uuid.uuid4()
|
||
await conn.execute(text("INSERT INTO dkb.regional_office (id, name) VALUES (:id, 'ГО')"), {"id": office_id})
|
||
return office_id
|
||
|
||
|
||
async def _truncate_model_tables(engine: AsyncEngine) -> None:
|
||
async with engine.begin() as conn:
|
||
await conn.execute(text("TRUNCATE " + ", ".join(f"dkb.{table}" for table in sorted(ALL_TABLES)) + " CASCADE"))
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_full_cycle_upgrade_downgrade_upgrade(scratch_db_url: str) -> None:
|
||
"""Сценарий «Полный цикл»: upgrade head → downgrade base → upgrade head; после отката схема dkb пуста."""
|
||
engine = create_async_engine(scratch_db_url)
|
||
try:
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
async with engine.connect() as conn:
|
||
await _assert_full_model(conn)
|
||
|
||
await alembic(scratch_db_url, "downgrade", "base")
|
||
async with engine.connect() as conn:
|
||
assert await tables_in_dkb(conn) == {VERSION_TABLE} # служебная таблица alembic остаётся, она пуста
|
||
assert await _scalars(conn, "SELECT count(*) FROM dkb.alembic_version") == [0]
|
||
assert await enums_in_dkb(conn) == set()
|
||
# в схеме не осталось других объектов (последовательностей, типов, индексов) кроме alembic_version
|
||
leftovers = await _scalars(
|
||
conn,
|
||
"SELECT relname FROM pg_class WHERE relnamespace = 'dkb'::regnamespace "
|
||
"AND relname NOT IN ('alembic_version', 'alembic_version_pkc')",
|
||
)
|
||
assert leftovers == []
|
||
types = await _scalars(
|
||
conn,
|
||
"SELECT typname FROM pg_type WHERE typnamespace = 'dkb'::regnamespace "
|
||
"AND typname NOT IN ('alembic_version', '_alembic_version')",
|
||
)
|
||
assert types == []
|
||
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
async with engine.connect() as conn:
|
||
await _assert_full_model(conn)
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_downgrade_to_v5_13(scratch_db_url: str) -> None:
|
||
"""Откат head → 0001 (два шага) возвращает схему v5.13 (без kind, старый индекс и audit_action)."""
|
||
engine = create_async_engine(scratch_db_url)
|
||
try:
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
await alembic(scratch_db_url, "downgrade", PREVIOUS)
|
||
async with engine.connect() as conn:
|
||
assert await _scalars(conn, "SELECT version_num FROM dkb.alembic_version") == [PREVIOUS]
|
||
kind_columns = await _scalars(
|
||
conn,
|
||
"SELECT column_name FROM information_schema.columns WHERE table_schema = 'dkb' "
|
||
"AND table_name = 'related_group_member' AND column_name = 'kind'",
|
||
)
|
||
assert kind_columns == []
|
||
indexes = set(
|
||
await _scalars(
|
||
conn,
|
||
"SELECT indexname FROM pg_indexes WHERE schemaname = 'dkb' AND tablename = 'related_group_member'",
|
||
)
|
||
)
|
||
assert "uq_related_group_member_entity_current" in indexes
|
||
assert "uq_related_group_member_entity_kind_current" not in indexes
|
||
[labels] = await _scalars(conn, "SELECT enum_range(NULL::dkb.audit_action)::text[]")
|
||
assert "LEGAL_ENTITY_CHANGE" not in labels
|
||
assert "USER_CHANGE" not in labels
|
||
sequences = await _scalars(
|
||
conn, "SELECT sequence_name FROM information_schema.sequences WHERE sequence_schema = 'dkb'"
|
||
)
|
||
assert sequences == []
|
||
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
async with engine.connect() as conn:
|
||
await _assert_full_model(conn)
|
||
[labels] = await _scalars(conn, "SELECT enum_range(NULL::dkb.audit_action)::text[]")
|
||
assert labels == [m.value for m in ENUMS["audit_action"]]
|
||
sequences = set(
|
||
await _scalars(
|
||
conn, "SELECT sequence_name FROM information_schema.sequences WHERE sequence_schema = 'dkb'"
|
||
)
|
||
)
|
||
assert sequences == {"company_group_code_seq", "related_group_code_seq"}
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_downgrade_one_step(scratch_db_url: str) -> None:
|
||
"""Сценарий «Откат на один шаг»: head → 0002 убирает `NU` из `osv_data_type`; регистр email не возвращается."""
|
||
engine = create_async_engine(scratch_db_url)
|
||
try:
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
async with engine.begin() as conn:
|
||
[labels] = await _scalars(conn, "SELECT enum_range(NULL::dkb.osv_data_type)::text[]")
|
||
assert labels == ["BU", "NU", "CURRENCY", "QUANTITY", "SUM"]
|
||
await _insert_user(conn, await _insert_office(conn), "lower@x.ru")
|
||
await engine.dispose() # пул asyncpg кэширует OID типов; downgrade пересоздаёт enum
|
||
|
||
await alembic(scratch_db_url, "downgrade", V5_14)
|
||
async with engine.connect() as conn:
|
||
assert await _scalars(conn, "SELECT version_num FROM dkb.alembic_version") == [V5_14]
|
||
[labels] = await _scalars(conn, "SELECT enum_range(NULL::dkb.osv_data_type)::text[]")
|
||
assert labels == ["BU", "CURRENCY", "QUANTITY", "SUM"]
|
||
assert await _scalars(conn, "SELECT email FROM dkb.app_user") == ["lower@x.ru"]
|
||
# колонка осталась типизированной новым типом: старые значения читаются, остальные схемы не сломаны
|
||
assert await _scalars(
|
||
conn,
|
||
"SELECT udt_name FROM information_schema.columns WHERE table_schema = 'dkb' "
|
||
"AND table_name = 'osv_entry' AND column_name = 'data_type'",
|
||
) == ["osv_data_type"]
|
||
await engine.dispose()
|
||
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
async with engine.connect() as conn:
|
||
await _assert_full_model(conn)
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_downgrade_one_step_fails_while_nu_rows_exist(scratch_db_url: str) -> None:
|
||
"""Строка `osv_entry` с `data_type = 'NU'` делает откат невозможным (осознанно, как USING в 0002)."""
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
engine = create_async_engine(scratch_db_url)
|
||
try:
|
||
async with AsyncSession(engine, expire_on_commit=False) as session:
|
||
chain = await _make_file_with_job(session)
|
||
await _add_osv_entry(session, chain, OsvDataType.NU, "91")
|
||
await session.commit()
|
||
with pytest.raises(AssertionError):
|
||
await alembic(scratch_db_url, "downgrade", V5_14)
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
async def _v5_15_indexes(conn: AsyncConnection) -> dict[str, str]:
|
||
rows = (
|
||
await conn.execute(
|
||
text("SELECT indexname, indexdef FROM pg_indexes WHERE schemaname = 'dkb' AND indexname = ANY(:names)"),
|
||
{"names": list(V5_15_INDEXES)},
|
||
)
|
||
).all()
|
||
return {row[0]: row[1] for row in rows}
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_upgrade_0003_creates_fk_lookup_indexes(scratch_db_url: str) -> None:
|
||
"""0003 создаёт обычные (не unique) btree-индексы v5.15 п.4; откат на 0002 удаляет, повторный upgrade возвращает."""
|
||
engine = create_async_engine(scratch_db_url)
|
||
try:
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
async with engine.connect() as conn:
|
||
indexes = await _v5_15_indexes(conn)
|
||
assert set(indexes) == set(V5_15_INDEXES)
|
||
for name, (table, column) in V5_15_INDEXES.items():
|
||
assert indexes[name] == f"CREATE INDEX {name} ON dkb.{table} USING btree ({column})"
|
||
await engine.dispose()
|
||
|
||
await alembic(scratch_db_url, "downgrade", V5_14)
|
||
async with engine.connect() as conn:
|
||
assert await _v5_15_indexes(conn) == {}
|
||
await engine.dispose()
|
||
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
async with engine.connect() as conn:
|
||
assert set(await _v5_15_indexes(conn)) == set(V5_15_INDEXES)
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_upgrade_0003_lowercases_existing_emails(scratch_db_url: str) -> None:
|
||
"""0003 приводит `app_user.email` к нижнему регистру; уже строчные и NULL не затрагиваются."""
|
||
engine = create_async_engine(scratch_db_url)
|
||
try:
|
||
await alembic(scratch_db_url, "upgrade", V5_14)
|
||
async with engine.begin() as conn:
|
||
office_id = await _insert_office(conn)
|
||
ids = [await _insert_user(conn, office_id, email) for email in ("Email@X.RU", "already@x.ru", None)]
|
||
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
async with engine.connect() as conn:
|
||
rows = (await conn.execute(text("SELECT id, email FROM dkb.app_user"))).all()
|
||
assert {tuple(r) for r in rows} == {(ids[0], "email@x.ru"), (ids[1], "already@x.ru"), (ids[2], None)}
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
async def _kind_column(conn: AsyncConnection) -> list[Any]:
|
||
return await _scalars(
|
||
conn,
|
||
"SELECT column_name FROM information_schema.columns WHERE table_schema = 'dkb' "
|
||
"AND table_name = 'osv_entry' AND column_name = 'counterparty_match_kind'",
|
||
)
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_upgrade_0004_backfills_counterparty_match_kind(scratch_db_url: str) -> None:
|
||
"""0004: колонка и enum создаются; смапленные строки получают INN / NAME / ALIAS, без контрагента — NULL;
|
||
`downgrade 0003` убирает колонку и тип, повторный upgrade возвращает."""
|
||
engine = create_async_engine(scratch_db_url)
|
||
now = datetime.datetime.now(datetime.UTC)
|
||
ids = {k: uuid.uuid4() for k in ("ro", "le", "owner", "s3", "file", "job", "doc", "e1", "e2", "e3", "e4")}
|
||
try:
|
||
await alembic(scratch_db_url, "upgrade", V5_15)
|
||
async with engine.begin() as conn:
|
||
assert await _kind_column(conn) == []
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.legal_entity (id, inn, name, valid_from) VALUES "
|
||
"(:le, '7701000001', 'ООО Ромашка', :now), (:owner, '7702000002', 'ООО Владелец', :now)"
|
||
),
|
||
{**ids, "now": now},
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.legal_entity_alias (id, legal_entity_id, alias, normalized_alias) "
|
||
"VALUES (:id, :le, 'Ромашка Плюс', 'ромашка плюс')"
|
||
),
|
||
{"id": uuid.uuid4(), **ids},
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.s3_object (id, bucket_role, bucket, key, created_at) "
|
||
"VALUES (:s3, 'uploads', 'b', 'k', :now)"
|
||
),
|
||
{**ids, "now": now},
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.stored_file (id, legal_entity_id, file_type, s3_object_id, status) "
|
||
"VALUES (:file, :owner, 'XLSX', :s3, 'PROCESSED')"
|
||
),
|
||
ids,
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.recognition_job (id, file_id, pipeline, request_subject, schema_version, status) "
|
||
"VALUES (:job, :file, 'tabular', 'files.tabular_uploaded', '1', 'queued')"
|
||
),
|
||
ids,
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.document "
|
||
"(id, stored_file_id, legal_entity_id, document_type, osv_account_code, status) "
|
||
"VALUES (:doc, :file, :owner, 'OSV', '62', 'RECOGNIZED')"
|
||
),
|
||
ids,
|
||
)
|
||
entry = text(
|
||
"INSERT INTO dkb.osv_entry (id, recognition_job_id, document_id, owner_legal_entity_id, row_number, "
|
||
"counterparty_name, counterparty_inn, counterparty_id) "
|
||
"VALUES (:id, :job, :doc, :owner, :n, :name, :inn, :cp)"
|
||
)
|
||
base = {"job": ids["job"], "doc": ids["doc"], "owner": ids["owner"]}
|
||
rows = [
|
||
# ИНН с пробелами совпал (и имя тоже) → INN имеет приоритет
|
||
("e1", 1, "ООО Ромашка", " 7701 000 001 ", ids["le"]),
|
||
# ИНН не совпал, имя совпало после btrim → NAME
|
||
("e2", 2, " ООО Ромашка ", "9999999999", ids["le"]),
|
||
# только алиас → ALIAS
|
||
("e3", 3, "Ромашка Плюс", None, ids["le"]),
|
||
# контрагент не сопоставлен → NULL
|
||
("e4", 4, "ООО Неизвестный", None, None),
|
||
]
|
||
for key, n, name, inn, cp in rows:
|
||
await conn.execute(entry, {**base, "id": ids[key], "n": n, "name": name, "inn": inn, "cp": cp})
|
||
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
async with engine.connect() as conn:
|
||
assert await _kind_column(conn) == ["counterparty_match_kind"]
|
||
[labels] = await _scalars(conn, "SELECT enum_range(NULL::dkb.counterparty_match_kind)::text[]")
|
||
assert labels == ["INN", "NAME", "ALIAS", "MANUAL", "PROPAGATED"]
|
||
kinds = dict(
|
||
(await conn.execute(text("SELECT id, counterparty_match_kind::text FROM dkb.osv_entry"))).all()
|
||
)
|
||
assert kinds == {ids["e1"]: "INN", ids["e2"]: "NAME", ids["e3"]: "ALIAS", ids["e4"]: None}
|
||
await engine.dispose()
|
||
|
||
await alembic(scratch_db_url, "downgrade", V5_15)
|
||
async with engine.connect() as conn:
|
||
assert await _kind_column(conn) == []
|
||
with pytest.raises(ProgrammingError, match="counterparty_match_kind"):
|
||
await conn.execute(text("SELECT enum_range(NULL::dkb.counterparty_match_kind)"))
|
||
await engine.dispose()
|
||
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
async with engine.connect() as conn:
|
||
await _assert_full_model(conn)
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
async def _kpp_columns(conn: AsyncConnection) -> list[Any]:
|
||
return list(
|
||
(
|
||
await conn.execute(
|
||
text(
|
||
"SELECT table_name, is_nullable, data_type FROM information_schema.columns "
|
||
"WHERE table_schema = 'dkb' AND column_name = 'kpp' ORDER BY table_name"
|
||
)
|
||
)
|
||
).all()
|
||
)
|
||
|
||
|
||
async def _consolidation_id_column(conn: AsyncConnection) -> list[Any]:
|
||
return (
|
||
await conn.execute(
|
||
text(
|
||
"SELECT is_nullable, data_type FROM information_schema.columns "
|
||
"WHERE table_schema = 'dkb' AND table_name = 'stored_file' AND column_name = 'consolidation_id'"
|
||
)
|
||
)
|
||
).all()
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_upgrade_0006_backfills_consolidation_id(scratch_db_url: str) -> None:
|
||
"""0006: `stored_file.consolidation_id uuid` nullable, последняя колонка, индекс
|
||
`(legal_entity_id, consolidation_id)`; backfill по `consolidation_document`: файл A привязан к версиям 2025 Q1 и
|
||
2024 Q4 → 2024 Q4 (самый ранний период, год важнее квартала); файл B без привязок → NULL; файл C привязан к
|
||
версиям 2025 Q2 и 2025 Q1 → 2025 Q1 (квартал внутри года по возрастанию); `downgrade 0005`
|
||
убирает индекс и колонку, повторный upgrade возвращает."""
|
||
engine = create_async_engine(scratch_db_url)
|
||
now = datetime.datetime.now(datetime.UTC)
|
||
ids = {
|
||
name: uuid.uuid4()
|
||
for name in (
|
||
"office",
|
||
"group",
|
||
"entity",
|
||
"user",
|
||
"s3a",
|
||
"s3b",
|
||
"s3c",
|
||
"file_a",
|
||
"file_b",
|
||
"file_c",
|
||
"doc_a",
|
||
"doc_c1",
|
||
"doc_c2",
|
||
"c_2025q1",
|
||
"c_2025q2",
|
||
"c_2024q4",
|
||
"v_2025q1",
|
||
"v_2025q2",
|
||
"v_2024q4",
|
||
)
|
||
}
|
||
try:
|
||
await alembic(scratch_db_url, "upgrade", V5_17)
|
||
async with engine.begin() as conn:
|
||
assert await _consolidation_id_column(conn) == []
|
||
await conn.execute(text("INSERT INTO dkb.regional_office (id, name) VALUES (:office, 'ГО')"), ids)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.company_group (id, code, name, base_regional_office_id, valid_from) "
|
||
"VALUES (:group, 'G-1', 'ГК', :office, :now)"
|
||
),
|
||
ids | {"now": now},
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.legal_entity (id, inn, name, company_group_id, valid_from) "
|
||
"VALUES (:entity, '7701000001', 'ООО', :group, :now)"
|
||
),
|
||
ids | {"now": now},
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.app_user (id, email, full_name, regional_office_id) "
|
||
"VALUES (:user, 'm@x.ru', 'Тест', :office)"
|
||
),
|
||
ids,
|
||
)
|
||
for s3, file in (("s3a", "file_a"), ("s3b", "file_b"), ("s3c", "file_c")):
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.s3_object (id, bucket_role, bucket, key, created_at) "
|
||
f"VALUES (:{s3}, 'uploads', 'b', 'k-{s3}', :now)"
|
||
),
|
||
ids | {"now": now},
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.stored_file (id, legal_entity_id, file_type, s3_object_id, status, "
|
||
f"version_no) VALUES (:{file}, :entity, 'XLS', :{s3}, 'PROCESSED', 1)"
|
||
),
|
||
ids,
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.document (id, stored_file_id, legal_entity_id, document_type, osv_account_code, "
|
||
"period_year, period_quarter, status) "
|
||
"VALUES (:doc_a, :file_a, :entity, 'OSV', '60', 2024, 'Q4', 'CONFIRMED')"
|
||
),
|
||
ids,
|
||
)
|
||
for doc, account in (("doc_c1", "62"), ("doc_c2", "66")):
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.document (id, stored_file_id, legal_entity_id, document_type, "
|
||
"osv_account_code, period_year, period_quarter, status) "
|
||
f"VALUES (:{doc}, :file_c, :entity, 'OSV', '{account}', 2025, 'Q2', 'CONFIRMED')"
|
||
),
|
||
ids,
|
||
)
|
||
for c, v, year, quarter, rd in (
|
||
("c_2025q1", "v_2025q1", 2025, "Q1", "2025-04-01"),
|
||
("c_2025q2", "v_2025q2", 2025, "Q2", "2025-07-01"),
|
||
("c_2024q4", "v_2024q4", 2024, "Q4", "2025-01-01"),
|
||
):
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.consolidation (id, company_group_id, year, quarter, reporting_date) "
|
||
f"VALUES (:{c}, :group, {year}, '{quarter}', '{rd}')"
|
||
),
|
||
ids,
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.consolidation_version (id, consolidation_id, version_no, status, scenario, "
|
||
"approach, goodwill_enabled, initiator_id, initiator_office_id, is_in_work) "
|
||
f"VALUES (:{v}, :{c}, 1, 'INITIATED', 'CREATE', 'BASIC', false, :user, :office, true)"
|
||
),
|
||
ids,
|
||
)
|
||
for doc in {"v_2025q1": ("doc_a", "doc_c2"), "v_2025q2": ("doc_c1",), "v_2024q4": ("doc_a",)}[v]:
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.consolidation_document (id, consolidation_version_id, document_id) "
|
||
f"VALUES (gen_random_uuid(), :{v}, :{doc})"
|
||
),
|
||
ids,
|
||
)
|
||
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
async with engine.begin() as conn:
|
||
assert await _consolidation_id_column(conn) == [("YES", "uuid")]
|
||
last = await _scalars(
|
||
conn,
|
||
"SELECT column_name FROM information_schema.columns WHERE table_schema = 'dkb' "
|
||
"AND table_name = 'stored_file' ORDER BY ordinal_position DESC LIMIT 1",
|
||
)
|
||
assert last == ["consolidation_id"]
|
||
rows = (await conn.execute(text("SELECT id, consolidation_id FROM dkb.stored_file ORDER BY id"))).all()
|
||
assert dict(rows) == {ids["file_a"]: ids["c_2024q4"], ids["file_b"]: None, ids["file_c"]: ids["c_2025q1"]}
|
||
assert await _scalars(
|
||
conn,
|
||
"SELECT indexname FROM pg_indexes WHERE schemaname = 'dkb' "
|
||
"AND indexname = 'ix_stored_file_legal_entity_id_consolidation_id'",
|
||
) == ["ix_stored_file_legal_entity_id_consolidation_id"]
|
||
await engine.dispose()
|
||
|
||
await alembic(scratch_db_url, "downgrade", V5_17)
|
||
async with engine.connect() as conn:
|
||
assert await _consolidation_id_column(conn) == []
|
||
assert (
|
||
await _scalars(
|
||
conn,
|
||
"SELECT indexname FROM pg_indexes WHERE schemaname = 'dkb' "
|
||
"AND indexname = 'ix_stored_file_legal_entity_id_consolidation_id'",
|
||
)
|
||
== []
|
||
)
|
||
await engine.dispose()
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
async with engine.connect() as conn:
|
||
assert await _consolidation_id_column(conn) == [("YES", "uuid")]
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
async def _seed_v5_18_statements(conn: AsyncConnection, ids: dict[str, uuid.UUID]) -> None:
|
||
"""Данные на ревизии 0006: ЮЛ, файл, `INCOME` (t0) и `BALANCE` (t1 > t0) с формой ББ; ГСК без `crm_id`."""
|
||
t0 = datetime.datetime(2025, 10, 1, 9, 0, tzinfo=datetime.UTC)
|
||
t1 = t0 + datetime.timedelta(minutes=5)
|
||
await conn.execute(
|
||
text("INSERT INTO dkb.legal_entity (id, inn, name, valid_from) VALUES (:le, '7701234567', 'ООО Ромашка', :t0)"),
|
||
{**ids, "t0": t0},
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.s3_object (id, bucket_role, bucket, key, created_at) "
|
||
"VALUES (:s3, 'uploads', 'b', 'k', :t0)"
|
||
),
|
||
{**ids, "t0": t0},
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.stored_file (id, legal_entity_id, file_type, s3_object_id, status) "
|
||
"VALUES (:file, :le, 'XLSX', :s3, 'PROCESSED')"
|
||
),
|
||
ids,
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.recognition_job (id, file_id, pipeline, request_subject, schema_version, status) "
|
||
"VALUES (:job, :file, 'tabular', 'files.tabular_uploaded', '1', 'done')"
|
||
),
|
||
ids,
|
||
)
|
||
for key, doc_type, created in (("income", "INCOME", t0), ("balance", "BALANCE", t1)):
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.document (id, stored_file_id, legal_entity_id, document_type, status, created_at) "
|
||
f"VALUES (:{key}, :file, :le, '{doc_type}', 'NEEDS_CHECK', :created)"
|
||
),
|
||
{**ids, "created": created},
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.financial_statement_meta (id, document_id, recognition_job_id, statement_type, inn, "
|
||
"inn_original_value, inn_confidence, report_date_original_value, units_original_value, organization_name) "
|
||
"VALUES (:meta, :balance, :job, 'balance', '7701234567', '7701234567', 87, '30.09.2025', 'тыс. руб.', "
|
||
"'ООО Ромашка')"
|
||
),
|
||
ids,
|
||
)
|
||
for key, index, period in (("c1", 0, datetime.date(2025, 9, 30)), ("c2", 1, None)):
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.financial_statement_column (id, meta_id, column_index, period_date) "
|
||
"VALUES (:id, :meta, :idx, :period)"
|
||
),
|
||
{"id": ids[key], "meta": ids["meta"], "idx": index, "period": period},
|
||
)
|
||
for key, code, original in (("l1600", "1600", "1600"), ("l1100", "1100", "1100"), ("l1250", "1250", None)):
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.financial_statement_line (id, meta_id, line_code, line_code_original_value) "
|
||
"VALUES (:id, :meta, :code, :original)"
|
||
),
|
||
{"id": ids[key], "meta": ids["meta"], "code": code, "original": original},
|
||
)
|
||
values = [
|
||
# (строка, колонка, сумма, строка-значение, уверенность, распознано, правлено)
|
||
("l1100", "c1", 1, None, 100, True, False),
|
||
("l1100", "c2", 2, None, 80, True, False),
|
||
("l1600", "c1", 3, None, 80, True, True),
|
||
("l1600", "c2", 4, None, None, False, False),
|
||
("l1250", "c1", 5, "12 3", None, None, False),
|
||
]
|
||
for line, column, amount, text_value, confidence, recognized, modified in values:
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.financial_statement_value (id, line_id, column_id, amount, string_value, confidence, "
|
||
"is_recognized, is_modified) VALUES (:id, :line, :column, :amount, :sv, :conf, :rec, :mod)"
|
||
),
|
||
{
|
||
"id": uuid.uuid4(),
|
||
"line": ids[line],
|
||
"column": ids[column],
|
||
"amount": amount,
|
||
"sv": text_value,
|
||
"conf": confidence,
|
||
"rec": recognized,
|
||
"mod": modified,
|
||
},
|
||
)
|
||
await conn.execute(
|
||
text("INSERT INTO dkb.company_group (id, code, name, valid_from) VALUES (:grp, 'GK-1', 'ГК', :t0)"),
|
||
{**ids, "t0": t0},
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.related_group (id, code, company_group_id, kind, name) "
|
||
"VALUES (:rg, 'GSK-1', :grp, 'gsk', 'ГСК')"
|
||
),
|
||
ids,
|
||
)
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_upgrade_0007_backfills_v5_19_columns(scratch_db_url: str) -> None:
|
||
"""0007: заполнение по умолчанию на данных v5.18 (row_index, needs_review, source_text, period_kind, header_source,
|
||
file_ordinal, crm_id); `document_issue` пуста."""
|
||
engine = create_async_engine(scratch_db_url)
|
||
ids = {
|
||
k: uuid.uuid4()
|
||
for k in (
|
||
"le",
|
||
"s3",
|
||
"file",
|
||
"job",
|
||
"income",
|
||
"balance",
|
||
"meta",
|
||
"c1",
|
||
"c2",
|
||
"l1600",
|
||
"l1100",
|
||
"l1250",
|
||
"grp",
|
||
"rg",
|
||
)
|
||
}
|
||
try:
|
||
await alembic(scratch_db_url, "upgrade", V5_18)
|
||
async with engine.begin() as conn:
|
||
await _seed_v5_18_statements(conn, ids)
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
async with engine.connect() as conn:
|
||
lines = (
|
||
await conn.execute(
|
||
text(
|
||
"SELECT line_code, row_index, line_status::text, needs_review, is_confirmed "
|
||
"FROM dkb.financial_statement_line"
|
||
)
|
||
)
|
||
).all()
|
||
rows = {r.line_code: r for r in lines}
|
||
assert rows["1100"].row_index == 0 and rows["1600"].row_index == 1 and rows["1250"].row_index is None
|
||
assert {r.line_status for r in rows.values()} == {"ok"} and not any(r.needs_review for r in rows.values())
|
||
values = (
|
||
await conn.execute(
|
||
text(
|
||
"SELECT v.needs_review, v.source_text, v.string_value, v.is_confirmed FROM "
|
||
"dkb.financial_statement_value v JOIN dkb.financial_statement_line l ON l.id = v.line_id "
|
||
"ORDER BY v.amount"
|
||
)
|
||
)
|
||
).all()
|
||
assert [v.needs_review for v in values] == [False, True, False, True, False]
|
||
assert not any(v.is_confirmed for v in values)
|
||
value_by_text = {v.string_value: v for v in values if v.string_value}
|
||
assert value_by_text["12 3"].source_text == "12 3"
|
||
period_kind = dict(
|
||
(await conn.execute(text("SELECT column_index, period_kind FROM dkb.financial_statement_column"))).all()
|
||
)
|
||
assert period_kind == {0: "absolute", 1: "none"}
|
||
meta = (await conn.execute(text("SELECT * FROM dkb.financial_statement_meta"))).one()
|
||
assert meta.form_variant is None and meta.needs_review_fields is None
|
||
assert meta.header_source == {
|
||
"inn": {"text": "7701234567", "confidence": 87},
|
||
"report_date": {"text": "30.09.2025", "confidence": None},
|
||
"units": {"text": "тыс. руб.", "confidence": None},
|
||
"organization_name": {"text": "ООО Ромашка", "confidence": None},
|
||
}
|
||
assert meta.organization_name_original_value == "ООО Ромашка" and meta.organization_name_modified is False
|
||
ordinals = dict(
|
||
(await conn.execute(text("SELECT document_type::text, file_ordinal FROM dkb.document"))).all()
|
||
)
|
||
assert ordinals == {"INCOME": 0, "BALANCE": 1}
|
||
assert await _scalars(conn, "SELECT crm_id FROM dkb.related_group") == [None]
|
||
assert await _scalars(conn, "SELECT count(*) FROM dkb.document_issue") == [0]
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
async def _insert_document(conn: AsyncConnection, ids: dict[str, uuid.UUID], doc_type: str, ordinal: int) -> None:
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.document (id, stored_file_id, legal_entity_id, document_type, status, file_ordinal) "
|
||
f"VALUES (:id, :file, :le, '{doc_type}', 'NEEDS_CHECK', :ordinal)"
|
||
),
|
||
{"id": uuid.uuid4(), "file": ids["file"], "le": ids["le"], "ordinal": ordinal},
|
||
)
|
||
|
||
|
||
async def _insert_line(
|
||
conn: AsyncConnection, ids: dict[str, uuid.UUID], code: str | None, source_ref: str | None
|
||
) -> None:
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.financial_statement_line (id, meta_id, line_code, source_ref) "
|
||
"VALUES (:id, :meta, :code, :ref)"
|
||
),
|
||
{"id": uuid.uuid4(), "meta": ids["meta"], "code": code, "ref": source_ref},
|
||
)
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_v5_19_constraints(scratch_db_url: str) -> None:
|
||
"""Поведение индексов v5.19: формы одного типа, одна расшифровка, повтор кода, `source_ref`, `file_ordinal`."""
|
||
engine = create_async_engine(scratch_db_url)
|
||
ids = {
|
||
k: uuid.uuid4()
|
||
for k in (
|
||
"le",
|
||
"s3",
|
||
"file",
|
||
"job",
|
||
"income",
|
||
"balance",
|
||
"meta",
|
||
"c1",
|
||
"c2",
|
||
"l1600",
|
||
"l1100",
|
||
"l1250",
|
||
"grp",
|
||
"rg",
|
||
)
|
||
}
|
||
try:
|
||
await alembic(scratch_db_url, "upgrade", V5_18)
|
||
async with engine.begin() as conn:
|
||
await _seed_v5_18_statements(conn, ids)
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
async with engine.begin() as conn:
|
||
await _insert_document(conn, ids, "BALANCE", 2) # вторая форма того же типа — можно
|
||
await _insert_document(conn, ids, "DISCLOSURE", 3)
|
||
await _insert_line(conn, ids, "1100", None) # повтор кода — можно
|
||
await _insert_line(conn, ids, None, None)
|
||
await _insert_line(conn, ids, None, None) # source_ref IS NULL — можно повторять
|
||
await _insert_line(conn, ids, "1500", "row:8")
|
||
with pytest.raises(IntegrityError): # вторая DISCLOSURE (uq_document_file_type_account)
|
||
async with engine.begin() as conn:
|
||
await _insert_document(conn, ids, "DISCLOSURE", 4)
|
||
with pytest.raises(IntegrityError): # два документа файла с одним file_ordinal
|
||
async with engine.begin() as conn:
|
||
await _insert_document(conn, ids, "OTHER", 3)
|
||
with pytest.raises(IntegrityError): # тот же source_ref в форме
|
||
async with engine.begin() as conn:
|
||
await _insert_line(conn, ids, "1501", "row:8")
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
_INSERT_2000 = (
|
||
"INSERT INTO dkb.financial_statement_line (id, meta_id, line_code) VALUES (gen_random_uuid(), :meta, '2000')"
|
||
)
|
||
_SEED_KEYS = (
|
||
"le",
|
||
"s3",
|
||
"file",
|
||
"job",
|
||
"income",
|
||
"balance",
|
||
"meta",
|
||
"c1",
|
||
"c2",
|
||
"l1600",
|
||
"l1100",
|
||
"l1250",
|
||
"grp",
|
||
"rg",
|
||
)
|
||
|
||
|
||
async def _count(conn: AsyncConnection, table: str) -> int:
|
||
return (await _scalars(conn, f"SELECT count(*) FROM dkb.{table}"))[0]
|
||
|
||
|
||
async def _expect_downgrade_0007_refused(scratch_db_url: str, mutate_sql: str, reason: str) -> None:
|
||
"""Данные v5.19, несовместимые с v5.18: откат отказывает целиком, ревизия и данные не меняются."""
|
||
engine = create_async_engine(scratch_db_url)
|
||
ids = {k: uuid.uuid4() for k in _SEED_KEYS}
|
||
try:
|
||
await alembic(scratch_db_url, "upgrade", V5_18)
|
||
async with engine.begin() as conn:
|
||
await _seed_v5_18_statements(conn, ids)
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
async with engine.begin() as conn:
|
||
await conn.execute(text(mutate_sql.format(**{k: str(v) for k, v in ids.items()})))
|
||
lines_before = await _count(conn, "financial_statement_line")
|
||
documents_before = await _count(conn, "document")
|
||
with pytest.raises(AssertionError) as error:
|
||
await alembic(scratch_db_url, "downgrade", V5_18)
|
||
assert f"откат v5.19 невозможен: {reason}: 1" in str(error.value)
|
||
async with engine.connect() as conn:
|
||
assert await _scalars(conn, "SELECT version_num FROM dkb.alembic_version") == [HEAD]
|
||
assert await _count(conn, "financial_statement_line") == lines_before # данные не изменены
|
||
assert await _count(conn, "document") == documents_before
|
||
assert await _scalars(
|
||
conn, "SELECT count(*) FROM information_schema.tables WHERE table_name = 'document_issue'"
|
||
) == [1]
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_downgrade_0007_refuses_duplicate_code(scratch_db_url: str) -> None:
|
||
await _expect_downgrade_0007_refused(
|
||
scratch_db_url,
|
||
"INSERT INTO dkb.financial_statement_line (id, meta_id, line_code) "
|
||
"VALUES (gen_random_uuid(), '{meta}', '1100')",
|
||
"повтор кода строки в форме",
|
||
)
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_downgrade_0007_refuses_null_code(scratch_db_url: str) -> None:
|
||
await _expect_downgrade_0007_refused(
|
||
scratch_db_url,
|
||
"INSERT INTO dkb.financial_statement_line (id, meta_id, line_code, line_status) "
|
||
"VALUES (gen_random_uuid(), '{meta}', NULL, 'unreadable')",
|
||
"строки формы без кода",
|
||
)
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_downgrade_0007_refuses_two_forms_of_type(scratch_db_url: str) -> None:
|
||
await _expect_downgrade_0007_refused(
|
||
scratch_db_url,
|
||
"INSERT INTO dkb.document (id, stored_file_id, legal_entity_id, document_type, status, file_ordinal) "
|
||
"VALUES (gen_random_uuid(), '{file}', '{le}', 'BALANCE', 'NEEDS_CHECK', 2)",
|
||
"две формы одного типа в файле",
|
||
)
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_downgrade_0007_refuses_related_group_crm_id(scratch_db_url: str) -> None:
|
||
await _expect_downgrade_0007_refused(
|
||
scratch_db_url, "UPDATE dkb.related_group SET crm_id = 'CRM-1' WHERE id = '{rg}'", "crm_id ГСК/ГСЗ задан"
|
||
)
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_downgrade_0007_refuses_modified_organization_name(scratch_db_url: str) -> None:
|
||
await _expect_downgrade_0007_refused(
|
||
scratch_db_url,
|
||
"UPDATE dkb.financial_statement_meta SET organization_name_modified = true WHERE id = '{meta}'",
|
||
"наименование организации правилось",
|
||
)
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_downgrade_0007_drops_signals_and_logs_counts(scratch_db_url: str) -> None:
|
||
"""Без несовместимых данных откат проходит, журнал называет потерянное; цикл замыкается."""
|
||
engine = create_async_engine(scratch_db_url)
|
||
ids = {k: uuid.uuid4() for k in _SEED_KEYS}
|
||
try:
|
||
await alembic(scratch_db_url, "upgrade", V5_18)
|
||
async with engine.begin() as conn:
|
||
await _seed_v5_18_statements(conn, ids)
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
async with engine.begin() as conn:
|
||
await conn.execute(
|
||
text("UPDATE dkb.financial_statement_value SET is_confirmed = true WHERE amount IN (1, 2)")
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.document_issue (id, document_id, kind, code, severity, message, source, "
|
||
"created_at) "
|
||
"VALUES (gen_random_uuid(), :balance, 'warning', 'X', 'info', 'm', 'core', now())"
|
||
),
|
||
ids,
|
||
)
|
||
output = await alembic(scratch_db_url, "downgrade", V5_18)
|
||
assert "откат v5.19: удалены непустые значения" in output
|
||
assert "financial_statement_value.is_confirmed: 2" in output
|
||
assert "document_issue: 1" in output
|
||
async with engine.connect() as conn:
|
||
assert await _scalars(conn, "SELECT version_num FROM dkb.alembic_version") == [V5_18]
|
||
assert await _scalars(
|
||
conn,
|
||
"SELECT count(*) FROM information_schema.columns WHERE table_schema = 'dkb' AND "
|
||
"((table_name = 'financial_statement_line' "
|
||
"AND column_name IN ('row_index', 'source_ref', 'is_confirmed')) "
|
||
"OR (table_name = 'document' AND column_name = 'file_ordinal') "
|
||
"OR (table_name = 'related_group' AND column_name = 'crm_id'))",
|
||
) == [0]
|
||
assert await _scalars(
|
||
conn, "SELECT count(*) FROM information_schema.tables WHERE table_name = 'document_issue'"
|
||
) == [0]
|
||
assert await _scalars(conn, "SELECT count(*) FROM pg_type WHERE typname = 'document_issue_kind'") == [0]
|
||
await engine.dispose()
|
||
async with engine.begin() as conn: # уникальность (meta_id, line_code) вернулась
|
||
await conn.execute(
|
||
text(_INSERT_2000),
|
||
ids,
|
||
)
|
||
with pytest.raises(IntegrityError):
|
||
async with engine.begin() as conn:
|
||
await conn.execute(
|
||
text(_INSERT_2000),
|
||
ids,
|
||
)
|
||
await engine.dispose()
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_upgrade_0005_adds_nullable_kpp_columns(scratch_db_url: str) -> None:
|
||
"""0005: `kpp varchar` nullable у `legal_entity` и `legal_entity_history`, существующие ЮЛ — с `kpp IS NULL`;
|
||
`downgrade 0004` убирает обе колонки, повторный upgrade возвращает."""
|
||
engine = create_async_engine(scratch_db_url)
|
||
now = datetime.datetime.now(datetime.UTC)
|
||
entity_id = uuid.uuid4()
|
||
try:
|
||
await alembic(scratch_db_url, "upgrade", V5_16)
|
||
async with engine.begin() as conn:
|
||
assert await _kpp_columns(conn) == []
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.legal_entity (id, inn, name, valid_from) VALUES (:id, '7701000001', 'ООО', :now)"
|
||
),
|
||
{"id": entity_id, "now": now},
|
||
)
|
||
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
async with engine.begin() as conn:
|
||
assert await _kpp_columns(conn) == [
|
||
("legal_entity", "YES", "character varying"),
|
||
("legal_entity_history", "YES", "character varying"),
|
||
]
|
||
assert await _scalars(conn, "SELECT kpp FROM dkb.legal_entity") == [None]
|
||
await conn.execute(text("UPDATE dkb.legal_entity SET kpp = '770101001'")) # уникальности нет, тип varchar
|
||
order = await _scalars(
|
||
conn,
|
||
"SELECT column_name FROM information_schema.columns WHERE table_schema = 'dkb' "
|
||
"AND table_name = 'legal_entity' ORDER BY ordinal_position DESC LIMIT 1",
|
||
)
|
||
assert order == ["kpp"] # последней колонкой — как в ORM и DBML (порядок сверяет check_dbml)
|
||
await engine.dispose()
|
||
|
||
await alembic(scratch_db_url, "downgrade", V5_16)
|
||
async with engine.connect() as conn:
|
||
assert await _kpp_columns(conn) == []
|
||
assert await _scalars(conn, "SELECT inn FROM dkb.legal_entity") == ["7701000001"]
|
||
await engine.dispose()
|
||
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
async with engine.connect() as conn:
|
||
await _assert_full_model(conn)
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_upgrade_0002_keeps_membership_and_fills_kind(scratch_db_url: str) -> None:
|
||
"""0002 пересоздаёт `related_group_member`: строки v5.13 сохраняются, `kind` = `related_group.kind` группы."""
|
||
engine = create_async_engine(scratch_db_url)
|
||
now = datetime.datetime.now(datetime.UTC)
|
||
ids = {k: uuid.uuid4() for k in ("ro", "cg", "le", "gsk", "gsz", "m1", "m2")}
|
||
try:
|
||
await alembic(scratch_db_url, "upgrade", PREVIOUS)
|
||
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 kind in ("gsk", "gsz"):
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.related_group (id, code, company_group_id, kind, name) "
|
||
"VALUES (:g, :code, :cg, :kind, 'Группа')"
|
||
),
|
||
{"g": ids[kind], "code": kind, "cg": ids["cg"], "kind": kind},
|
||
)
|
||
# v5.13: одно текущее членство (в ГСЗ) и закрытое (в ГСК)
|
||
member = text(
|
||
"INSERT INTO dkb.related_group_member (id, related_group_id, legal_entity_id, valid_from, valid_to, "
|
||
"is_current) VALUES (:id, :g, :le, :now, :to, :cur)"
|
||
)
|
||
await conn.execute(
|
||
member, {"id": ids["m1"], "g": ids["gsz"], "le": ids["le"], "now": now, "to": None, "cur": True}
|
||
)
|
||
await conn.execute(
|
||
member, {"id": ids["m2"], "g": ids["gsk"], "le": ids["le"], "now": now, "to": now, "cur": False}
|
||
)
|
||
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
async with engine.connect() as conn:
|
||
rows = (
|
||
await conn.execute(
|
||
text("SELECT id, related_group_id, kind::text, valid_to, is_current FROM dkb.related_group_member")
|
||
)
|
||
).all()
|
||
assert {tuple(r) for r in rows} == {
|
||
(ids["m1"], ids["gsz"], "gsz", None, True),
|
||
(ids["m2"], ids["gsk"], "gsk", now, False),
|
||
}
|
||
leftovers = await _scalars(
|
||
conn, "SELECT relname FROM pg_class WHERE relnamespace = 'dkb'::regnamespace AND relname LIKE '%v5_13'"
|
||
)
|
||
assert leftovers == [] # временная копия удалена
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_autogenerate_finds_nothing_after_upgrade(scratch_db_url: str) -> None:
|
||
"""Сценарий «Автогенерация ничего не находит»: compare_metadata пуст, `alembic check` не видит операций."""
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
|
||
def diff(sync_conn: Connection) -> list[Any]:
|
||
context = MigrationContext.configure(
|
||
sync_conn,
|
||
opts={
|
||
# те же опции сравнения, что в alembic/env.py
|
||
"compare_type": True,
|
||
"compare_server_default": True,
|
||
"include_schemas": True,
|
||
"include_name": lambda name, type_, _parents: name == "dkb" if type_ == "schema" else True,
|
||
"version_table_schema": "dkb",
|
||
},
|
||
)
|
||
return compare_metadata(context, Base.metadata)
|
||
|
||
engine = create_async_engine(scratch_db_url)
|
||
try:
|
||
async with engine.connect() as conn:
|
||
assert await conn.run_sync(diff) == []
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
output = await alembic(scratch_db_url, "check")
|
||
assert "No new upgrade operations detected" in output
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_migrated_schema_equals_orm_create_all(scratch_db_url: str) -> None:
|
||
"""Каталог после миграции = каталог `create_all` ORM: колонки, default, ограничения, индексы, enum (имена и DDL)."""
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
reference_name = f"armcore_scratch_{uuid.uuid4().hex[:12]}"
|
||
reference_url = await create_scratch_db(reference_name)
|
||
migrated = create_async_engine(scratch_db_url)
|
||
reference = create_async_engine(reference_url)
|
||
try:
|
||
async with reference.begin() as conn:
|
||
await conn.execute(text("CREATE SCHEMA IF NOT EXISTS dkb"))
|
||
await conn.run_sync(Base.metadata.create_all)
|
||
async with migrated.connect() as conn:
|
||
got = await schema_snapshot(conn)
|
||
async with reference.connect() as conn:
|
||
expected = await schema_snapshot(conn)
|
||
assert got["enums"] == expected["enums"]
|
||
assert got["columns"] == expected["columns"]
|
||
assert got["constraints"] == expected["constraints"]
|
||
assert got["indexes"] == expected["indexes"]
|
||
finally:
|
||
await migrated.dispose()
|
||
await reference.dispose()
|
||
await drop_scratch_db(reference_name)
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_migrated_schema_behaviour(scratch_db_url: str) -> None:
|
||
"""После миграции: FK-цикл, partial unique, NULLS NOT DISTINCT и server_default работают на реальной БД."""
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
engine = create_async_engine(scratch_db_url)
|
||
try:
|
||
async with engine.connect() as conn:
|
||
fk = (
|
||
await conn.execute(
|
||
text(
|
||
"SELECT conrelid::regclass::text, pg_get_constraintdef(oid) FROM pg_constraint "
|
||
"WHERE conname = :name AND connamespace = 'dkb'::regnamespace"
|
||
),
|
||
{"name": FK_CYCLE},
|
||
)
|
||
).one()
|
||
assert fk[0] == "dkb.stored_file"
|
||
assert fk[1] == "FOREIGN KEY (current_recognition_job_id) REFERENCES dkb.recognition_job(id)"
|
||
rows = await conn.execute(
|
||
text(
|
||
"SELECT indexname, indexdef FROM pg_indexes WHERE schemaname = 'dkb' AND indexname IN "
|
||
"('uq_document_file_type_account', 'uq_consolidation_lock_version_active', "
|
||
"'uq_related_group_member_entity_kind_current', 'uq_related_group_member_group_entity_current')"
|
||
)
|
||
)
|
||
indexdefs = {row[0]: row[1] for row in rows}
|
||
assert "NULLS NOT DISTINCT" in indexdefs["uq_document_file_type_account"]
|
||
assert "WHERE (released_at IS NULL)" in indexdefs["uq_consolidation_lock_version_active"]
|
||
for name in ("uq_related_group_member_entity_kind_current", "uq_related_group_member_group_entity_current"):
|
||
assert indexdefs[name].startswith("CREATE UNIQUE INDEX")
|
||
assert "WHERE is_current" in indexdefs[name]
|
||
|
||
# поведенческие проверки из тестов ORM (create_all) — теперь на схеме из миграции; данные между ними
|
||
# очищаются: помощники создают одинаковые справочные записи (ГО, группа G1)
|
||
await _check_related_group_member_uniqueness(engine)
|
||
await _truncate_model_tables(engine)
|
||
async with AsyncSession(engine, expire_on_commit=False) as session:
|
||
chain = await _make_file_with_job(session) # замыкает FK-цикл, проверяет server_default
|
||
await session.commit()
|
||
await _check_nulls_not_distinct(session, chain)
|
||
await _check_nu_entry_inserts(session, chain) # v5.15: osv_data_type содержит NU
|
||
await _truncate_model_tables(engine)
|
||
async with AsyncSession(engine, expire_on_commit=False) as session:
|
||
await _check_active_lock_is_unique(session)
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_upgrade_into_preprovisioned_schema_without_database_create(scratch_db_url: str) -> None:
|
||
"""Схему dkb выдала платформа во владение роли миграций, права CREATE на БД у роли нет: цикл миграций проходит.
|
||
|
||
PostgreSQL проверяет право CREATE на БД раньше, чем IF NOT EXISTS, поэтому схема создаётся только при отсутствии.
|
||
"""
|
||
role = f"armcore_mig_{uuid.uuid4().hex[:12]}" # роли общие для кластера — имя уникальное, удаляется в finally
|
||
password = uuid.uuid4().hex
|
||
scratch_url = make_url(scratch_db_url)
|
||
role_url = scratch_url.set(username=role, password=password).render_as_string(hide_password=False)
|
||
server = create_async_engine(service_url(), isolation_level="AUTOCOMMIT")
|
||
scratch = create_async_engine(scratch_db_url, isolation_level="AUTOCOMMIT")
|
||
role_created = False
|
||
try:
|
||
async with server.connect() as conn:
|
||
await conn.execute(text(f"CREATE ROLE \"{role}\" LOGIN PASSWORD '{password}'"))
|
||
role_created = True
|
||
async with scratch.connect() as conn:
|
||
await conn.execute(text(f'REVOKE CREATE ON DATABASE "{scratch_url.database}" FROM PUBLIC'))
|
||
await conn.execute(text(f'CREATE SCHEMA dkb AUTHORIZATION "{role}"'))
|
||
can_create = await conn.scalar(
|
||
text("SELECT has_database_privilege(:role, :db, 'CREATE')"),
|
||
{"role": role, "db": scratch_url.database},
|
||
)
|
||
assert can_create is False
|
||
|
||
await alembic(role_url, "upgrade", "head")
|
||
await alembic(role_url, "check")
|
||
await alembic(role_url, "downgrade", "base")
|
||
await alembic(role_url, "upgrade", "head")
|
||
async with scratch.connect() as conn:
|
||
await _assert_full_model(conn)
|
||
owner = await conn.scalar(text("SELECT nspowner::regrole::text FROM pg_namespace WHERE nspname = 'dkb'"))
|
||
assert owner == role
|
||
finally:
|
||
if role_created:
|
||
async with scratch.connect() as conn:
|
||
await conn.execute(text(f'DROP OWNED BY "{role}" CASCADE'))
|
||
async with server.connect() as conn:
|
||
await conn.execute(text(f'DROP ROLE "{role}"'))
|
||
await scratch.dispose()
|
||
await server.dispose()
|
||
|
||
|
||
def test_offline_sql_creates_schema_before_version_table() -> None:
|
||
"""Офлайн-режим (`upgrade head --sql`): условный CREATE SCHEMA dkb идёт до таблицы версий и до enum-типов."""
|
||
output = run_alembic("postgresql+asyncpg://user:pass@localhost/offline", "upgrade", "head", "--sql")
|
||
schema_at = output.index(
|
||
"DO $$ BEGIN IF NOT EXISTS (SELECT 1 FROM pg_namespace WHERE nspname = 'dkb') "
|
||
"THEN CREATE SCHEMA dkb; END IF; END $$"
|
||
)
|
||
assert "CREATE SCHEMA IF NOT EXISTS" not in output # требует права CREATE на базу даже при существующей схеме
|
||
assert schema_at < output.index("CREATE TABLE dkb.alembic_version")
|
||
assert schema_at < output.index("CREATE TYPE dkb.entity_type AS ENUM ('ЮЛ', 'ИП')")
|
||
assert f"INSERT INTO dkb.alembic_version (version_num) VALUES ('{PREVIOUS}')" in output
|
||
# версия head пишется UPDATE'ами по цепочке 0001 → … → 0007 — офлайн-SQL доходит до head
|
||
for new_version, old_version in (
|
||
(V5_14, PREVIOUS),
|
||
(V5_15, V5_14),
|
||
(V5_16, V5_15),
|
||
(V5_17, V5_16),
|
||
(V5_18, V5_17),
|
||
(V5_19, V5_18),
|
||
(V5_20, V5_19),
|
||
(HEAD, V5_20),
|
||
):
|
||
assert (
|
||
f"UPDATE dkb.alembic_version SET version_num='{new_version}' "
|
||
f"WHERE dkb.alembic_version.version_num = '{old_version}'" in output
|
||
)
|
||
|
||
|
||
def test_cluster_commands_use_service_database() -> None:
|
||
"""CREATE/DROP DATABASE и роли — через служебную БД `postgres`, а не рабочую `armka` (финальное ревью M-9)."""
|
||
assert service_url().database == "postgres"
|
||
base = base_url()
|
||
assert (service_url().host, service_url().port, service_url().username) == (base.host, base.port, base.username)
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_v5_21_downgrade_preserves_document_fields(scratch_db_url: str) -> None:
|
||
engine = create_async_engine(scratch_db_url)
|
||
try:
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
await alembic(scratch_db_url, "downgrade", V5_20)
|
||
async with engine.connect() as conn:
|
||
names = await _scalars(
|
||
conn,
|
||
"SELECT column_name FROM information_schema.columns WHERE table_schema='dkb' AND table_name='document'",
|
||
)
|
||
assert {"original_values", "is_modified", "modified_by", "modified_at"} <= set(names)
|
||
labels = await _scalars(conn, "SELECT unnest(enum_range(NULL::dkb.audit_action))::text")
|
||
assert "FILE_REPROCESS" not in labels
|
||
await engine.dispose()
|
||
await alembic(scratch_db_url, "upgrade", "head")
|
||
async with engine.begin() as conn:
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO dkb.audit_event "
|
||
"(id, occurred_at, action, section, object_type) "
|
||
"VALUES (:id, now(), 'FILE_REPROCESS', 'PRIMARY_DOCUMENTS', 'stored_file')"
|
||
),
|
||
{"id": uuid.uuid4()},
|
||
)
|
||
with pytest.raises(AssertionError, match="откат v5.21 невозможен"):
|
||
await alembic(scratch_db_url, "downgrade", V5_20)
|
||
async with engine.connect() as conn:
|
||
assert await _scalars(conn, "SELECT version_num FROM dkb.alembic_version") == [HEAD]
|
||
assert await _count(conn, "audit_event") == 1
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_v5_20_downgrade_refuses_modified_document(scratch_db_url: str) -> None:
|
||
engine = create_async_engine(scratch_db_url)
|
||
ids = {k: uuid.uuid4() for k in _SEED_KEYS}
|
||
try:
|
||
await alembic(scratch_db_url, "upgrade", V5_18)
|
||
async with engine.begin() as conn:
|
||
await _seed_v5_18_statements(conn, ids)
|
||
await alembic(scratch_db_url, "upgrade", V5_20)
|
||
async with engine.begin() as conn:
|
||
await conn.execute(text("UPDATE dkb.document SET is_modified=true"))
|
||
with pytest.raises(AssertionError, match="откат v5.20 невозможен"):
|
||
await alembic(scratch_db_url, "downgrade", V5_19)
|
||
async with engine.connect() as conn:
|
||
assert await _scalars(conn, "SELECT version_num FROM dkb.alembic_version") == [V5_20]
|
||
assert await conn.scalar(text("SELECT count(*) FROM dkb.document WHERE is_modified"))
|
||
finally:
|
||
await engine.dispose()
|