From 36bc966bf6d32d0187a1f86b6b00a0975f2d3f5b Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Tue, 6 Oct 2026 04:57:15 +0400 Subject: [PATCH] =?UTF-8?q?feat(consolidation):=20=D0=BF=D1=80=D0=B8=D0=B2?= =?UTF-8?q?=D1=8F=D0=B7=D0=BA=D0=B0=20=D0=B4=D0=BE=D0=BA=D1=83=D0=BC=D0=B5?= =?UTF-8?q?=D0=BD=D1=82=D0=BE=D0=B2=20=D0=BA=20=D0=B2=D0=B5=D1=80=D1=81?= =?UTF-8?q?=D0=B8=D1=8F=D0=BC=20=D0=BF=D1=80=D0=B8=20=D0=BF=D0=BE=D0=B4?= =?UTF-8?q?=D1=82=D0=B2=D0=B5=D1=80=D0=B6=D0=B4=D0=B5=D0=BD=D0=B8=D0=B8,?= =?UTF-8?q?=20=D0=BF=D0=BE=D0=B4=D1=85=D0=B2=D0=B0=D1=82=20=D0=B8=20=D0=BE?= =?UTF-8?q?=D1=82=D0=B2=D1=8F=D0=B7=D0=BA=D0=B0=20=D0=BF=D1=80=D0=B8=20?= =?UTF-8?q?=D0=B0=D0=B2=D1=82=D0=BE=D0=B7=D0=B0=D0=BC=D0=B5=D0=BD=D0=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - consolidation_documents: чистая fits (D-5), bind_confirmed (подхват, самый поздний confirmed_at на ключ R-1), unbind_entity, on_document_confirmed (привязка ко всем подходящим версиям «в работе», переход в DATA_CORRECTION через transition), on_documents_superseded (отвязка документов заменённых файлов). - DocumentService.confirm вызывает хук привязки; supersede_previous снимает привязки и пишет FILE_REPLACE.after_value.unbound_version_ids; create и PUT …/members подхватывают по-настоящему. - R-2: advisory-блокировка привязок ГК (lock_group_bindings) в create и lock_versions_for_entity закрывает гонку «создание версии ∥ подтверждение/автозамена». - R-3: lock_document и lock_file_documents — FOR NO KEY UPDATE (совместимо с FOR KEY SHARE вставки привязки). - Тесты: таблица fits, сценарии documents-api и primary-files, две версии одного документа, гонки на двух сессиях (confirm ∥ members в обе стороны, create ∥ confirm в обе стороны, цикл блокировок документа и версии). - design.md: решения №30–№36; plan и tasks.md (3.1) отмечены. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2 --- api/src/repositories/consolidations.py | 21 +- api/src/repositories/files.py | 18 +- api/src/services/consolidation_documents.py | 248 ++++++- api/src/services/consolidations.py | 13 +- api/src/services/documents.py | 15 +- api/src/services/result_mapping.py | 25 +- api/tests/test_api_consolidation_members.py | 67 +- api/tests/test_consolidation_documents.py | 681 ++++++++++++++++++ api/tests/test_result_mapping.py | 121 +++- .../plans/2026-10-06-slice-4-consolidation.md | 6 +- .../changes/slice-4-consolidation/design.md | 7 + .../changes/slice-4-consolidation/tasks.md | 2 +- 12 files changed, 1174 insertions(+), 50 deletions(-) create mode 100644 api/tests/test_consolidation_documents.py diff --git a/api/src/repositories/consolidations.py b/api/src/repositories/consolidations.py index daa2527..2ccfa13 100644 --- a/api/src/repositories/consolidations.py +++ b/api/src/repositories/consolidations.py @@ -2,7 +2,8 @@ Функции не делают `commit` и, кроме `get_version`, `get_consolidation_for_group` и `list_periods` (они получают `VisibilityScope` и скрывают ГК чужого филиала), не применяют policy: проверку видимости делает вызывающий сервис. -Порядок блокировок строк — файл → документ → версии; `lock_versions_for_entity` берёт версии пачкой по `id`. +Порядок блокировок — файл → документ → advisory ГК (`lock_group_bindings`) → версии; `lock_versions_for_entity` берёт +advisory ГК ЮЛ и версии пачкой по `id`. """ import uuid @@ -29,6 +30,7 @@ from src.models.files import Document, StoredFile from src.models.osv import OsvEntry FILES_IN_PROGRESS = (FileStatus.UPLOADED, FileStatus.PROCESSING) +GROUP_BINDINGS_LOCK_PREFIX = "consolidation-group" # пространство ключей advisory-блокировки привязок ГК (R-2) SUMMARY_DATA_TYPES = (OsvDataType.BU, OsvDataType.SUM) @@ -96,11 +98,23 @@ async def lock_version(session: AsyncSession, version_id: uuid.UUID) -> Consolid return (await session.execute(stmt)).scalar_one_or_none() +async def lock_group_bindings(session: AsyncSession, group_id: uuid.UUID) -> None: + """`pg_advisory_xact_lock` привязок документов ГК (R-2, design №30): сериализует создание версии консолидации ГК с + подтверждением и автозаменой документов ЮЛ этой ГК. Новая версия невидима другим транзакциям до commit, поэтому + `FOR UPDATE` версий её не ловит: создание берёт этот lock до чтения подтверждённых документов, а + `lock_versions_for_entity` — до блокировки версий. Снимается в конце транзакции.""" + key = func.hashtextextended(f"{GROUP_BINDINGS_LOCK_PREFIX}:{group_id}", 0) + await session.execute(select(func.pg_advisory_xact_lock(key))) + + async def lock_versions_for_entity( session: AsyncSession, entity_id: uuid.UUID, statuses: Collection[ConsolidationStatus] ) -> list[ConsolidationVersion]: """Версии со статусом из `statuses`, где ЮЛ — член периметра; блокируются `FOR UPDATE ORDER BY id` (R-2). + Первой берётся advisory-блокировка привязок ГК ЮЛ (`lock_group_bindings`; ЮЛ без ГК — без неё: членом версии оно + быть не может): иначе версия, создаваемая параллельно, не видна ни здесь, ни подтверждённый нами документ — там. + Блокируется НАДМНОЖЕСТВО: версии этих статусов, где ЮЛ — член ИЛИ `consolidation.company_group_id` равен ГК ЮЛ (`legal_entity.company_group_id`). Иначе подтверждение документа ∥ `PUT …/members` теряет привязку: подтверждение не видит ещё не зафиксированного нового члена и не ждёт версию, которую правит периметр. Блокировка версии ГК, @@ -113,6 +127,11 @@ async def lock_versions_for_entity( ConsolidationMember.consolidation_version_id == ConsolidationVersion.id, ConsolidationMember.legal_entity_id == entity_id, ) + group_id = ( + await session.execute(select(LegalEntity.company_group_id).where(LegalEntity.id == entity_id)) + ).scalar_one_or_none() + if group_id is not None: + await lock_group_bindings(session, group_id) entity_group = select(LegalEntity.company_group_id).where(LegalEntity.id == entity_id).scalar_subquery() superset = ( select(ConsolidationVersion.id) diff --git a/api/src/repositories/files.py b/api/src/repositories/files.py index eaf81a1..07fd79c 100644 --- a/api/src/repositories/files.py +++ b/api/src/repositories/files.py @@ -123,9 +123,16 @@ async def lock_file(session: AsyncSession, file_id: uuid.UUID) -> StoredFile | N async def lock_document(session: AsyncSession, document_id: uuid.UUID) -> Document | None: - """`SELECT document … FOR UPDATE` с перечитыванием; вызывать после `lock_file` его файла (порядок блокировок).""" + """`SELECT document … FOR NO KEY UPDATE` с перечитыванием; вызывать после `lock_file` его файла. + + R-3 (design №13): `FOR NO KEY UPDATE`, а не `FOR UPDATE` — другие писатели документа ждут так же, но вставка + `consolidation_document` (FK берёт `FOR KEY SHARE` документа) не ждёт: иначе подтверждение (держит документ, ждёт + версию) и подхват (держит версию, вставляет привязку) взаимоблокировались бы.""" stmt = ( - select(Document).where(Document.id == document_id).with_for_update().execution_options(populate_existing=True) + select(Document) + .where(Document.id == document_id) + .with_for_update(key_share=True) + .execution_options(populate_existing=True) ) return (await session.execute(stmt)).scalar_one_or_none() @@ -144,12 +151,15 @@ async def lock_jobs(session: AsyncSession, file_id: uuid.UUID) -> list[Recogniti async def lock_file_documents(session: AsyncSession, file_id: uuid.UUID) -> list[Document]: - """`SELECT document … FOR UPDATE` всех документов файла (по `id`) с перечитыванием; после `lock_file`.""" + """`SELECT document … FOR NO KEY UPDATE` всех документов файла (по `id`) с перечитыванием; после `lock_file`. + + Режим — как у `lock_document` (R-3): совместим с `FOR KEY SHARE` вставки привязки; `DELETE` документа (удаление + файла) сам повышает блокировку до `FOR UPDATE`.""" stmt = ( select(Document) .where(Document.stored_file_id == file_id) .order_by(Document.id) - .with_for_update() + .with_for_update(key_share=True) .execution_options(populate_existing=True) ) return list((await session.execute(stmt)).scalars().all()) diff --git a/api/src/services/consolidation_documents.py b/api/src/services/consolidation_documents.py index 63f7d23..8c10ba8 100644 --- a/api/src/services/consolidation_documents.py +++ b/api/src/services/consolidation_documents.py @@ -1,39 +1,177 @@ -"""Привязка подтверждённых документов к версии консолидации (spec documents-api «Подхват документов»; design D-5). +"""Привязка подтверждённых документов к версиям консолидации (spec documents-api «Подтверждение документа», «Подхват +документов при создании версии и расширении периметра»; primary-files «Автозамена прежних версий»; audit-trail +`CONSOLIDATION_STATUS_CHANGE`, `FILE_REPLACE.unbound_version_ids`; design D-5, D-6, D-9; рулинги R-1, R-2, R-3). -ЗАГЛУШКА Среза 4, задача 3: `bind_confirmed` ничего не привязывает и возвращает 0. Настоящую реализацию (отбор -подтверждённых документов ЮЛ по правилу «подходящести» D-5, `consolidation_document`, порядок блокировок -файл → документ → версии, R-1…R-3) вносит Task 6 — она заменяет тело функции, сигнатуру и вызовы из -`services.consolidations.ConsolidationService` (`create`, `set_members`) сохраняет. +Точки привязки: подтверждение документа (`on_document_confirmed` — ко всем подходящим версиям «в работе», с переходом +`INITIATED`/`DOCUMENTS_PROCESSING → DATA_CORRECTION`), создание версии и добавление ЮЛ в периметр (`bind_confirmed`, +статус не меняется). Точки отвязки: автозамена файла (`on_documents_superseded`), удаление ЮЛ из периметра +(`unbind_entity`). Версии вне «в работе» не трогаются. -`unbind_entity` (Task 5) — прямое удаление привязок ЮЛ к версии; Task 6 владеет функцией и вправе переписать тело -(имя и сигнатура — прежние): вызывается из `set_members` для ЮЛ, ушедших из периметра. +Ключ замены (R-1): у одного ЮЛ в одной версии на ключ `(document_type, osv_account_code, period_quarter, cumulative)` +привязан один документ — с самым поздним `confirmed_at`; `cumulative = is_cumulative is not False` (NULL — ББ/ОФР — +нарастающий, решение №10). Квартальный и нарастающий документы одного квартала друг друга не вытесняют. + +Порядок блокировок (D-9, R-2, R-3): файл → документ (`FOR NO KEY UPDATE`) → advisory ГК → версии `FOR UPDATE` по `id` +(`repo.lock_versions_for_entity`); автозамена — advisory ЮЛ → старые файлы → advisory ГК → версии; создание версии — +advisory ГК первой; правка периметра — версия первой, advisory не берёт. Вставка привязки берёт на документ +`FOR KEY SHARE`, с `FOR NO KEY UPDATE` подтверждения она совместима. """ +import datetime import uuid -from collections.abc import Sequence +from collections.abc import Collection, Iterable, Sequence +from typing import Any from sqlalchemy import delete, select +from sqlalchemy.engine import CursorResult from sqlalchemy.ext.asyncio import AsyncSession -from src.models.consolidation import ConsolidationDocument, ConsolidationVersion -from src.models.files import Document +from src.domain.policy import IN_WORK_STATUSES +from src.models.consolidation import Consolidation, ConsolidationDocument, ConsolidationMember, ConsolidationVersion +from src.models.enums import ConsolidationStatus, DocumentStatus, DocumentType, FileStatus, PeriodQuarter +from src.models.files import Document, StoredFile +from src.repositories import consolidations as repo +from src.services.audit import AuditService + +__all__ = ["bind_confirmed", "fits", "on_document_confirmed", "on_documents_superseded", "unbind_entity"] + +#: Статусы, из которых подтверждение привязанного документа переводит версию в `DATA_CORRECTION` (D-6). +MOVES_TO_CORRECTION = frozenset({ConsolidationStatus.INITIATED, ConsolidationStatus.DOCUMENTS_PROCESSING}) +_QUARTER_ORDER = {quarter: position for position, quarter in enumerate(PeriodQuarter)} + +BindingKey = tuple[uuid.UUID, DocumentType, str | None, PeriodQuarter | None, bool] + + +def _rowcount(result: Any) -> int: + """Число строк DML (как `services.file_status`): у `CursorResult` — `rowcount`, иначе 0.""" + return result.rowcount if isinstance(result, CursorResult) else 0 + + +def _cumulative(document: Document) -> bool: + """Нарастающий ли документ: `is_cumulative` true или NULL (ББ/ОФР всегда за период с начала года, решение №10).""" + return document.is_cumulative is not False + + +def _key(document: Document) -> BindingKey: + """Ключ замены привязки (R-1): ЮЛ, тип, счёт, квартал и признак нарастающего.""" + return ( + document.legal_entity_id, + document.document_type, + document.osv_account_code, + document.period_quarter, + _cumulative(document), + ) + + +def _latest_first(document: Document) -> tuple[datetime.datetime, uuid.UUID]: + """Порядок «самый поздний `confirmed_at` первым»; равные — по `id` (детерминированно).""" + confirmed_at = document.confirmed_at or datetime.datetime.min.replace(tzinfo=datetime.UTC) + return confirmed_at, document.id + + +def fits( + document: Document, + version: ConsolidationVersion, # noqa: ARG001 - сигнатура D-5: подходящесть — функция документа и версии + consolidation: Consolidation, + member_ids: Collection[uuid.UUID], +) -> bool: + """Подходит ли документ версии (D-5): ЮЛ — член периметра, год версии, квартал определён и: нарастающий (или + NULL) — квартал версии; квартальный — не позже квартала версии. Статус документа и файла не проверяется — это + дело вызывающего (привязываются только `CONFIRMED` незаменённых файлов). Чистая функция, без БД.""" + if document.legal_entity_id not in member_ids or document.period_year != consolidation.year: + return False + if document.period_quarter is None: + return False + if _cumulative(document): + return document.period_quarter == consolidation.quarter + return _QUARTER_ORDER[document.period_quarter] <= _QUARTER_ORDER[consolidation.quarter] + + +async def _member_ids(session: AsyncSession, version_id: uuid.UUID) -> set[uuid.UUID]: + stmt = select(ConsolidationMember.legal_entity_id).where(ConsolidationMember.consolidation_version_id == version_id) + return set((await session.execute(stmt)).scalars()) + + +async def _bound_documents( + session: AsyncSession, version_id: uuid.UUID, entity_ids: Iterable[uuid.UUID] +) -> list[Document]: + """Документы ЮЛ `entity_ids`, уже привязанные к версии.""" + stmt = ( + select(Document) + .join(ConsolidationDocument, ConsolidationDocument.document_id == Document.id) + .where( + ConsolidationDocument.consolidation_version_id == version_id, + Document.legal_entity_id.in_(list(entity_ids)), + ) + ) + return list((await session.execute(stmt)).scalars()) + + +async def _unbind(session: AsyncSession, version_id: uuid.UUID, document_ids: Collection[uuid.UUID]) -> int: + if not document_ids: + return 0 + result = await session.execute( + delete(ConsolidationDocument).where( + ConsolidationDocument.consolidation_version_id == version_id, + ConsolidationDocument.document_id.in_(list(document_ids)), + ) + ) + return _rowcount(result) + + +async def _bind(session: AsyncSession, version_id: uuid.UUID, document: Document) -> None: + """Привязывает `document` к версии, предварительно отвязав привязанные документы с тем же ключом (R-1).""" + bound = await _bound_documents(session, version_id, [document.legal_entity_id]) + await _unbind(session, version_id, [other.id for other in bound if _key(other) == _key(document)]) + session.add(ConsolidationDocument(consolidation_version_id=version_id, document_id=document.id)) + await session.flush() async def bind_confirmed( - session: AsyncSession, # noqa: ARG001 - сигнатура окончательная, тело — заглушка до Task 6 - version: ConsolidationVersion, # noqa: ARG001 - entity_ids: Sequence[uuid.UUID], # noqa: ARG001 + session: AsyncSession, + *, + version: ConsolidationVersion, + consolidation: Consolidation, + entity_ids: Sequence[uuid.UUID], ) -> int: - """Привязывает к `version` подходящие подтверждённые документы ЮЛ `entity_ids`; возвращает число привязок.""" - return 0 + """Подхват (spec «Подхват документов…»): привязывает к версии `CONFIRMED` документы ЮЛ `entity_ids` из файлов не + `SUPERSEDED`, подходящие версии (`fits`); на ключ (R-1) — документ с самым поздним `confirmed_at`. Привязанный + документ того же ключа заменяется; уже привязанный победитель не трогается. Статус версии не меняется. Возвращает + число новых привязок. + + Версия должна быть под `FOR UPDATE` вызывающего (правка периметра) либо только что создана под advisory ГК + (`ConsolidationService.create`, R-2). + """ + if not entity_ids: + return 0 + members = await _member_ids(session, version.id) + stmt = ( + select(Document) + .join(StoredFile, StoredFile.id == Document.stored_file_id) + .where( + Document.legal_entity_id.in_(list(entity_ids)), + Document.status == DocumentStatus.CONFIRMED, + Document.period_year == consolidation.year, + StoredFile.status != FileStatus.SUPERSEDED, + ) + ) + candidates = [doc for doc in (await session.execute(stmt)).scalars() if fits(doc, version, consolidation, members)] + winners: dict[BindingKey, Document] = {} + for document in sorted(candidates, key=_latest_first, reverse=True): + winners.setdefault(_key(document), document) + bound = await _bound_documents(session, version.id, entity_ids) + bound_ids = {document.id for document in bound} + replaced = [doc.id for doc in bound if _key(doc) in winners and winners[_key(doc)].id != doc.id] + await _unbind(session, version.id, replaced) + fresh = [document for document in winners.values() if document.id not in bound_ids] + session.add_all(ConsolidationDocument(consolidation_version_id=version.id, document_id=doc.id) for doc in fresh) + await session.flush() + return len(fresh) async def unbind_entity(session: AsyncSession, *, version_id: uuid.UUID, entity_id: uuid.UUID) -> int: - """Снимает все привязки документов ЮЛ `entity_id` к версии `version_id`; возвращает число снятых. - - Task 5: простое прямое удаление `consolidation_document` по `document.legal_entity_id`. Принадлежит Task 6 - (отвязка по правилам D-5, порядок блокировок); версия должна быть под `FOR UPDATE` вызывающего. - """ + """Снимает все привязки документов ЮЛ `entity_id` к версии `version_id` (удаление ЮЛ из периметра); возвращает + число снятых. Статус версии не меняется; версия должна быть под `FOR UPDATE` вызывающего.""" owned = select(Document.id).where(Document.legal_entity_id == entity_id) result = await session.execute( delete(ConsolidationDocument).where( @@ -41,4 +179,72 @@ async def unbind_entity(session: AsyncSession, *, version_id: uuid.UUID, entity_ ConsolidationDocument.document_id.in_(owned), ) ) - return int(result.rowcount) # type: ignore[attr-defined] # CursorResult у DML; у базового Result атрибута нет + return _rowcount(result) + + +async def on_document_confirmed( + session: AsyncSession, + audit: AuditService, + *, + document: Document, + user_id: uuid.UUID, + now: datetime.datetime, +) -> list[uuid.UUID]: + """Хук подтверждения (spec «Подтверждение документа»): привязывает уже `CONFIRMED` документ ко всем версиям «в + работе», которым он подходит, заменяя привязанный документ того же ключа (R-1); версия из `INITIATED` / + `DOCUMENTS_PROCESSING` переходит в `DATA_CORRECTION` (`changed_by = user_id`, D-6). Возвращает id версий, + получивших привязку (по возрастанию). + + Вызывать после блокировки файла и документа (D-9): версии блокирует `repo.lock_versions_for_entity` — advisory ГК + (создание версии ∥ подтверждение) и `FOR UPDATE` надмножества версий ГК (правка периметра ∥ подтверждение, R-2). + Блокировка версии (`consolidation_lock`) не требуется. + """ + # цикл импортов: consolidation_status → consolidations → consolidation_documents + from src.services.consolidation_status import transition # noqa: PLC0415 + + versions = await repo.lock_versions_for_entity(session, document.legal_entity_id, IN_WORK_STATUSES) + bound: list[uuid.UUID] = [] + for version in versions: + consolidation = await session.get(Consolidation, version.consolidation_id) + if consolidation is None: # FK гарантирует наличие + continue + if not fits(document, version, consolidation, await _member_ids(session, version.id)): + continue + await _bind(session, version.id, document) + if version.status in MOVES_TO_CORRECTION: + await transition( + session, + audit, + version=version, + to=ConsolidationStatus.DATA_CORRECTION, + changed_by=user_id, + comment=None, + now=now, + ) + bound.append(version.id) + return bound + + +async def on_documents_superseded( + session: AsyncSession, + *, + documents: Sequence[Document], + now: datetime.datetime, # noqa: ARG001 - сигнатура фиксирована планом (Task 7/8); статус версии не меняется +) -> list[uuid.UUID]: + """Хук автозамены (primary-files «Автозамена прежних версий»): снимает привязки `documents` (документы заменённого + файла) ко всем версиям «в работе»; версии других статусов и статус версии не трогаются. Возвращает id версий, где + привязки сняты (`unbound_version_ids`, по возрастанию). + + Вызывать после блокировки старых файлов: версии блокирует `repo.lock_versions_for_entity` по ЮЛ документов (R-2: + подхват, ждавший нас, уже закоммитил привязку — она видна и снимается). + """ + document_ids = [document.id for document in documents] + if not document_ids: + return [] + unbound: set[uuid.UUID] = set() + for entity_id in sorted({document.legal_entity_id for document in documents}): + versions = await repo.lock_versions_for_entity(session, entity_id, IN_WORK_STATUSES) + for version in versions: + if await _unbind(session, version.id, document_ids): + unbound.add(version.id) + return sorted(unbound) diff --git a/api/src/services/consolidations.py b/api/src/services/consolidations.py index 1deca69..aa75be9 100644 --- a/api/src/services/consolidations.py +++ b/api/src/services/consolidations.py @@ -87,10 +87,15 @@ class ConsolidationService: Проверки по порядку: ГК видима (404) → ГСК видима (404), принадлежит ГК и `kind = gsk` (422 `RELATED_GROUP_MISMATCH`) → в ГСК не меньше двух текущих членов (422 `PERIMETER_TOO_SMALL`) → периода ещё нет (409 `CONSOLIDATION_EXISTS`; гонку закрывает UNIQUE). Всё — в транзакции запроса: любая ошибка откатывает всё. + После проверки ГК берётся advisory-блокировка привязок ГК (R-2): подхват документов (`bind_confirmed`) не + разминётся с параллельным подтверждением или автозаменой документа ЮЛ этой ГК. """ group = await directories_repo.get_company_group(self._session, self._scope, data.company_group_id) if group is None: raise NotFoundError(COMPANY_GROUP_NOT_FOUND) + # R-2 (design №30): до чтения подтверждённых документов — подтверждение и автозамена документов ЮЛ ГК ждут нас + # (новая версия им невидима до commit), а мы — их + await repo.lock_group_bindings(self._session, group.id) related = await directories_repo.get_related_group(self._session, self._scope, data.related_group_id) if related is None: raise NotFoundError(RELATED_GROUP_NOT_FOUND) @@ -199,7 +204,9 @@ class ConsolidationService: for entity_id in removed: await consolidation_documents.unbind_entity(self._session, version_id=version.id, entity_id=entity_id) if added: - await consolidation_documents.bind_confirmed(self._session, version, added) + await consolidation_documents.bind_confirmed( + self._session, version=version, consolidation=consolidation, entity_ids=added + ) await self._audit.record( action=AuditAction.MEMBERSHIP_CHANGE, section=AuditSection.CONSOLIDATION, @@ -255,7 +262,9 @@ class ConsolidationService: ) ) await self._session.flush() - await consolidation_documents.bind_confirmed(self._session, version, [entity.id for entity in entities]) + await consolidation_documents.bind_confirmed( + self._session, version=version, consolidation=consolidation, entity_ids=[entity.id for entity in entities] + ) return consolidation, version @staticmethod diff --git a/api/src/services/documents.py b/api/src/services/documents.py index 672dc16..15fcfce 100644 --- a/api/src/services/documents.py +++ b/api/src/services/documents.py @@ -4,9 +4,10 @@ Видимость — через ЮЛ-владельца (`repositories.files`); документ вне видимости — тот же 404, что несуществующий. Подтверждение и автозамена (`services.result_mapping.supersede_previous`) могут идти одновременно. Порядок блокировок -подтверждения — сначала `stored_file … FOR UPDATE`, затем `document … FOR UPDATE`: автозамена блокирует файлы (новый, -затем старые по id) и документы не блокирует, поэтому взаимной блокировки нет. Файл и документ перечитываются под -блокировкой: файл, заменённый параллельной автозаменой, даёт 409 `FILE_SUPERSEDED` (Review Focus 5). +подтверждения — `stored_file … FOR UPDATE` → `document … FOR NO KEY UPDATE` (R-3) → advisory ГК → версии консолидации +«в работе» по `id` (`consolidation_documents.on_document_confirmed`, D-9, R-2): автозамена блокирует файлы (новый, затем +старые по id), затем advisory ГК и версии, документы не блокирует, поэтому взаимной блокировки нет. Файл и документ +перечитываются под блокировкой: файл, заменённый параллельной автозаменой, даёт 409 `FILE_SUPERSEDED` (Review Focus 5). """ import uuid @@ -25,6 +26,7 @@ from src.models.osv import OsvEntry from src.repositories import files as files_repo from src.schemas.documents import DocumentOut, EntriesPageParams, GroupOut from src.schemas.files import DocumentBrief, FileOut, JobBrief +from src.services import consolidation_documents from src.services.result_mapping import MAPPING_ERROR_KEY, MAPPING_WARNINGS_KEY DOCUMENT_NOT_FOUND = "документ не найден" # одинаково для чужого и несуществующего @@ -148,7 +150,9 @@ class DocumentService: """`RECOGNIZED`/`NEEDS_CHECK` → `CONFIRMED` (`confirmed_by/at`), аудит `DOCUMENT_CONFIRM`. 409: файл `SUPERSEDED` — `FILE_SUPERSEDED`; уже `CONFIRMED` — `ALREADY_CONFIRMED`; иной статус — - `DOCUMENT_NOT_CONFIRMABLE`. Порядок блокировок и перечитывание — см. docstring модуля. + `DOCUMENT_NOT_CONFIRMABLE`. Порядок блокировок и перечитывание — см. docstring модуля. В той же транзакции + документ привязывается к подходящим версиям «в работе», версии из `INITIATED`/`DOCUMENTS_PROCESSING` переходят + в `DATA_CORRECTION` (`on_document_confirmed`, spec documents-api «Подтверждение документа»). """ visible = await self._visible(document_id) file = await files_repo.lock_file(self._session, visible.stored_file_id) @@ -173,6 +177,9 @@ class DocumentService: document.confirmed_at = now document.updated_at = now await self._session.flush() + await consolidation_documents.on_document_confirmed( + self._session, self._audit.service, document=document, user_id=self._audit.user_id, now=now + ) await self._audit.record( action=AuditAction.DOCUMENT_CONFIRM, section=AuditSection.PRIMARY_DOCUMENTS, diff --git a/api/src/services/result_mapping.py b/api/src/services/result_mapping.py index 1ec0abe..6e4df05 100644 --- a/api/src/services/result_mapping.py +++ b/api/src/services/result_mapping.py @@ -65,6 +65,7 @@ from src.models.enums import ( from src.models.files import Document, RecognitionJob, S3Object, StoredFile from src.models.osv import OsvEntry from src.repositories.counterparties import CounterpartyIndex +from src.services import consolidation_documents from src.services.audit import AuditService from src.services.file_status import set_file_status_for_job @@ -472,9 +473,14 @@ async def supersede_previous( `confirmed=true`, если среди заменённых документов есть `CONFIRMED`; решение владельца 05.10.2026). Документы и строки заменённых файлов не трогаются (история). Возвращает заменённые файлы; в транзакции вызывающего, без commit. + Привязки документов заменённых файлов к версиям консолидации «в работе» снимаются + (`consolidation_documents.on_documents_superseded`); id версий — в `after_value.unbound_version_ids` события + заменённого файла (spec audit-trail), статус версий не меняется. + Блокировки: advisory-блокировка ЮЛ сериализует автозамены одного ЮЛ (два новых файла одного документа, смапленные параллельно, иначе не увидели бы друг друга и оба остались бы действующими); заменяемые файлы — `FOR UPDATE` в - порядке `id` (сериализация с подтверждением документа, которое перечитывает файл под `FOR UPDATE`). + порядке `id` (сериализация с подтверждением документа, которое перечитывает файл под `FOR UPDATE`); затем + advisory ГК и версии консолидации по `id` (D-9, R-2). """ keys = {key for key in map(_document_key, new_documents) if key is not None} if not keys: @@ -528,6 +534,17 @@ async def supersede_previous( new_file.version_no = max(old.version_no for old in locked) + 1 new_file.updated_at = now await session.flush() + # привязки к версиям «в работе» снимаются у ВСЕХ документов заменённого файла (spec primary-files), а не только у + # совпавших по ключу: подхват берёт документы лишь незаменённых файлов (design №33) + file_documents = await session.execute(select(Document).where(Document.stored_file_id.in_(ids))) + documents_by_file: dict[uuid.UUID, list[Document]] = {} + for document in file_documents.scalars(): + documents_by_file.setdefault(document.stored_file_id, []).append(document) + unbound: dict[uuid.UUID, list[uuid.UUID]] = {} + for old in locked: # по возрастанию id; advisory ГК и версии берутся первым вызовом (порядок D-9) + unbound[old.id] = await consolidation_documents.on_documents_superseded( + session, documents=documents_by_file.get(old.id, []), now=now + ) audit = AuditService(session) uploader = f"user:{uploader_id}" if uploader_id is not None else "неизвестен" @@ -541,7 +558,11 @@ async def supersede_previous( object_type=STORED_FILE_OBJECT_TYPE, object_id=old.id, user_id=None, - after_value={"file_id": new_file.id, "version_no": new_file.version_no}, + after_value={ + "file_id": new_file.id, + "version_no": new_file.version_no, + "unbound_version_ids": unbound[old.id], + }, description=( f"заменён файлом {new_file.original_name} (загрузил {uploader}); " f"документы: {', '.join(map(_describe_document, documents))}; " diff --git a/api/tests/test_api_consolidation_members.py b/api/tests/test_api_consolidation_members.py index c258bf8..8c887a2 100644 --- a/api/tests/test_api_consolidation_members.py +++ b/api/tests/test_api_consolidation_members.py @@ -1,9 +1,9 @@ """Периметр версии: `PUT /consolidations/versions/{id}/members` (spec consolidations-api «Периметр версии»; directory-policy «редактируемость версии»; audit-trail `MEMBERSHIP_CHANGE`; design D-5, D-8, D-9). -Подхват подтверждённых документов добавленных ЮЛ (`bind_confirmed`) до Task 6 — заглушка: здесь проверяется ВЫЗОВ -(шпион на функцию модуля), а настоящие привязки появятся с реализацией и ячейкой `CONFIRMED` (Task 6, Task 8). -Отвязка документов удалённых ЮЛ (`unbind_entity`) работает уже сейчас — прямое удаление `consolidation_document`. +Подхват подтверждённых документов добавленных ЮЛ (`bind_confirmed`): шпион проверяет, для кого он вызван, а +`test_add_member_really_binds_confirmed_documents` — настоящую привязку (Task 6; ячейка `CONFIRMED` — Task 8). +Отвязка документов удалённых ЮЛ — `unbind_entity`. """ import uuid @@ -16,7 +16,12 @@ from sqlalchemy.ext.asyncio import AsyncSession from src.models.access import AppUser from src.models.audit_ex import AuditEvent -from src.models.consolidation import ConsolidationDocument, ConsolidationMember, ConsolidationVersion +from src.models.consolidation import ( + Consolidation, + ConsolidationDocument, + ConsolidationMember, + ConsolidationVersion, +) from src.models.directories import CompanyGroup, LegalEntity, RegionalOffice, RelatedGroupMember from src.models.enums import AuditAction, AuditSection, ConsolidationStatus, PeriodQuarter from src.services import consolidation_documents @@ -125,8 +130,14 @@ async def test_add_member_updates_perimeter_and_picks_up_documents( ) calls: list[tuple[uuid.UUID, list[uuid.UUID]]] = [] - async def spy(session: AsyncSession, bound_version: ConsolidationVersion, entity_ids: Sequence[uuid.UUID]) -> int: - calls.append((bound_version.id, list(entity_ids))) + async def spy( + session: AsyncSession, + *, + version: ConsolidationVersion, + consolidation: Consolidation, + entity_ids: Sequence[uuid.UUID], + ) -> int: + calls.append((version.id, list(entity_ids))) return 0 monkeypatch.setattr(consolidation_documents, "bind_confirmed", spy) @@ -142,6 +153,40 @@ async def test_add_member_updates_perimeter_and_picks_up_documents( assert await _member_ids(seeded, version.id) == {e.id for e in [*inside, outside[0]]} +@pytest.mark.integration +async def test_add_member_really_binds_confirmed_documents( + auth_client: AuthClientFactory, seeded: AsyncSession +) -> None: + """Добавленное ЮЛ с подтверждённой ОСВ за квартал версии: `consolidation_document` создан без шпионов; статус + версии подхват не меняет; подтверждённая ОСВ другого года не подхватывается.""" + version, inside, outside, _ = await _version(seeded) + fitting = await make_confirmed_osv( + seeded, + entity=outside[0], + account=ACCOUNT, + year=2024, + quarter=PeriodQuarter.Q3, + is_cumulative=True, + rows=[{"name": "ООО Контрагент", "inn": "7701234567", "closing_debit": 10}], + ) + await make_confirmed_osv( + seeded, + entity=outside[0], + account=ACCOUNT, + year=2023, + quarter=PeriodQuarter.Q3, + is_cumulative=True, + rows=[{"name": "ООО Контрагент", "inn": "7701234567", "closing_debit": 10}], + ) + client = await auth_client(KA_RF) + + response = await client.put(_url(version.id), json=_body([*inside, outside[0]])) + + assert response.status_code == HTTP_OK, response.text + assert await _bound_documents(seeded, version.id) == {fitting.id} + assert response.json()["status"] == ConsolidationStatus.DATA_CORRECTION.value + + @pytest.mark.integration async def test_unchanged_members_are_not_touched_by_picking_up( auth_client: AuthClientFactory, seeded: AsyncSession, monkeypatch: pytest.MonkeyPatch @@ -150,7 +195,13 @@ async def test_unchanged_members_are_not_touched_by_picking_up( version, inside, _, _ = await _version(seeded, members=THREE, spare=0) calls: list[list[uuid.UUID]] = [] - async def spy(session: AsyncSession, bound_version: ConsolidationVersion, entity_ids: Sequence[uuid.UUID]) -> int: + async def spy( + session: AsyncSession, + *, + version: ConsolidationVersion, + consolidation: Consolidation, + entity_ids: Sequence[uuid.UUID], + ) -> int: calls.append(list(entity_ids)) return 0 @@ -216,7 +267,7 @@ async def test_removing_member_drops_its_bindings_only(auth_client: AuthClientFa @pytest.mark.integration async def test_unbind_entity_removes_only_that_entitys_documents_of_that_version(seeded: AsyncSession) -> None: - """`unbind_entity` (Task 6 заменит тело, имя и сигнатура — прежние) возвращает число снятых привязок.""" + """`unbind_entity` возвращает число снятых привязок.""" version, inside, _, _ = await _version(seeded, locked_by=None) docs = [] for entity in inside: diff --git a/api/tests/test_consolidation_documents.py b/api/tests/test_consolidation_documents.py new file mode 100644 index 0000000..66964ff --- /dev/null +++ b/api/tests/test_consolidation_documents.py @@ -0,0 +1,681 @@ +"""Привязка документов к версиям консолидации (spec documents-api «Подтверждение документа», «Подхват документов при +создании версии и расширении периметра»; audit-trail `CONSOLIDATION_STATUS_CHANGE`; design D-5, D-6, D-9; рулинги +R-1 — ключ замены с признаком нарастающего, R-2 — гонки потерянной привязки, R-3 — `FOR NO KEY UPDATE` документа). + +Сценарии подтверждения и подхвата — поверх `seeded`/`auth_client` (транзакция теста). Гонки — на `committed_*` +(настоящие параллельные транзакции, таймаут 5 с): подтверждение ∥ правка периметра в обе стороны, создание версии ∥ +подтверждение в обе стороны и совместимость блокировки документа подтверждения со вставкой привязки. +""" + +import asyncio +import datetime +import uuid +from collections.abc import Sequence +from typing import Any + +import pytest +from sqlalchemy import select, text +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from src.common.clock import utc_now +from src.models.access import AppUser +from src.models.audit_ex import AuditEvent +from src.models.consolidation import ( + Consolidation, + ConsolidationDocument, + ConsolidationMember, + ConsolidationStatusHistory, + ConsolidationVersion, +) +from src.models.directories import CompanyGroup, LegalEntity, RegionalOffice, RelatedGroup +from src.models.enums import ( + AuditAction, + ConsolidationStatus, + DocumentStatus, + DocumentType, + FileStatus, + PeriodQuarter, + RecognitionStatus, +) +from src.models.files import Document +from src.repositories import consolidations as repo +from src.repositories import files as files_repo +from src.services import consolidation_documents +from src.services.audit import AuditService +from src.services.consolidation_documents import fits +from tests.conftest import AuthClientFactory +from tests.factories import ( + make_company_group, + make_confirmed_osv, + make_consolidation, + make_document, + make_legal_entity, + make_lock, + make_membership, + make_related_group, + make_stored_file, +) + +HTTP_OK = 200 +HTTP_CREATED = 201 +ROSTOV = "Ростовский региональный филиал" +KA_RF = "ka_rf@armka.local" +ACCOUNT = "60" +YEAR = 2024 +LOCK_WAIT_SECONDS = 5 +ROWS = [{"name": "ООО Контрагент", "inn": "7701234567", "closing_debit": 10}] +Q1, Q2, Q3, Q4 = PeriodQuarter.Q1, PeriodQuarter.Q2, PeriodQuarter.Q3, PeriodQuarter.Q4 + + +# --- fits: чистая функция (D-5) -------------------------------------------------------------------------------------- + +_MEMBER = uuid.uuid4() +_STRANGER = uuid.uuid4() + + +def _pure_document( + *, entity: uuid.UUID = _MEMBER, year: int = YEAR, quarter: PeriodQuarter | None = Q3, cumulative: bool | None +) -> Document: + return Document( + legal_entity_id=entity, + document_type=DocumentType.OSV, + osv_account_code=ACCOUNT, + period_year=year, + period_quarter=quarter, + is_cumulative=cumulative, + status=DocumentStatus.CONFIRMED, + ) + + +@pytest.mark.parametrize( + ("document", "expected"), + [ + (_pure_document(cumulative=False, quarter=Q1), True), # член, квартальный Q1 ≤ Q3 + (_pure_document(entity=_STRANGER, cumulative=True), False), # не член периметра + (_pure_document(year=YEAR - 1, cumulative=True), False), # другой год + (_pure_document(cumulative=True, quarter=Q3), True), # нарастающий за квартал версии + (_pure_document(cumulative=True, quarter=Q2), False), # нарастающий за Q2 при версии Q3 (Review Focus 2) + (_pure_document(cumulative=False, quarter=Q4), False), # квартальный позже квартала версии + (_pure_document(cumulative=False, quarter=None), False), # квартал не определён + (_pure_document(cumulative=None, quarter=Q3), True), # ББ/ОФР: NULL — нарастающий (решение №10) + (_pure_document(cumulative=None, quarter=Q2), False), # NULL за Q2 при версии Q3 — как нарастающий + ], +) +def test_fits(document: Document, expected: bool) -> None: + version = ConsolidationVersion(status=ConsolidationStatus.INITIATED) + consolidation = Consolidation(year=YEAR, quarter=Q3) + + assert fits(document, version, consolidation, {_MEMBER}) is expected + + +# --- данные --------------------------------------------------------------------------------------------------------- + + +async def _user(session: AsyncSession, email: str = KA_RF) -> AppUser: + return (await session.execute(select(AppUser).where(AppUser.email == email))).scalar_one() + + +async def _group(session: AsyncSession, members: int = 2) -> tuple[CompanyGroup, list[LegalEntity]]: + office = (await session.execute(select(RegionalOffice).where(RegionalOffice.name == ROSTOV))).scalar_one() + group = await make_company_group(session, name=f"ГК {uuid.uuid4().hex[:6]}", office=office) + entities = [await make_legal_entity(session, name=f"ООО Член {n}", group=group) for n in range(members)] + return group, entities + + +async def _version( + session: AsyncSession, + group: CompanyGroup, + members: Sequence[LegalEntity], + *, + quarter: PeriodQuarter = Q3, + status: ConsolidationStatus = ConsolidationStatus.INITIATED, +) -> ConsolidationVersion: + _, version = await make_consolidation( + session, group=group, year=YEAR, quarter=quarter, members=members, initiator=await _user(session), status=status + ) + return version + + +async def _document( # noqa: PLR0913 - тестовая фабрика: ЮЛ, счёт, период и статусы документа и файла + session: AsyncSession, + entity: LegalEntity, + *, + quarter: PeriodQuarter = Q2, + cumulative: bool = False, + account: str = ACCOUNT, + status: DocumentStatus = DocumentStatus.NEEDS_CHECK, +) -> Document: + """Документ ОСВ файла `PROCESSED` (по умолчанию — квартальный `NEEDS_CHECK` за 2024 Q2).""" + file, job = await make_stored_file( + session, entity=entity, status=FileStatus.PROCESSED, job_status=RecognitionStatus.done + ) + return await make_document( + session, + file=file, + job=job, + account=account, + year=YEAR, + quarter=quarter, + status=status, + is_cumulative=cumulative, + ) + + +async def _confirmed( # noqa: PLR0913 - тестовая фабрика: ЮЛ, период, счёт и время подтверждения + session: AsyncSession, + entity: LegalEntity, + *, + quarter: PeriodQuarter, + cumulative: bool = False, + account: str = ACCOUNT, + confirmed_at: datetime.datetime | None = None, +) -> Document: + document = await make_confirmed_osv( + session, entity=entity, account=account, year=YEAR, quarter=quarter, is_cumulative=cumulative, rows=ROWS + ) + if confirmed_at is not None: + document.confirmed_at = confirmed_at + await session.flush() + return document + + +async def _bound(session: AsyncSession, version_id: uuid.UUID) -> set[uuid.UUID]: + stmt = select(ConsolidationDocument.document_id).where(ConsolidationDocument.consolidation_version_id == version_id) + return set((await session.execute(stmt)).scalars()) + + +async def _status(session: AsyncSession, version_id: uuid.UUID) -> ConsolidationStatus: + stmt = select(ConsolidationVersion.status).where(ConsolidationVersion.id == version_id) + return (await session.execute(stmt.execution_options(populate_existing=True))).scalar_one() + + +async def _history(session: AsyncSession, version_id: uuid.UUID) -> list[ConsolidationStatusHistory]: + return await repo.history(session, version_id) + + +async def _bind(session: AsyncSession, version: ConsolidationVersion, document: Document) -> None: + session.add(ConsolidationDocument(consolidation_version_id=version.id, document_id=document.id)) + await session.flush() + + +# --- подтверждение ---------------------------------------------------------------------------------------------------- + + +@pytest.mark.integration +async def test_confirm_binds_document_and_moves_version(seeded: AsyncSession, auth_client: AuthClientFactory) -> None: + """«Привязка к версии и переход статуса»: квартальная ОСВ 60 за Q2 при версии Q3 в `INITIATED`.""" + group, entities = await _group(seeded) + version = await _version(seeded, group, entities) + document = await _document(seeded, entities[0]) + user = await _user(seeded) + client = await auth_client(KA_RF) + + response = await client.post(f"/documents/{document.id}/confirm") + + assert response.status_code == HTTP_OK, response.text + assert await _bound(seeded, version.id) == {document.id} + assert await _status(seeded, version.id) is ConsolidationStatus.DATA_CORRECTION + last = (await _history(seeded, version.id))[-1] + assert (last.from_status, last.to_status, last.changed_by) == ( + ConsolidationStatus.INITIATED, + ConsolidationStatus.DATA_CORRECTION, + user.id, + ) + [event] = ( + await seeded.execute(select(AuditEvent).where(AuditEvent.action == AuditAction.CONSOLIDATION_STATUS_CHANGE)) + ).scalars() + assert (event.user_id, event.consolidation_version_id, event.object_id) == (user.id, version.id, version.id) + assert (event.before_value, event.after_value) == ({"status": "INITIATED"}, {"status": "DATA_CORRECTION"}) + + +@pytest.mark.integration +@pytest.mark.parametrize( + ("status", "moved"), + [ + (ConsolidationStatus.DOCUMENTS_PROCESSING, True), + (ConsolidationStatus.DATA_CORRECTION, False), # уже там: привязка без перехода + ], +) +async def test_confirm_binds_in_other_in_work_statuses( + seeded: AsyncSession, auth_client: AuthClientFactory, status: ConsolidationStatus, moved: bool +) -> None: + group, entities = await _group(seeded) + version = await _version(seeded, group, entities, status=status) + before = len(await _history(seeded, version.id)) + document = await _document(seeded, entities[0], quarter=Q3, cumulative=True) + client = await auth_client(KA_RF) + + response = await client.post(f"/documents/{document.id}/confirm") + + assert response.status_code == HTTP_OK, response.text + assert await _bound(seeded, version.id) == {document.id} + assert await _status(seeded, version.id) is ConsolidationStatus.DATA_CORRECTION + assert len(await _history(seeded, version.id)) == before + (1 if moved else 0) + + +@pytest.mark.integration +async def test_confirm_does_not_bind_cumulative_of_earlier_quarter( + seeded: AsyncSession, auth_client: AuthClientFactory +) -> None: + """«Не подходит по периоду»: нарастающая ОСВ 60 за Q2 при версии Q3 — привязки нет, статус прежний.""" + group, entities = await _group(seeded) + version = await _version(seeded, group, entities) + document = await _document(seeded, entities[0], quarter=Q2, cumulative=True) + client = await auth_client(KA_RF) + + response = await client.post(f"/documents/{document.id}/confirm") + + assert response.status_code == HTTP_OK, response.text + assert await _bound(seeded, version.id) == set() + assert await _status(seeded, version.id) is ConsolidationStatus.INITIATED + assert len(await _history(seeded, version.id)) == 1 + + +@pytest.mark.integration +async def test_confirm_does_not_bind_cancelled_version(seeded: AsyncSession, auth_client: AuthClientFactory) -> None: + """«Версия в CANCELLED»: подходящий документ не привязывается.""" + group, entities = await _group(seeded) + version = await _version(seeded, group, entities, status=ConsolidationStatus.CANCELLED) + document = await _document(seeded, entities[0]) + client = await auth_client(KA_RF) + + response = await client.post(f"/documents/{document.id}/confirm") + + assert response.status_code == HTTP_OK, response.text + assert await _bound(seeded, version.id) == set() + assert await _status(seeded, version.id) is ConsolidationStatus.CANCELLED + + +@pytest.mark.integration +async def test_confirm_does_not_bind_entity_outside_perimeter( + seeded: AsyncSession, auth_client: AuthClientFactory +) -> None: + """ЮЛ той же ГК, но вне периметра версии: привязки нет (блокируется надмножество, привязывается только член).""" + group, entities = await _group(seeded, members=3) + version = await _version(seeded, group, entities[:2]) + document = await _document(seeded, entities[2]) + client = await auth_client(KA_RF) + + response = await client.post(f"/documents/{document.id}/confirm") + + assert response.status_code == HTTP_OK, response.text + assert await _bound(seeded, version.id) == set() + assert await _status(seeded, version.id) is ConsolidationStatus.INITIATED + + +@pytest.mark.integration +async def test_confirm_replaces_bound_document_with_same_key( + seeded: AsyncSession, auth_client: AuthClientFactory +) -> None: + """Привязанный документ того же ЮЛ с той же тройкой и тем же признаком нарастающего отвязывается (D-5); нарастающий + той же пары (ЮЛ, счёт) за тот же квартал остаётся (R-1); документ другого ЮЛ не трогается.""" + group, entities = await _group(seeded) + version = await _version(seeded, group, entities, status=ConsolidationStatus.DATA_CORRECTION) + old_quarterly = await _confirmed(seeded, entities[0], quarter=Q3) + cumulative = await _confirmed(seeded, entities[0], quarter=Q3, cumulative=True) + other_entity = await _confirmed(seeded, entities[1], quarter=Q3) + for document in (old_quarterly, cumulative, other_entity): + await _bind(seeded, version, document) + new_quarterly = await _document(seeded, entities[0], quarter=Q3) + client = await auth_client(KA_RF) + + response = await client.post(f"/documents/{new_quarterly.id}/confirm") + + assert response.status_code == HTTP_OK, response.text + assert await _bound(seeded, version.id) == {new_quarterly.id, cumulative.id, other_entity.id} + + +@pytest.mark.integration +async def test_one_document_binds_to_two_versions(seeded: AsyncSession, auth_client: AuthClientFactory) -> None: + """Review Focus 1: консолидации ГК за Q2 и Q3 одного года; квартальная ОСВ Q1 подходит обеим — подтверждение + привязывает её к обеим и переводит обе в `DATA_CORRECTION`.""" + group, entities = await _group(seeded) + q2 = await _version(seeded, group, entities, quarter=Q2) + q3 = await _version(seeded, group, entities, quarter=Q3) + document = await _document(seeded, entities[0], quarter=Q1) + client = await auth_client(KA_RF) + + response = await client.post(f"/documents/{document.id}/confirm") + + assert response.status_code == HTTP_OK, response.text + for version in (q2, q3): + assert await _bound(seeded, version.id) == {document.id} + assert await _status(seeded, version.id) is ConsolidationStatus.DATA_CORRECTION + events = ( + await seeded.execute( + select(AuditEvent.consolidation_version_id).where( + AuditEvent.action == AuditAction.CONSOLIDATION_STATUS_CHANGE + ) + ) + ).scalars() + assert sorted(events) == sorted([q2.id, q3.id]) + + +@pytest.mark.integration +async def test_on_document_confirmed_returns_bound_versions(seeded: AsyncSession) -> None: + """Хук возвращает id версий, получивших привязку (по возрастанию `id`); неподходящие и аннулированные — нет.""" + group, entities = await _group(seeded) + q2 = await _version(seeded, group, entities, quarter=Q2) + q3 = await _version(seeded, group, entities, quarter=Q3) + q1 = await _version(seeded, group, entities, quarter=Q1) # квартальный Q2 не подходит версии Q1 + q4 = await _version(seeded, group, entities, quarter=Q4, status=ConsolidationStatus.CANCELLED) + document = await _document(seeded, entities[0], quarter=Q2, status=DocumentStatus.CONFIRMED) + user = await _user(seeded) + + bound = await consolidation_documents.on_document_confirmed( + seeded, AuditService(seeded), document=document, user_id=user.id, now=utc_now() + ) + + assert bound == sorted([q2.id, q3.id]) + assert await _bound(seeded, q1.id) == set() + assert await _bound(seeded, q4.id) == set() + + +# --- подхват ---------------------------------------------------------------------------------------------------------- + + +async def _gsk(session: AsyncSession, group: CompanyGroup, entities: Sequence[LegalEntity]) -> RelatedGroup: + gsk = await make_related_group(session, group=group, name=f"ГСК {uuid.uuid4().hex[:6]}") + for entity in entities: + await make_membership(session, related_group=gsk, entity=entity) + return gsk + + +def _create_body(group: CompanyGroup, gsk: RelatedGroup, quarter: str = "Q3") -> dict[str, Any]: + return {"company_group_id": str(group.id), "related_group_id": str(gsk.id), "year": YEAR, "quarter": quarter} + + +@pytest.mark.integration +async def test_create_picks_up_documents_confirmed_before(seeded: AsyncSession, auth_client: AuthClientFactory) -> None: + """«Фикстуры, подтверждённые до консолидации»: ОСВ 60 за Q1, Q2, Q3 подтверждены, затем создана консолидация + за Q3 — три привязки, версия в `INITIATED` (ячейка чек-листа — Task 8).""" + group, entities = await _group(seeded) + gsk = await _gsk(seeded, group, entities) + documents = [await _confirmed(seeded, entities[0], quarter=quarter) for quarter in (Q1, Q2, Q3)] + client = await auth_client(KA_RF) + + response = await client.post("/consolidations", json=_create_body(group, gsk)) + + assert response.status_code == HTTP_CREATED, response.text + version_id = uuid.UUID(response.json()["id"]) + assert await _bound(seeded, version_id) == {document.id for document in documents} + assert await _status(seeded, version_id) is ConsolidationStatus.INITIATED + assert len(await _history(seeded, version_id)) == 1 + + +@pytest.mark.integration +async def test_create_picks_latest_confirmed_of_same_key(seeded: AsyncSession, auth_client: AuthClientFactory) -> None: + """«Две подтверждённые версии одного документа»: два незаменённых файла с `CONFIRMED` ОСВ 62 за Q3 — привязан + документ с более поздним `confirmed_at`.""" + group, entities = await _group(seeded) + gsk = await _gsk(seeded, group, entities) + now = utc_now() + await _confirmed(seeded, entities[0], quarter=Q3, account="62", confirmed_at=now - datetime.timedelta(hours=1)) + later = await _confirmed(seeded, entities[0], quarter=Q3, account="62", confirmed_at=now) + client = await auth_client(KA_RF) + + response = await client.post("/consolidations", json=_create_body(group, gsk)) + + assert response.status_code == HTTP_CREATED, response.text + assert await _bound(seeded, uuid.UUID(response.json()["id"])) == {later.id} + + +@pytest.mark.integration +async def test_bind_confirmed_filters_documents(seeded: AsyncSession) -> None: + """Подхват берёт только `CONFIRMED` документы незаменённых файлов, подходящие версии; квартальный и нарастающий + за один квартал — оба (R-1); статус версии не меняется; повторный подхват ничего не добавляет.""" + group, entities = await _group(seeded) + version = await _version(seeded, group, entities) + consolidation = await seeded.get(Consolidation, version.consolidation_id) + assert consolidation is not None + quarterly = await _confirmed(seeded, entities[0], quarter=Q3) + cumulative = await _confirmed(seeded, entities[0], quarter=Q3, cumulative=True) + await _confirmed(seeded, entities[0], quarter=Q2, cumulative=True) # нарастающий за Q2 — не подходит + await _document(seeded, entities[0], quarter=Q1) # не подтверждён + superseded = await _confirmed(seeded, entities[0], quarter=Q1) + await seeded.execute( + text("UPDATE stored_file SET status = 'SUPERSEDED' WHERE id = :id"), {"id": superseded.stored_file_id} + ) + + count = await consolidation_documents.bind_confirmed( + seeded, version=version, consolidation=consolidation, entity_ids=[entities[0].id] + ) + + assert count == 2 # noqa: PLR2004 - квартальный и нарастающий Q3 + assert await _bound(seeded, version.id) == {quarterly.id, cumulative.id} + assert await _status(seeded, version.id) is ConsolidationStatus.INITIATED + again = await consolidation_documents.bind_confirmed( + seeded, version=version, consolidation=consolidation, entity_ids=[entities[0].id] + ) + assert again == 0 + assert await _bound(seeded, version.id) == {quarterly.id, cumulative.id} + + +@pytest.mark.integration +async def test_bind_confirmed_replaces_older_binding_of_same_key(seeded: AsyncSession) -> None: + """Существующая привязка той же тройки заменяется документом с более поздним `confirmed_at`.""" + group, entities = await _group(seeded) + version = await _version(seeded, group, entities) + consolidation = await seeded.get(Consolidation, version.consolidation_id) + assert consolidation is not None + now = utc_now() + older = await _confirmed(seeded, entities[0], quarter=Q3, confirmed_at=now - datetime.timedelta(hours=1)) + newer = await _confirmed(seeded, entities[0], quarter=Q3, confirmed_at=now) + await _bind(seeded, version, older) + + count = await consolidation_documents.bind_confirmed( + seeded, version=version, consolidation=consolidation, entity_ids=[entities[0].id] + ) + + assert count == 1 + assert await _bound(seeded, version.id) == {newer.id} + + +# --- гонки (настоящие параллельные транзакции) ------------------------------------------------------------------------ + + +async def _wait_for_lock_waiter(maker: async_sessionmaker[AsyncSession]) -> None: + """Ждёт, пока какой-то сеанс БД встанет в ожидание блокировки (строки или advisory).""" + deadline = asyncio.get_running_loop().time() + LOCK_WAIT_SECONDS + while asyncio.get_running_loop().time() < deadline: + async with maker() as session: + waiting = ( + await session.execute( + text( + "SELECT count(*) FROM pg_stat_activity " + "WHERE datname = current_database() AND wait_event_type = 'Lock'" + ) + ) + ).scalar_one() + if waiting: + return + await asyncio.sleep(0.05) + raise AssertionError("запрос не встал в ожидание блокировки") + + +async def _committed_setup( + maker: async_sessionmaker[AsyncSession], *, members: int = 2, spare: int = 1, with_version: bool = True +) -> dict[str, Any]: + """ГК филиала: `members` ЮЛ в периметре версии 2024 Q3 (`INITIATED`, блокировка у ka_rf), `spare` ЮЛ ГК вне + периметра, ГСК из всех ЮЛ; документ `NEEDS_CHECK` квартальной ОСВ 60 за Q2 у первого запасного ЮЛ (или у + первого члена, если запасных нет).""" + async with maker() as session, session.begin(): + group, entities = await _group(session, members=members + spare) + gsk = await _gsk(session, group, entities) + version_id = None + if with_version: + version = await _version(session, group, entities[:members]) + await make_lock(session, version=version, user=await _user(session)) + version_id = version.id + owner = entities[members] if spare else entities[0] + document = await _document(session, owner) + return { + "group_id": group.id, + "gsk_id": gsk.id, + "entity_ids": [entity.id for entity in entities], + "version_id": version_id, + "document_id": document.id, + "file_id": document.stored_file_id, + "owner_id": owner.id, + "user_id": (await _user(session)).id, + } + + +async def _confirm_in(session: AsyncSession, data: dict[str, Any]) -> list[uuid.UUID]: + """Подтверждение, как в `DocumentService.confirm`: файл → документ → статус → хук привязки (транзакция открыта).""" + file = await files_repo.lock_file(session, data["file_id"]) + document = await files_repo.lock_document(session, data["document_id"]) + assert file is not None + assert document is not None + now = utc_now() + document.status = DocumentStatus.CONFIRMED + document.confirmed_by = data["user_id"] + document.confirmed_at = now + await session.flush() + return await consolidation_documents.on_document_confirmed( + session, AuditService(session), document=document, user_id=data["user_id"], now=now + ) + + +async def _committed_bound(maker: async_sessionmaker[AsyncSession], version_id: uuid.UUID) -> set[uuid.UUID]: + async with maker() as session: + return await _bound(session, version_id) + + +@pytest.mark.integration +async def test_confirm_waits_for_member_edit_and_binds_new_member( + committed_sessionmaker: async_sessionmaker[AsyncSession], committed_client: AuthClientFactory +) -> None: + """R-2: правка периметра (версия `FOR UPDATE`, добавлен новый член) ∥ подтверждение документа нового члена. + + Подтверждение блокирует надмножество версий ГК, ждёт правку и после её commit видит нового члена — привязка не + теряется, версия в `DATA_CORRECTION`.""" + maker = committed_sessionmaker + data = await _committed_setup(maker) + client = await committed_client(KA_RF) + + async with maker() as editor, editor.begin(): + assert await repo.lock_version(editor, data["version_id"]) is not None + editor.add(ConsolidationMember(consolidation_version_id=data["version_id"], legal_entity_id=data["owner_id"])) + await editor.flush() + confirm = asyncio.create_task(client.post(f"/documents/{data['document_id']}/confirm")) + await _wait_for_lock_waiter(maker) + assert not confirm.done() + response = await asyncio.wait_for(confirm, LOCK_WAIT_SECONDS) + + assert response.status_code == HTTP_OK, response.text + assert await _committed_bound(maker, data["version_id"]) == {data["document_id"]} + async with maker() as session: + assert await _status(session, data["version_id"]) is ConsolidationStatus.DATA_CORRECTION + + +@pytest.mark.integration +async def test_member_edit_waits_for_confirm_and_picks_document_up( + committed_sessionmaker: async_sessionmaker[AsyncSession], committed_client: AuthClientFactory +) -> None: + """R-2/R-3, обратный порядок: подтверждение держит файл, документ и версии ГК; `PUT …/members`, добавляющий ЮЛ + документа, ждёт версию, после commit подтверждения подхватывает уже `CONFIRMED` документ. Взаимоблокировки нет.""" + maker = committed_sessionmaker + data = await _committed_setup(maker) + client = await committed_client(KA_RF) + body = {"legal_entity_ids": [str(entity_id) for entity_id in data["entity_ids"]]} + + async with maker() as confirmer, confirmer.begin(): + assert await _confirm_in(confirmer, data) == [] # ЮЛ ещё не член — привязывать не к чему + edit = asyncio.create_task(client.put(f"/consolidations/versions/{data['version_id']}/members", json=body)) + await _wait_for_lock_waiter(maker) + assert not edit.done() + response = await asyncio.wait_for(edit, LOCK_WAIT_SECONDS) + + assert response.status_code == HTTP_OK, response.text + assert await _committed_bound(maker, data["version_id"]) == {data["document_id"]} + async with maker() as session: # подхват статус не меняет + assert await _status(session, data["version_id"]) is ConsolidationStatus.INITIATED + + +@pytest.mark.integration +async def test_document_lock_does_not_block_binding_insert( + committed_sessionmaker: async_sessionmaker[AsyncSession], +) -> None: + """R-3: подтверждение держит документ (`lock_document`) и затем ждёт версию; подхват держит версию и вставляет + привязку на этот документ (FK берёт `FOR KEY SHARE` документа). С `FOR UPDATE` документа это взаимоблокировка; + с `FOR NO KEY UPDATE` вставка проходит, обе транзакции завершаются за таймаут.""" + maker = committed_sessionmaker + data = await _committed_setup(maker, spare=0) + async with maker() as session, session.begin(): + document = await session.get(Document, data["document_id"]) + assert document is not None + document.status = DocumentStatus.CONFIRMED + document.confirmed_at = utc_now() + + version_held = asyncio.Event() + + async def binder() -> None: + async with maker() as session, session.begin(): + assert await repo.lock_version(session, data["version_id"]) is not None + version_held.set() + session.add( + ConsolidationDocument(consolidation_version_id=data["version_id"], document_id=data["document_id"]) + ) + await session.flush() # FK: FOR KEY SHARE документа, который держит «подтверждение» + + async with maker() as confirmer, confirmer.begin(): + assert await files_repo.lock_file(confirmer, data["file_id"]) is not None + assert await files_repo.lock_document(confirmer, data["document_id"]) is not None + binding = asyncio.create_task(binder()) + await asyncio.wait_for(version_held.wait(), LOCK_WAIT_SECONDS) + # «подтверждение» ждёт версию, которую держит подхват; с FOR UPDATE документа — взаимоблокировка + assert await asyncio.wait_for(repo.lock_version(confirmer, data["version_id"]), LOCK_WAIT_SECONDS) + await asyncio.wait_for(binding, LOCK_WAIT_SECONDS) + + assert await _committed_bound(maker, data["version_id"]) == {data["document_id"]} + + +@pytest.mark.integration +async def test_create_waits_for_confirm_and_picks_document_up( + committed_sessionmaker: async_sessionmaker[AsyncSession], committed_client: AuthClientFactory +) -> None: + """R-2, создание версии ∥ подтверждение: подтверждение держит advisory-блокировку ГК; `POST /consolidations` ждёт + её и после commit подхватывает уже `CONFIRMED` документ.""" + maker = committed_sessionmaker + data = await _committed_setup(maker, spare=0, with_version=False) + client = await committed_client(KA_RF) + body = {"company_group_id": str(data["group_id"]), "related_group_id": str(data["gsk_id"]), "year": YEAR} + + async with maker() as confirmer, confirmer.begin(): + assert await _confirm_in(confirmer, data) == [] + create = asyncio.create_task(client.post("/consolidations", json={**body, "quarter": "Q3"})) + await _wait_for_lock_waiter(maker) + assert not create.done() + response = await asyncio.wait_for(create, LOCK_WAIT_SECONDS) + + assert response.status_code == HTTP_CREATED, response.text + assert await _committed_bound(maker, uuid.UUID(response.json()["id"])) == {data["document_id"]} + + +@pytest.mark.integration +async def test_confirm_waits_for_create_and_binds_new_version( + committed_sessionmaker: async_sessionmaker[AsyncSession], committed_client: AuthClientFactory +) -> None: + """R-2, подтверждение ∥ создание версии: создание держит advisory-блокировку ГК (новая версия до commit невидима); + подтверждение ждёт её и после commit привязывает документ к новой версии и двигает её статус.""" + maker = committed_sessionmaker + data = await _committed_setup(maker, spare=0, with_version=False) + client = await committed_client(KA_RF) + + async with maker() as creator, creator.begin(): + await repo.lock_group_bindings(creator, data["group_id"]) # как `ConsolidationService.create` + group = await creator.get(CompanyGroup, data["group_id"]) + assert group is not None + entities = [await creator.get(LegalEntity, entity_id) for entity_id in data["entity_ids"]] + version = await _version(creator, group, [entity for entity in entities if entity is not None]) + version_id = version.id + confirm = asyncio.create_task(client.post(f"/documents/{data['document_id']}/confirm")) + await _wait_for_lock_waiter(maker) + assert not confirm.done() + response = await asyncio.wait_for(confirm, LOCK_WAIT_SECONDS) + + assert response.status_code == HTTP_OK, response.text + assert await _committed_bound(maker, version_id) == {data["document_id"]} + async with maker() as session: + assert await _status(session, version_id) is ConsolidationStatus.DATA_CORRECTION diff --git a/api/tests/test_result_mapping.py b/api/tests/test_result_mapping.py index 47c87b1..9798bef 100644 --- a/api/tests/test_result_mapping.py +++ b/api/tests/test_result_mapping.py @@ -29,10 +29,12 @@ from src.core.config import Settings, get_settings from src.models.access import AppUser from src.models.accounting import ChartOfAccount from src.models.audit_ex import AuditEvent +from src.models.consolidation import ConsolidationDocument, ConsolidationVersion from src.models.directories import LegalEntity from src.models.enums import ( AuditAction, AuditSection, + ConsolidationStatus, CounterpartyMatchKind, DocumentCheckResult, DocumentStatus, @@ -59,7 +61,15 @@ from src.services.result_mapping import ( validate_result_key, ) from tests.conftest import FakeStorage -from tests.factories import make_document, make_legal_entity, make_result_object, make_stored_file +from tests.factories import ( + make_company_group, + make_consolidation, + make_document, + make_legal_entity, + make_result_object, + make_stored_file, + make_version, +) from tests.test_recognition_inbox import envelope, fake_msg API_DIR = Path(__file__).resolve().parents[1] @@ -915,7 +925,7 @@ async def test_new_osv_60_supersedes_previous_file( assert event.user_id is None assert event.section is AuditSection.PRIMARY_DOCUMENTS assert (event.object_type, event.object_id) == ("stored_file", old_id) - assert event.after_value == {"file_id": str(new_id), "version_no": 2} + assert event.after_value == {"file_id": str(new_id), "version_no": 2, "unbound_version_ids": []} assert event.description is not None assert f"user:{uploader}" in event.description assert f"OSV 60 2025 Q1 ({old_status.value})" in event.description @@ -1026,7 +1036,7 @@ async def test_version_chain(committed_sessionmaker: async_sessionmaker[AsyncSes assert v3.version_no == 3 # noqa: PLR2004 - третья версия [event] = await _replace_events(maker) assert event.object_id == v2_id - assert event.after_value == {"file_id": str(v3_id), "version_no": 3} + assert event.after_value == {"file_id": str(v3_id), "version_no": 3, "unbound_version_ids": []} async def test_one_event_per_replaced_file(committed_sessionmaker: async_sessionmaker[AsyncSession]): @@ -1096,7 +1106,110 @@ async def test_new_file_supersedes_several_files(committed_sessionmaker: async_s assert sorted(event.object_id for event in events) == ordered # type: ignore[type-var] for event in events: assert event.user_id is None - assert event.after_value == {"file_id": str(new_id), "version_no": expected_version} + assert event.after_value == { + "file_id": str(new_id), + "version_no": expected_version, + "unbound_version_ids": [], + } + + +async def test_replaced_document_is_unbound_from_versions_in_work( + committed_sessionmaker: async_sessionmaker[AsyncSession], +): + """primary-files «Заменённый документ был в версии»: `CONFIRMED` ОСВ 60 за 2024 Q2 привязана к версии 2024 Q3 в + `DATA_CORRECTION` и к аннулированной версии; автозамена отвязывает документ только от версии «в работе» (статус + прежний), `FILE_REPLACE.after_value.unbound_version_ids` — эта версия. Привязка другого документа заменённого + файла (ОСВ 62, ключ замены не совпал) тоже снимается: заменён весь файл.""" + maker = committed_sessionmaker + async with maker() as session, session.begin(): + initiator = (await session.execute(select(AppUser).order_by(AppUser.email).limit(1))).scalar_one() + group = await make_company_group(session, name="ГК Автозамена") + entity = await make_legal_entity(session, name=OWNER_NAME, inn="7700000000", group=group) + other = await make_legal_entity(session, name="ООО Второе", group=group) + consolidation, in_work = await make_consolidation( + session, + group=group, + year=2024, + quarter=PeriodQuarter.Q3, + members=[entity, other], + initiator=initiator, + status=ConsolidationStatus.DATA_CORRECTION, + ) + cancelled = await make_version( + session, + consolidation=consolidation, + version_no=2, + members=[entity, other], + initiator=initiator, + status=ConsolidationStatus.CANCELLED, + ) + old_file, old_job = await make_stored_file( + session, entity=entity, status=FileStatus.PROCESSED, job_status=RecognitionStatus.done + ) + old_60 = await make_document( + session, + file=old_file, + job=old_job, + account="60", + year=2024, + quarter=PeriodQuarter.Q2, + status=DocumentStatus.CONFIRMED, + is_cumulative=False, + ) + old_62 = await make_document( + session, + file=old_file, + job=old_job, + account="62", + year=2024, + quarter=PeriodQuarter.Q3, + status=DocumentStatus.CONFIRMED, + is_cumulative=True, + ) + for version in (in_work, cancelled): + for document in (old_60, old_62): + session.add(ConsolidationDocument(consolidation_version_id=version.id, document_id=document.id)) + new_file, new_job = await make_stored_file( + session, entity=entity, status=FileStatus.PROCESSED, job_status=RecognitionStatus.done + ) + new_60 = await make_document( + session, + file=new_file, + job=new_job, + account="60", + year=2024, + quarter=PeriodQuarter.Q2, + status=DocumentStatus.NEEDS_CHECK, + is_cumulative=False, + ) + ids = (in_work.id, cancelled.id, old_file.id, old_60.id, old_62.id) + + async with maker() as session, session.begin(): + new = await session.get(StoredFile, new_file.id) + assert new is not None + replaced = await supersede_previous(session, new_file=new, new_documents=[new_60], uploader_id=None, now=_now()) + in_work_id, cancelled_id, old_file_id, old_60_id, old_62_id = ids + + assert [file.id for file in replaced] == [old_file_id] + async with maker() as session: + + def bound(version_id: uuid.UUID) -> Any: + return select(ConsolidationDocument.document_id).where( + ConsolidationDocument.consolidation_version_id == version_id + ) + + assert set((await session.execute(bound(in_work_id))).scalars()) == set() + assert set((await session.execute(bound(cancelled_id))).scalars()) == {old_60_id, old_62_id} + statuses = dict( + (await session.execute(select(ConsolidationVersion.id, ConsolidationVersion.status))).tuples().all() + ) + assert statuses == { + in_work_id: ConsolidationStatus.DATA_CORRECTION, + cancelled_id: ConsolidationStatus.CANCELLED, + } + [event] = await _replace_events(maker) + assert event.object_id == old_file_id + assert event.after_value == {"file_id": str(new_file.id), "version_no": 2, "unbound_version_ids": [str(in_work_id)]} # --- audit-trail: обмен и маппинг не пишут audit_event (ASSUMPTION(OQ-55)) ---------------------------------------- diff --git a/docs/superpowers/plans/2026-10-06-slice-4-consolidation.md b/docs/superpowers/plans/2026-10-06-slice-4-consolidation.md index 971ddc1..6cde4b7 100644 --- a/docs/superpowers/plans/2026-10-06-slice-4-consolidation.md +++ b/docs/superpowers/plans/2026-10-06-slice-4-consolidation.md @@ -139,9 +139,9 @@ - `async def on_documents_superseded(session, *, documents: Sequence[Document], now) -> list[uuid.UUID]` — версии «в работе», где эти документы привязаны (`FOR UPDATE` по `id`), удалить привязки; возвращает `unbound_version_ids`. - `DocumentService.confirm`: после установки `CONFIRMED` и перед аудитом — `on_document_confirmed(...)`; `supersede_previous`: после `UPDATE stored_file` — `on_documents_superseded(replaced_documents всех файлов)`, `unbound_version_ids` в `after_value` `FILE_REPLACE`. -- [ ] **Step 1: Тесты.** Unit `fits`: таблица из 8 случаев (член/не член, год, нарастающий Q2/Q3 при версии Q3, квартальный Q1/Q4, `period_quarter` NULL, `is_cumulative` NULL). Integration: сценарии `documents-api` «Привязка к версии и переход статуса», «Не подходит по периоду», «Версия в CANCELLED», «Фикстуры, подтверждённые до консолидации» (через `make_confirmed_osv` × 3 + создание через API), «Две подтверждённые версии одного документа»; `primary-files` «Заменённый документ был в версии» (через `supersede_previous` напрямую, как в `test_result_mapping`); Review Focus 1 — две версии Q2 и Q3 получают один квартальный документ Q1 и обе переходят в `DATA_CORRECTION`; порядок блокировок: подтверждение ∥ `PUT …/members` на двух сессиях не взаимоблокируется (таймаут 5 с). -- [ ] **Step 2–4:** падает → реализация → `make check`, `pytest -m integration tests/test_consolidation_documents.py tests/test_api_documents.py tests/test_result_mapping.py tests/test_api_consolidations.py tests/test_api_consolidation_members.py -q` → PASS. -- [ ] **Step 5: Commit** `feat(consolidation): привязка документов к версиям при подтверждении, подхват и отвязка при автозамене`. +- [x] **Step 1: Тесты.** Unit `fits`: таблица из 8 случаев (член/не член, год, нарастающий Q2/Q3 при версии Q3, квартальный Q1/Q4, `period_quarter` NULL, `is_cumulative` NULL). Integration: сценарии `documents-api` «Привязка к версии и переход статуса», «Не подходит по периоду», «Версия в CANCELLED», «Фикстуры, подтверждённые до консолидации» (через `make_confirmed_osv` × 3 + создание через API), «Две подтверждённые версии одного документа»; `primary-files` «Заменённый документ был в версии» (через `supersede_previous` напрямую, как в `test_result_mapping`); Review Focus 1 — две версии Q2 и Q3 получают один квартальный документ Q1 и обе переходят в `DATA_CORRECTION`; порядок блокировок: подтверждение ∥ `PUT …/members` на двух сессиях не взаимоблокируется (таймаут 5 с). +- [x] **Step 2–4:** падает → реализация → `make check`, `pytest -m integration tests/test_consolidation_documents.py tests/test_api_documents.py tests/test_result_mapping.py tests/test_api_consolidations.py tests/test_api_consolidation_members.py -q` → PASS. +- [x] **Step 5: Commit** `feat(consolidation): привязка документов к версиям при подтверждении, подхват и отвязка при автозамене`. --- diff --git a/openspec/changes/slice-4-consolidation/design.md b/openspec/changes/slice-4-consolidation/design.md index c8f6429..b9e6fd5 100644 --- a/openspec/changes/slice-4-consolidation/design.md +++ b/openspec/changes/slice-4-consolidation/design.md @@ -214,3 +214,10 @@ legal_entity_name, account_code, missing_quarters[]}}`. Фильтры `GET …/ | 27 | 2026-10-06 | Члены, остающиеся в периметре, не пересоздаются (сохраняются `id` и `is_head_entity`); новые — `is_head_entity = false`. Непустая смена периметра выставляет `consolidation_version.updated_by/at` | признак головной организации — решение аналитика, смена состава его не сбрасывает; `updated_*` показывают последнюю правку версии (как у `transition`) | нулевая | | 28 | 2026-10-06 | Общий пролог операций над версией вынесен в `services/consolidation_guard.py`: `lock_visible_version` (`FOR UPDATE` → видимость → 404) и `lock_editable_version` (+ активная блокировка и `ensure_version_editable`); `VERSION_NOT_FOUND` живёт там и реэкспортируется из `services.consolidations`. `take`/`release` берут только `lock_visible_version` (редактируемость им не нужна). Поведение сервисов блокировки и статусов не менялось | закрывает T4.2 (три копии пролога; четвёртая — Task 11); порядок «версия → видимость → редактируемость» одинаков везде | правка пролога меняет все операции сразу (задумано) | | 29 | 2026-10-06 | До Task 6 отвязка ушедших ЮЛ — `consolidation_documents.unbind_entity(session, *, version_id, entity_id) -> int`: прямой `DELETE` строк `consolidation_document` документов этого ЮЛ в этой версии; `bind_confirmed` для добавленных — заглушка (вызывается, возвращает 0), тест подхвата проверяет вызов шпионом | план Task 5: до Task 6 отвязка прямым удалением; имена и сигнатуры зафиксированы за Task 6 | до Task 6 добавленное ЮЛ не получает привязок документов (ячейка `CONFIRMED` появится в Task 6/8) | +| 30 | 2026-10-06 | R-2, гонка «создание версии ∥ подтверждение/автозамена» закрыта `pg_advisory_xact_lock(hashtextextended('consolidation-group:<ГК>', 0))` (`repositories.consolidations.lock_group_bindings`): `ConsolidationService.create` берёт его сразу после проверки видимости ГК (до чтения ГСК, проверки дубля периода и подхвата), `lock_versions_for_entity` — первым шагом, если у ЮЛ есть ГК (ЮЛ без ГК не может быть членом версии). Полный порядок: файл → документ → advisory ГК → версии; автозамена — advisory ЮЛ → старые файлы → advisory ГК → версии; `PUT …/members` advisory не берёт. Тесты на двух сессиях в обе стороны (`test_create_waits_for_confirm_…`, `test_confirm_waits_for_create_…`), мутанты без advisory в любой из сторон их роняют | новая версия невидима до commit, `FOR UPDATE` версий её не ловит; advisory по ГК — минимальная сериализация, не трогающая правку периметра | создание консолидаций одной ГК (любых периодов) сериализуется с подтверждениями и автозаменами документов ЮЛ этой ГК — ожидание на время короткой транзакции; параллельные создания одного периода теперь получают 409 `CONSOLIDATION_EXISTS` проверкой, а не UNIQUE | +| 31 | 2026-10-06 | R-3: на `FOR NO KEY UPDATE` переведены только `lock_document` и `lock_file_documents`; `lock_file` (и `_lock_file` маппинга) остаётся `FOR UPDATE` | на `stored_file` нет FK из таблиц консолидации, в циклы привязки файл не входит; удаление файла (`DELETE stored_file`) под `FOR NO KEY UPDATE` повышало бы блокировку уже после захвата и могло бы встречно ждать `KEY SHARE` вставки задания | нулевая; если файл войдёт в цикл — та же правка `key_share=True` | +| 32 | 2026-10-06 | R-3 в текущих потоках не воспроизводится через API: подхват привязывает только документы, `CONFIRMED` в зафиксированном снимке, а подтверждение держит документ, который ещё не `CONFIRMED` (иначе сразу 409). Тест `test_document_lock_does_not_block_binding_insert` проверяет сам цикл блокировок на примитивах (`lock_document` + ожидание версии ∥ `lock_version` + вставка привязки): с `FOR UPDATE` — `DeadlockDetectedError`, с `FOR NO KEY UPDATE` — проходит. API-тесты «подтверждение ∥ `PUT …/members`» в обе стороны проверяют R-2 (ожидание и непотерянную привязку) | диспатч требовал мутант-проверку R-3; на уровне API цикла нет, но страховка дешёвая и защищает будущие пути (Task 7/11) | нулевая | +| 33 | 2026-10-06 | Автозамена снимает привязки ВСЕХ документов заменённого файла, а не только совпавших с новым по ключу автозамены | spec primary-files: «привязки документов заменённых файлов»; файл `SUPERSEDED` целиком, а подхват берёт документы лишь незаменённых файлов — оставшаяся привязка была бы «висячей» | документ заменённого файла, не повторённый в новом файле, выпадает из версии (пробел в чек-листе) — так и задумано спекой | +| 34 | 2026-10-06 | `FILE_REPLACE.after_value.unbound_version_ids` — версии, где сняты привязки документов ИМЕННО этого заменённого файла (по возрастанию `id`; пустой список, если не было); `on_documents_superseded` вызывается по одному разу на заменённый файл | событие пишется на каждый заменённый файл; общий список версий по всей автозамене приписал бы файлу чужие отвязки. Повторные вызовы берут ту же advisory ГК и те же версии (уже удержаны) — порядок блокировок не меняется | нулевая | +| 35 | 2026-10-06 | Выбор «самого позднего `confirmed_at`» при равенстве — по наибольшему `id`; `confirmed_at` NULL у `CONFIRMED` (только тестовые данные) считается самым ранним. `bind_confirmed` заменяет привязанный документ того же ключа победителем и не трогает уже привязанного победителя (повторный подхват — 0 новых привязок) | детерминизм; спека порядок при равенстве не задаёт | при одинаковом `confirmed_at` привязан «случайный по id» из двух — равноценных — документов | +| 36 | 2026-10-06 | Переход версии при подтверждении пишется `transition(...)` с `user_id` подтвердившего, но без IP/user-agent (сигнатура хука `on_document_confirmed(session, audit: AuditService, *, document, user_id, now)` фиксирована планом); `transition` импортируется внутри хука (цикл `consolidation_status → consolidations → consolidation_documents`) | audit-trail требует для автоматического перехода только `user_id` инициатора | в `CONSOLIDATION_STATUS_CHANGE` автоматического перехода нет IP/UA — они есть в соседнем `DOCUMENT_CONFIRM` той же транзакции | diff --git a/openspec/changes/slice-4-consolidation/tasks.md b/openspec/changes/slice-4-consolidation/tasks.md index e99661f..d705204 100644 --- a/openspec/changes/slice-4-consolidation/tasks.md +++ b/openspec/changes/slice-4-consolidation/tasks.md @@ -30,7 +30,7 @@ ## 3. Привязка документов и чек-лист -- [ ] 3.1 **Привязка.** `services/consolidation_documents.py:fits(document, version)`, `bind_confirmed(version, entity_ids)`, +- [x] 3.1 **Привязка.** `services/consolidation_documents.py:fits(document, version)`, `bind_confirmed(version, entity_ids)`, `unbind_entity`, `on_document_confirmed(document, user)`, `on_file_superseded(file, documents)`; правки `DocumentService.confirm` (версии `FOR UPDATE` после документа), `supersede_previous` (`unbound_version_ids` в аудите). Тесты: сценарии `documents-api` (привязка, период, `CANCELLED`, подхват, две версии документа), `primary-files`