"""Миграции 0001 (DBML v5.13), 0002 (v5.14), 0003 (v5.15), 0004 (v5.16), 0005 (v5.17), 0006 (v5.18) на 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 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, ) HEAD = "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() @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 → 0002 → 0003 → 0004 → 0005 → 0006 — офлайн-SQL доходит до head for new_version, old_version in ((V5_14, PREVIOUS), (V5_15, V5_14), (V5_16, V5_15), (V5_17, V5_16), (HEAD, V5_17)): 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)