feat(consolidation): переходы DOCUMENTS_PROCESSING и DATA_CORRECTION по загрузке и маппингу
Загрузка с consolidation_version_id: проверки до приёма (404/409/422), после приёма INITIATED -> DOCUMENTS_PROCESSING в отдельной транзакции без блокировки версии. Маппинг: on_documents_mapped переводит версии DOCUMENTS_PROCESSING в DATA_CORRECTION при подходящем документе NEEDS_CHECK (changed_by = uploaded_by), привязки не создаёт. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
This commit is contained in:
Andrey Epifancev
parent
0e1fbd8036
commit
1908288613
@ -20,7 +20,11 @@ def get_primary_files_service(
|
||||
app_settings: SettingsDep,
|
||||
) -> PrimaryFilesService:
|
||||
"""Сервис первичных файлов; S3 — `app.state.storage` (`None`, если S3 недоступно или не подключено)."""
|
||||
return PrimaryFilesService(session, audit, scope, getattr(request.app.state, "storage", None), app_settings)
|
||||
# отдельная короткая транзакция (переход версии после загрузки) — та же фабрика, что у аудита входа
|
||||
transaction = getattr(getattr(request.app.state, "session_audit", None), "transaction", None)
|
||||
return PrimaryFilesService(
|
||||
session, audit, scope, getattr(request.app.state, "storage", None), app_settings, transaction
|
||||
)
|
||||
|
||||
|
||||
PrimaryFilesServiceDep = Annotated[PrimaryFilesService, Depends(get_primary_files_service)]
|
||||
|
||||
@ -10,7 +10,7 @@ multipart: одно или несколько полей `files`; право `DO
|
||||
import uuid
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import APIRouter, Depends, File, HTTPException, Response, UploadFile, status
|
||||
from fastapi import APIRouter, Depends, File, HTTPException, Query, Response, UploadFile, status
|
||||
|
||||
from src.api.dependencies.auth import require
|
||||
from src.api.dependencies.files import PrimaryFilesServiceDep
|
||||
@ -30,8 +30,18 @@ router = APIRouter()
|
||||
status_code=status.HTTP_201_CREATED,
|
||||
dependencies=[Depends(require("DOCUMENT_EDIT"))],
|
||||
responses={
|
||||
status.HTTP_404_NOT_FOUND: {"model": ErrorDetail, "description": "ЮЛ нет или оно вне видимости"},
|
||||
status.HTTP_422_UNPROCESSABLE_CONTENT: {"model": UploadResult, "description": "ни один файл не принят"},
|
||||
status.HTTP_404_NOT_FOUND: {
|
||||
"model": ErrorDetail,
|
||||
"description": "ЮЛ нет или оно вне видимости; версия консолидации не найдена или чужая (VERSION_NOT_FOUND)",
|
||||
},
|
||||
status.HTTP_409_CONFLICT: {
|
||||
"model": ErrorDetail,
|
||||
"description": "версия консолидации не в работе (VERSION_NOT_EDITABLE)",
|
||||
},
|
||||
status.HTTP_422_UNPROCESSABLE_CONTENT: {
|
||||
"model": UploadResult,
|
||||
"description": "ни один файл не принят; ЮЛ вне периметра версии (ENTITY_NOT_IN_PERIMETER)",
|
||||
},
|
||||
status.HTTP_503_SERVICE_UNAVAILABLE: {"model": ErrorDetail, "description": "S3 не подключено"},
|
||||
},
|
||||
summary="Загрузка файлов организации",
|
||||
@ -41,9 +51,18 @@ async def upload_files(
|
||||
files: Annotated[list[UploadFile], File(description="Файлы ОСВ/ББ/ОФР: xls, xlsx, xlsb, pdf, png, jpg")],
|
||||
service: PrimaryFilesServiceDep,
|
||||
response: Response,
|
||||
consolidation_version_id: Annotated[
|
||||
uuid.UUID | None,
|
||||
Query(description="Версия консолидации, в контексте которой загружаются файлы (необязательно)"),
|
||||
] = None,
|
||||
) -> UploadResult:
|
||||
"""Принятые файлы (`file_id`, `job_id`, имя) и отклонённые (имя, код, пояснение, `file_id` дубля)."""
|
||||
"""Принятые файлы (`file_id`, `job_id`, имя) и отклонённые (имя, код, пояснение, `file_id` дубля).
|
||||
|
||||
С `consolidation_version_id`: до приёма — версия видима, «в работе», ЮЛ в периметре; после приёма хотя бы одного
|
||||
файла версия `INITIATED` → `DOCUMENTS_PROCESSING` (отдельная транзакция, блокировка версии не нужна)."""
|
||||
entity = await service.get_entity(legal_entity_id)
|
||||
if consolidation_version_id is not None:
|
||||
await service.check_version_for_upload(consolidation_version_id, entity)
|
||||
service.require_storage()
|
||||
accepted: list[AcceptedOut] = []
|
||||
rejected: list[RejectedOut] = []
|
||||
@ -69,4 +88,6 @@ async def upload_files(
|
||||
accepted.append(AcceptedOut(file_id=result.file_id, job_id=result.job_id, name=result.name))
|
||||
if not accepted:
|
||||
response.status_code = status.HTTP_422_UNPROCESSABLE_CONTENT
|
||||
elif consolidation_version_id is not None:
|
||||
await service.after_upload_to_version(consolidation_version_id)
|
||||
return UploadResult(accepted=accepted, rejected=rejected)
|
||||
|
||||
@ -22,6 +22,7 @@ import uuid
|
||||
from collections.abc import Collection, Iterable, Sequence
|
||||
from typing import Any
|
||||
|
||||
from loguru import logger
|
||||
from sqlalchemy import delete, select
|
||||
from sqlalchemy.engine import CursorResult
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
@ -33,7 +34,14 @@ 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"]
|
||||
__all__ = [
|
||||
"bind_confirmed",
|
||||
"fits",
|
||||
"on_document_confirmed",
|
||||
"on_documents_mapped",
|
||||
"on_documents_superseded",
|
||||
"unbind_entity",
|
||||
]
|
||||
|
||||
#: Статусы, из которых подтверждение привязанного документа переводит версию в `DATA_CORRECTION` (D-6).
|
||||
MOVES_TO_CORRECTION = frozenset({ConsolidationStatus.INITIATED, ConsolidationStatus.DOCUMENTS_PROCESSING})
|
||||
@ -248,3 +256,60 @@ async def on_documents_superseded(
|
||||
if await _unbind(session, version.id, document_ids):
|
||||
unbound.add(version.id)
|
||||
return sorted(unbound)
|
||||
|
||||
|
||||
async def on_documents_mapped(
|
||||
session: AsyncSession,
|
||||
*,
|
||||
file: StoredFile,
|
||||
documents: Sequence[Document],
|
||||
now: datetime.datetime,
|
||||
) -> list[uuid.UUID]:
|
||||
"""Хук маппинга (osv-result-mapping «Документ NEEDS_CHECK переводит версию в корректировку», D-6): каждая версия в
|
||||
`DOCUMENTS_PROCESSING`, в периметр которой входит ЮЛ файла, переходит в `DATA_CORRECTION`
|
||||
(`changed_by = file.uploaded_by`), если среди `documents` есть документ `NEEDS_CHECK`, подходящий версии (`fits`).
|
||||
Привязки `consolidation_document` НЕ создаются (их создаёт только подтверждение и подхват). Возвращает id версий,
|
||||
переведённых в `DATA_CORRECTION` (по возрастанию). Файл без `uploaded_by` — переход пропускается с записью в лог
|
||||
(`changed_by NOT NULL`, design №2).
|
||||
|
||||
Блокировки (R-2): `lock_versions_for_entity` берёт advisory ГК и блокирует НАДМНОЖЕСТВО версий «в работе» ГК
|
||||
(`IN_WORK_STATUSES`, а не только `DOCUMENTS_PROCESSING`): статус фильтруется ПОСЛЕ блокировки по свежему состоянию
|
||||
(`populate_existing`), поэтому версия, которую параллельно перевели в `DOCUMENTS_PROCESSING` или в которую
|
||||
параллельно добавили ЮЛ, не теряется. Вызывать в транзакции маппинга ПОСЛЕ `map_file` (файл и документы уже
|
||||
заблокированы, автозамена прежних версий уже взяла advisory ЮЛ → старые файлы → advisory ГК → версии): порядок
|
||||
«задание → файл → документы → версии» не нарушается, версии блокируются последними.
|
||||
"""
|
||||
if file.uploaded_by is None:
|
||||
logger.info(
|
||||
"Переход версий по маппингу пропущен: у файла нет uploaded_by (changed_by обязателен). file_id={file_id}",
|
||||
file_id=file.id,
|
||||
)
|
||||
return []
|
||||
needs_check = [document for document in documents if document.status is DocumentStatus.NEEDS_CHECK]
|
||||
if not needs_check:
|
||||
return []
|
||||
# цикл импортов: consolidation_status → consolidations → consolidation_documents (design №36)
|
||||
from src.services.consolidation_status import transition # noqa: PLC0415
|
||||
|
||||
audit = AuditService(session)
|
||||
moved: list[uuid.UUID] = []
|
||||
for version in await repo.lock_versions_for_entity(session, file.legal_entity_id, IN_WORK_STATUSES):
|
||||
if version.status is not ConsolidationStatus.DOCUMENTS_PROCESSING:
|
||||
continue
|
||||
consolidation = await session.get(Consolidation, version.consolidation_id)
|
||||
if consolidation is None: # FK гарантирует наличие
|
||||
continue
|
||||
members = await _member_ids(session, version.id)
|
||||
if not any(fits(document, version, consolidation, members) for document in needs_check):
|
||||
continue
|
||||
await transition(
|
||||
session,
|
||||
audit,
|
||||
version=version,
|
||||
to=ConsolidationStatus.DATA_CORRECTION,
|
||||
changed_by=file.uploaded_by,
|
||||
comment=None,
|
||||
now=now,
|
||||
)
|
||||
moved.append(version.id)
|
||||
return moved
|
||||
|
||||
@ -18,6 +18,8 @@ audit-trail «События файлов и документов»; design D-4)
|
||||
|
||||
import hashlib
|
||||
import uuid
|
||||
from collections.abc import Callable
|
||||
from contextlib import AbstractAsyncContextManager
|
||||
from dataclasses import dataclass
|
||||
|
||||
from loguru import logger
|
||||
@ -27,27 +29,32 @@ from starlette import status as http_status
|
||||
|
||||
from src.api.dependencies.audit import RequestAudit
|
||||
from src.common.clock import utc_now
|
||||
from src.common.exceptions import ConflictError, DomainError, NotFoundError
|
||||
from src.common.exceptions import ConflictError, DomainError, DomainValidationError, NotFoundError
|
||||
from src.common.filenames import safe_filename
|
||||
from src.contract import CONTRACT_VERSION, request_subject
|
||||
from src.core.config import Settings
|
||||
from src.core.storage import ObjectStorage
|
||||
from src.domain.file_types import MAX_ATTEMPTS, DetectedFile, detect_file
|
||||
from src.domain.policy import VisibilityScope
|
||||
from src.domain.policy import IN_WORK_STATUSES, VisibilityScope
|
||||
from src.models.consolidation import ConsolidationVersion
|
||||
from src.models.directories import LegalEntity
|
||||
from src.models.enums import (
|
||||
AuditAction,
|
||||
AuditSection,
|
||||
ConsolidationStatus,
|
||||
DocumentStatus,
|
||||
FileStatus,
|
||||
RecognitionStatus,
|
||||
S3BucketRole,
|
||||
)
|
||||
from src.models.files import RecognitionJob, S3Object, StoredFile
|
||||
from src.repositories import consolidations as consolidations_repo
|
||||
from src.repositories import directories as directories_repo
|
||||
from src.repositories import files as files_repo
|
||||
from src.schemas.common import PageParams
|
||||
from src.schemas.files import FileOut
|
||||
from src.services.audit import AuditService
|
||||
from src.services.consolidation_guard import VERSION_NOT_FOUND
|
||||
from src.services.documents import file_out
|
||||
from src.services.legal_entities import LEGAL_ENTITY_NOT_FOUND
|
||||
from src.services.result_mapping import RESULTS_PREFIX
|
||||
@ -62,6 +69,9 @@ FILE_NOT_FOUND = "файл не найден" # одинаково для чу
|
||||
FILE_HAS_CONFIRMED_DOCUMENTS = "FILE_HAS_CONFIRMED_DOCUMENTS"
|
||||
FILE_SUPERSEDES_OTHERS = "FILE_SUPERSEDES_OTHERS"
|
||||
|
||||
#: фабрика отдельной короткой транзакции (в проде — `src.db.db.session_transaction`): commit на выходе
|
||||
TransactionFactory = Callable[[], AbstractAsyncContextManager[AsyncSession]]
|
||||
|
||||
|
||||
class StorageUnavailableError(DomainError):
|
||||
"""503 `STORAGE_UNAVAILABLE`: S3 не подключено (dev без S3) или не приняло объект."""
|
||||
@ -108,19 +118,21 @@ def upload_key(file_id: uuid.UUID, job_id: uuid.UUID, name: str) -> str:
|
||||
class PrimaryFilesService:
|
||||
"""Загрузка первичных файлов ЮЛ от имени пользователя запроса (право `DOCUMENT_EDIT` — на маршруте)."""
|
||||
|
||||
def __init__(
|
||||
def __init__( # noqa: PLR0913, PLR0917 - зависимости сервиса; `transaction` — отдельная транзакция (D-6)
|
||||
self,
|
||||
session: AsyncSession,
|
||||
audit: RequestAudit,
|
||||
scope: VisibilityScope,
|
||||
storage: ObjectStorage | None,
|
||||
settings: Settings,
|
||||
transaction: TransactionFactory | None = None,
|
||||
) -> None:
|
||||
self._session = session
|
||||
self._audit = audit
|
||||
self._scope = scope
|
||||
self._storage = storage
|
||||
self._settings = settings
|
||||
self._transaction = transaction
|
||||
|
||||
@property
|
||||
def max_upload_bytes(self) -> int:
|
||||
@ -134,6 +146,53 @@ class PrimaryFilesService:
|
||||
raise NotFoundError(LEGAL_ENTITY_NOT_FOUND)
|
||||
return entity
|
||||
|
||||
async def check_version_for_upload(self, version_id: uuid.UUID, entity: LegalEntity) -> ConsolidationVersion:
|
||||
"""Загрузка в контексте версии консолидации (spec primary-files «Загрузка файлов организации», D-6): ДО приёма
|
||||
файлов версия видима (иначе 404 `VERSION_NOT_FOUND`) → статус «в работе» (иначе 409 `VERSION_NOT_EDITABLE`) →
|
||||
ЮЛ — член периметра (иначе 422 `ENTITY_NOT_IN_PERIMETER`). Блокировка версии (`consolidation_lock`) и
|
||||
`FOR UPDATE` не нужны: статус перепроверяется под блокировкой в `after_upload_to_version`."""
|
||||
found = await consolidations_repo.get_version(self._session, self._scope, version_id)
|
||||
if found is None:
|
||||
raise NotFoundError(VERSION_NOT_FOUND)
|
||||
version = found[0]
|
||||
if version.status not in IN_WORK_STATUSES:
|
||||
raise ConflictError("VERSION_NOT_EDITABLE", f"Версия в статусе {version.status.value} не редактируется")
|
||||
members = await consolidations_repo.members(self._session, version.id)
|
||||
if all(member.legal_entity_id != entity.id for member, _ in members):
|
||||
raise DomainValidationError("ENTITY_NOT_IN_PERIMETER", "ЮЛ не входит в периметр версии консолидации")
|
||||
return version
|
||||
|
||||
async def after_upload_to_version(self, version_id: uuid.UUID) -> None:
|
||||
"""После приёма хотя бы одного файла: версия `INITIATED` → `DOCUMENTS_PROCESSING` (`changed_by` — загрузивший).
|
||||
|
||||
В ОТДЕЛЬНОЙ транзакции (`session_transaction`, как аудит входа): `lock_version` (`FOR UPDATE`) → статус
|
||||
перечитывается — если уже не `INITIATED` (параллельный переход, аннулирование), ничего не меняется. Версия
|
||||
блокируется на короткое время и вне транзакции запроса, чтобы загрузка не держала её на время приёма файлов
|
||||
(D-9). Транзакция запроса к этому моменту версию не трогала, взаимных ожиданий нет."""
|
||||
# импорт внутри функции — как в consolidation_documents (design №36): модули консолидации импортируют друг друга
|
||||
from src.services.consolidation_status import transition # noqa: PLC0415
|
||||
|
||||
factory = self._transaction
|
||||
if factory is None:
|
||||
from src.db.db import session_transaction # noqa: PLC0415
|
||||
|
||||
factory = session_transaction
|
||||
async with factory() as session:
|
||||
version = await consolidations_repo.lock_version(session, version_id)
|
||||
if version is None or version.status is not ConsolidationStatus.INITIATED:
|
||||
return
|
||||
await transition(
|
||||
session,
|
||||
AuditService(session),
|
||||
version=version,
|
||||
to=ConsolidationStatus.DOCUMENTS_PROCESSING,
|
||||
changed_by=self._audit.user_id,
|
||||
comment=None,
|
||||
ip_address=self._audit.ip_address,
|
||||
user_agent=self._audit.user_agent,
|
||||
now=utc_now(),
|
||||
)
|
||||
|
||||
def require_storage(self) -> ObjectStorage:
|
||||
"""S3 приложения; без него загрузка невозможна — 503 `STORAGE_UNAVAILABLE`."""
|
||||
if self._storage is None:
|
||||
|
||||
@ -681,7 +681,13 @@ async def _map_candidate(
|
||||
rows = await session.execute(select(ChartOfAccount.code, ChartOfAccount.id))
|
||||
chart = {code: account_id for code, account_id in rows}
|
||||
index = await CounterpartyIndex.load(session)
|
||||
warnings = await map_file(session, job=job, file=file, result=result, chart=chart, index=index, now=clock())
|
||||
now = clock()
|
||||
warnings = await map_file(session, job=job, file=file, result=result, chart=chart, index=index, now=now)
|
||||
# Переход версий (D-6) — ПОСЛЕ `map_file`, а значит после автозамены (`supersede_previous` берёт advisory ЮЛ →
|
||||
# старые файлы → advisory ГК → версии): версии блокируются последними, порядок «задание → файл → документы →
|
||||
# версии» (D-9) не нарушается. До автозамены хук держал бы версии и затем ждал блокировки старых файлов.
|
||||
documents = list((await session.execute(select(Document).where(Document.stored_file_id == file.id))).scalars())
|
||||
await consolidation_documents.on_documents_mapped(session, file=file, documents=documents, now=now)
|
||||
logger.info(
|
||||
"Результат распознавания смапплен: файл PROCESSED. job_id={job_id}, file_id={file_id}, warnings={warnings}",
|
||||
job_id=candidate.job_id,
|
||||
|
||||
@ -24,12 +24,15 @@ from src.api.dependencies.directories import get_app_settings
|
||||
from src.core.config import get_settings
|
||||
from src.models.access import AppUser
|
||||
from src.models.audit_ex import AuditEvent
|
||||
from src.models.consolidation import ConsolidationStatusHistory, ConsolidationVersion
|
||||
from src.models.directories import LegalEntity, RegionalOffice
|
||||
from src.models.enums import (
|
||||
AuditAction,
|
||||
AuditSection,
|
||||
ConsolidationStatus,
|
||||
FileStatus,
|
||||
FileType,
|
||||
PeriodQuarter,
|
||||
RecognitionPipeline,
|
||||
RecognitionStatus,
|
||||
S3BucketRole,
|
||||
@ -37,7 +40,7 @@ from src.models.enums import (
|
||||
from src.models.files import RecognitionJob, S3Object, StoredFile
|
||||
from src.services.recognition_exchange import publish_pending
|
||||
from tests.conftest import AuthClientFactory, FakeStorage
|
||||
from tests.factories import make_company_group, make_legal_entity
|
||||
from tests.factories import make_company_group, make_consolidation, make_legal_entity, make_lock
|
||||
from tests.test_recognition_exchange import REQUEST_VALIDATOR, FakeNats
|
||||
|
||||
HTTP_CREATED = 201
|
||||
@ -531,3 +534,205 @@ async def test_concurrent_duplicates_accept_only_one(
|
||||
files = (await session.execute(select(func.count()).select_from(StoredFile))).scalar_one()
|
||||
assert files == 1
|
||||
assert len(committed_storage.objects) == 1
|
||||
|
||||
|
||||
# --- загрузка в контексте версии консолидации (spec primary-files «Загрузка в контексте версии») ---------------
|
||||
|
||||
HTTP_CONFLICT = 409
|
||||
KRASNODAR = "Краснодарский региональный филиал"
|
||||
|
||||
|
||||
async def _group_with_members(
|
||||
session: AsyncSession, *, office_name: str = ROSTOV, members: int = 2
|
||||
) -> tuple[Any, list[LegalEntity]]:
|
||||
office = (await session.execute(select(RegionalOffice).where(RegionalOffice.name == office_name))).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,
|
||||
*,
|
||||
status: ConsolidationStatus = ConsolidationStatus.INITIATED,
|
||||
office_name: str = ROSTOV,
|
||||
) -> tuple[ConsolidationVersion, list[LegalEntity], Any]:
|
||||
"""Версия 2024 Q3 ГК филиала `office_name` с двумя членами; третье ЮЛ той же ГК — вне периметра."""
|
||||
group, entities = await _group_with_members(session, office_name=office_name, members=3)
|
||||
initiator = (await session.execute(select(AppUser).where(AppUser.email == KA_RF))).scalar_one()
|
||||
_, version = await make_consolidation(
|
||||
session,
|
||||
group=group,
|
||||
year=2024,
|
||||
quarter=PeriodQuarter.Q3,
|
||||
members=entities[:2],
|
||||
initiator=initiator,
|
||||
status=status,
|
||||
)
|
||||
return version, entities, initiator
|
||||
|
||||
|
||||
async def _version_state(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 _post_in_version(
|
||||
client: httpx.AsyncClient, entity_id: uuid.UUID, version_id: uuid.UUID, *uploads: Upload
|
||||
) -> httpx.Response:
|
||||
files = [("files", (name, content, "application/octet-stream")) for name, content in uploads]
|
||||
return await client.post(_files_url(entity_id), params={"consolidation_version_id": str(version_id)}, files=files)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_upload_in_version_moves_initiated_to_documents_processing(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
|
||||
) -> None:
|
||||
version, (member, *_), _ = await _version(seeded)
|
||||
user_id = await _user_id(seeded, KA_RF)
|
||||
client = await auth_client(KA_RF)
|
||||
|
||||
response = await _post_in_version(client, member.id, version.id, ("osv.xls", _xls("in-version")))
|
||||
|
||||
assert response.status_code == HTTP_CREATED, response.text
|
||||
assert len(response.json()["accepted"]) == 1
|
||||
assert len(storage.objects) == 1
|
||||
assert await _version_state(seeded, version.id) is ConsolidationStatus.DOCUMENTS_PROCESSING
|
||||
history = (
|
||||
(
|
||||
await seeded.execute(
|
||||
select(ConsolidationStatusHistory)
|
||||
.where(ConsolidationStatusHistory.version_id == version.id)
|
||||
.order_by(ConsolidationStatusHistory.changed_at.desc())
|
||||
)
|
||||
)
|
||||
.scalars()
|
||||
.first()
|
||||
)
|
||||
assert history is not None
|
||||
assert (history.from_status, history.to_status) == (
|
||||
ConsolidationStatus.INITIATED,
|
||||
ConsolidationStatus.DOCUMENTS_PROCESSING,
|
||||
)
|
||||
assert history.changed_by == user_id
|
||||
[event] = (
|
||||
(await seeded.execute(select(AuditEvent).where(AuditEvent.action == AuditAction.CONSOLIDATION_STATUS_CHANGE)))
|
||||
.scalars()
|
||||
.all()
|
||||
)
|
||||
assert event.user_id == user_id
|
||||
assert event.consolidation_version_id == version.id
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.parametrize("status", [ConsolidationStatus.DOCUMENTS_PROCESSING, ConsolidationStatus.DATA_CORRECTION])
|
||||
async def test_upload_in_version_keeps_other_in_work_status(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage, status: ConsolidationStatus
|
||||
) -> None:
|
||||
version, (member, *_), _ = await _version(seeded, status=status)
|
||||
client = await auth_client(KA_RF)
|
||||
|
||||
response = await _post_in_version(client, member.id, version.id, ("osv.xls", _xls(f"keep-{status.value}")))
|
||||
|
||||
assert response.status_code == HTTP_CREATED, response.text
|
||||
assert await _version_state(seeded, version.id) is status
|
||||
assert len(storage.objects) == 1
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_upload_with_all_files_rejected_does_not_move_version(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
|
||||
) -> None:
|
||||
version, (member, *_), _ = await _version(seeded)
|
||||
client = await auth_client(KA_RF)
|
||||
|
||||
response = await _post_in_version(client, member.id, version.id, ("notes.docx", ZIP + b"x"))
|
||||
|
||||
assert response.status_code == HTTP_UNPROCESSABLE, response.text
|
||||
assert response.json()["accepted"] == []
|
||||
assert await _version_state(seeded, version.id) is ConsolidationStatus.INITIATED
|
||||
assert storage.objects == {}
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_upload_for_entity_outside_perimeter_is_422_and_writes_nothing(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
|
||||
) -> None:
|
||||
version, entities, _ = await _version(seeded)
|
||||
outsider = entities[2]
|
||||
client = await auth_client(KA_RF)
|
||||
before = await _counts(seeded)
|
||||
|
||||
response = await _post_in_version(client, outsider.id, version.id, ("osv.xls", _xls("outsider")))
|
||||
|
||||
assert response.status_code == HTTP_UNPROCESSABLE, response.text
|
||||
assert response.json()["detail"]["code"] == "ENTITY_NOT_IN_PERIMETER"
|
||||
assert await _counts(seeded) == before
|
||||
assert storage.objects == {}
|
||||
assert await _version_state(seeded, version.id) is ConsolidationStatus.INITIATED
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_upload_in_cancelled_version_is_409(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
|
||||
) -> None:
|
||||
version, (member, *_), _ = await _version(seeded, status=ConsolidationStatus.CANCELLED)
|
||||
client = await auth_client(KA_RF)
|
||||
before = await _counts(seeded)
|
||||
|
||||
response = await _post_in_version(client, member.id, version.id, ("osv.xls", _xls("cancelled")))
|
||||
|
||||
assert response.status_code == HTTP_CONFLICT, response.text
|
||||
assert response.json()["detail"]["code"] == "VERSION_NOT_EDITABLE"
|
||||
assert await _counts(seeded) == before
|
||||
assert storage.objects == {}
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_upload_in_foreign_version_is_404_like_missing(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
|
||||
) -> None:
|
||||
foreign, _, _ = await _version(seeded, office_name=KRASNODAR)
|
||||
own_entity = await _rostov_entity(seeded, "ООО Своё")
|
||||
client = await auth_client(KA_RF)
|
||||
before = await _counts(seeded)
|
||||
|
||||
foreign_response = await _post_in_version(client, own_entity.id, foreign.id, ("osv.xls", _xls("foreign-version")))
|
||||
missing_response = await _post_in_version(client, own_entity.id, uuid.uuid4(), ("osv.xls", _xls("missing-version")))
|
||||
|
||||
assert foreign_response.status_code == missing_response.status_code == HTTP_NOT_FOUND, foreign_response.text
|
||||
assert foreign_response.json() == missing_response.json()
|
||||
assert foreign_response.json()["detail"]["code"] == "NOT_FOUND"
|
||||
assert await _counts(seeded) == before
|
||||
assert storage.objects == {}
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_upload_does_not_require_version_lock(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
|
||||
) -> None:
|
||||
"""Блокировка версии для загрузки не нужна (spec): версию держит другой пользователь, загрузка проходит."""
|
||||
version, (member, *_), _ = await _version(seeded)
|
||||
holder = (await seeded.execute(select(AppUser).where(AppUser.email == CLIENT_MANAGER))).scalar_one()
|
||||
await make_lock(seeded, version=version, user=holder)
|
||||
client = await auth_client(KA_RF)
|
||||
|
||||
response = await _post_in_version(client, member.id, version.id, ("osv.xls", _xls("under-lock")))
|
||||
|
||||
assert response.status_code == HTTP_CREATED, response.text
|
||||
assert await _version_state(seeded, version.id) is ConsolidationStatus.DOCUMENTS_PROCESSING
|
||||
assert len(storage.objects) == 1
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_upload_without_version_does_not_touch_versions(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
|
||||
) -> None:
|
||||
version, (member, *_), _ = await _version(seeded)
|
||||
client = await auth_client(KA_RF)
|
||||
|
||||
response = await _post(client, member.id, ("osv.xls", _xls("no-version")))
|
||||
|
||||
assert response.status_code == HTTP_CREATED, response.text
|
||||
assert await _version_state(seeded, version.id) is ConsolidationStatus.INITIATED
|
||||
assert len(storage.objects) == 1
|
||||
|
||||
@ -14,6 +14,7 @@ from collections.abc import Sequence
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from loguru import logger
|
||||
from sqlalchemy import select, text
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||||
|
||||
@ -810,3 +811,166 @@ async def test_supersede_waits_for_create_and_unbinds_replaced_document(
|
||||
event = await _replace_event(maker)
|
||||
assert event.after_value is not None
|
||||
assert event.after_value["unbound_version_ids"] == [str(version_id)]
|
||||
|
||||
|
||||
# --- on_documents_mapped: NEEDS_CHECK при маппинге (osv-result-mapping, D-6) -------------------------------
|
||||
|
||||
|
||||
async def _mapped_file_documents(
|
||||
session: AsyncSession, entity: LegalEntity, *, uploaded_by: uuid.UUID | None, status: DocumentStatus, quarter=Q2
|
||||
) -> tuple[StoredFile, list[Document]]:
|
||||
"""Файл `PROCESSED` ЮЛ с одним квартальным документом ОСВ 60 за 2024 `quarter` в статусе `status`."""
|
||||
file, job = await make_stored_file(
|
||||
session, entity=entity, status=FileStatus.PROCESSED, job_status=RecognitionStatus.done, uploaded_by=uploaded_by
|
||||
)
|
||||
document = await make_document(
|
||||
session, file=file, job=job, account=ACCOUNT, year=YEAR, quarter=quarter, status=status, is_cumulative=False
|
||||
)
|
||||
return file, [document]
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_on_documents_mapped_moves_documents_processing_to_data_correction(seeded: AsyncSession) -> None:
|
||||
group, entities = await _group(seeded)
|
||||
version = await _version(seeded, group, entities, status=ConsolidationStatus.DOCUMENTS_PROCESSING)
|
||||
user = await _user(seeded)
|
||||
file, documents = await _mapped_file_documents(
|
||||
seeded, entities[0], uploaded_by=user.id, status=DocumentStatus.NEEDS_CHECK
|
||||
)
|
||||
|
||||
moved = await consolidation_documents.on_documents_mapped(seeded, file=file, documents=documents, now=utc_now())
|
||||
|
||||
assert moved == [version.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.DOCUMENTS_PROCESSING,
|
||||
ConsolidationStatus.DATA_CORRECTION,
|
||||
user.id,
|
||||
)
|
||||
[event] = (
|
||||
await seeded.execute(select(AuditEvent).where(AuditEvent.action == AuditAction.CONSOLIDATION_STATUS_CHANGE))
|
||||
).scalars()
|
||||
assert event.user_id == user.id
|
||||
assert await _bound(seeded, version.id) == set() # привязку маппинг не создаёт
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.parametrize("status", [ConsolidationStatus.INITIATED, ConsolidationStatus.DATA_CORRECTION])
|
||||
async def test_on_documents_mapped_ignores_other_statuses(seeded: AsyncSession, status: ConsolidationStatus) -> None:
|
||||
group, entities = await _group(seeded)
|
||||
version = await _version(seeded, group, entities, status=status)
|
||||
user = await _user(seeded)
|
||||
file, documents = await _mapped_file_documents(
|
||||
seeded, entities[0], uploaded_by=user.id, status=DocumentStatus.NEEDS_CHECK
|
||||
)
|
||||
history = len(await _history(seeded, version.id))
|
||||
|
||||
moved = await consolidation_documents.on_documents_mapped(seeded, file=file, documents=documents, now=utc_now())
|
||||
|
||||
assert moved == []
|
||||
assert await _status(seeded, version.id) is status
|
||||
assert len(await _history(seeded, version.id)) == history
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_on_documents_mapped_ignores_recognized_document(seeded: AsyncSession) -> None:
|
||||
group, entities = await _group(seeded)
|
||||
version = await _version(seeded, group, entities, status=ConsolidationStatus.DOCUMENTS_PROCESSING)
|
||||
file, documents = await _mapped_file_documents(
|
||||
seeded, entities[0], uploaded_by=(await _user(seeded)).id, status=DocumentStatus.RECOGNIZED
|
||||
)
|
||||
|
||||
moved = await consolidation_documents.on_documents_mapped(seeded, file=file, documents=documents, now=utc_now())
|
||||
|
||||
assert moved == []
|
||||
assert await _status(seeded, version.id) is ConsolidationStatus.DOCUMENTS_PROCESSING
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_on_documents_mapped_skips_document_that_does_not_fit_version(seeded: AsyncSession) -> None:
|
||||
"""Квартальный документ Q4 не подходит версии Q3 (D-5): `NEEDS_CHECK`, но перехода нет."""
|
||||
group, entities = await _group(seeded)
|
||||
version = await _version(seeded, group, entities, status=ConsolidationStatus.DOCUMENTS_PROCESSING)
|
||||
file, documents = await _mapped_file_documents(
|
||||
seeded, entities[0], uploaded_by=(await _user(seeded)).id, status=DocumentStatus.NEEDS_CHECK, quarter=Q4
|
||||
)
|
||||
|
||||
moved = await consolidation_documents.on_documents_mapped(seeded, file=file, documents=documents, now=utc_now())
|
||||
|
||||
assert moved == []
|
||||
assert await _status(seeded, version.id) is ConsolidationStatus.DOCUMENTS_PROCESSING
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_on_documents_mapped_skips_entity_outside_perimeter(seeded: AsyncSession) -> None:
|
||||
group, entities = await _group(seeded, members=3)
|
||||
version = await _version(seeded, group, entities[:2], status=ConsolidationStatus.DOCUMENTS_PROCESSING)
|
||||
file, documents = await _mapped_file_documents(
|
||||
seeded, entities[2], uploaded_by=(await _user(seeded)).id, status=DocumentStatus.NEEDS_CHECK
|
||||
)
|
||||
|
||||
moved = await consolidation_documents.on_documents_mapped(seeded, file=file, documents=documents, now=utc_now())
|
||||
|
||||
assert moved == []
|
||||
assert await _status(seeded, version.id) is ConsolidationStatus.DOCUMENTS_PROCESSING
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_on_documents_mapped_without_uploader_logs_and_skips(seeded: AsyncSession) -> None:
|
||||
"""`uploaded_by` NULL: `changed_by NOT NULL` — переход пропускается, в лог пишется причина, ошибки нет."""
|
||||
group, entities = await _group(seeded)
|
||||
version = await _version(seeded, group, entities, status=ConsolidationStatus.DOCUMENTS_PROCESSING)
|
||||
file, documents = await _mapped_file_documents(
|
||||
seeded, entities[0], uploaded_by=None, status=DocumentStatus.NEEDS_CHECK
|
||||
)
|
||||
messages: list[str] = []
|
||||
sink = logger.add(lambda message: messages.append(str(message)), level="INFO")
|
||||
try:
|
||||
moved = await consolidation_documents.on_documents_mapped(seeded, file=file, documents=documents, now=utc_now())
|
||||
finally:
|
||||
logger.remove(sink)
|
||||
|
||||
assert moved == []
|
||||
assert await _status(seeded, version.id) is ConsolidationStatus.DOCUMENTS_PROCESSING
|
||||
assert any(str(file.id) in message and "uploaded_by" in message for message in messages)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_on_documents_mapped_waits_for_member_edit_and_sees_new_member(
|
||||
committed_sessionmaker: async_sessionmaker[AsyncSession],
|
||||
) -> None:
|
||||
"""R-2: правка периметра держит версию `DOCUMENTS_PROCESSING` и добавила ЮЛ файла (не зафиксировано); хук маппинга
|
||||
блокирует надмножество версий ГК, ждёт правку и после commit видит нового члена — переход не теряется."""
|
||||
maker = committed_sessionmaker
|
||||
data = await _committed_setup(maker)
|
||||
async with maker() as session, session.begin():
|
||||
version = await session.get(ConsolidationVersion, data["version_id"])
|
||||
assert version is not None
|
||||
version.status = ConsolidationStatus.DOCUMENTS_PROCESSING
|
||||
file = await session.get(StoredFile, data["file_id"])
|
||||
assert file is not None
|
||||
file.uploaded_by = data["user_id"]
|
||||
|
||||
async def mapped() -> list[uuid.UUID]:
|
||||
async with maker() as session, session.begin():
|
||||
file = await session.get(StoredFile, data["file_id"])
|
||||
document = await session.get(Document, data["document_id"])
|
||||
assert file is not None
|
||||
assert document is not None
|
||||
return await consolidation_documents.on_documents_mapped(
|
||||
session, file=file, documents=[document], now=utc_now()
|
||||
)
|
||||
|
||||
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()
|
||||
hook = asyncio.create_task(mapped())
|
||||
await _wait_for_lock_waiter(maker)
|
||||
assert not hook.done()
|
||||
moved = await asyncio.wait_for(hook, LOCK_WAIT_SECONDS)
|
||||
|
||||
assert moved == [data["version_id"]]
|
||||
async with maker() as session:
|
||||
assert await _status(session, data["version_id"]) is ConsolidationStatus.DATA_CORRECTION
|
||||
|
||||
@ -29,7 +29,7 @@ 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.consolidation import ConsolidationDocument, ConsolidationStatusHistory, ConsolidationVersion
|
||||
from src.models.directories import LegalEntity
|
||||
from src.models.enums import (
|
||||
AuditAction,
|
||||
@ -78,6 +78,7 @@ GOLDEN_OSV_66 = Path("golden") / "expected_osv_parse" / "rb" / "2024-FY" / "ОС
|
||||
GOLDEN_OSV_60 = Path("golden") / "expected_osv_parse" / "rb" / "2024-FY" / "ОСВ_60 за 2024.json"
|
||||
BUCKET = "arm-ka"
|
||||
OWNER_NAME = "АО «Русский бисквит»"
|
||||
FACTORY_HISTORY = 2 # `make_consolidation(status=DOCUMENTS_PROCESSING)`: NULL → INITIATED → DOCUMENTS_PROCESSING
|
||||
PERF_BUDGET_SECONDS = 2.0 # цель плана: маппинг ~1000 строк ОСВ_60 за 2024 быстрее 2 с
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
@ -1212,6 +1213,135 @@ async def test_replaced_document_is_unbound_from_versions_in_work(
|
||||
assert event.after_value == {"file_id": str(new_file.id), "version_no": 2, "unbound_version_ids": [str(in_work_id)]}
|
||||
|
||||
|
||||
# --- osv-result-mapping «Документ NEEDS_CHECK переводит версию в корректировку» (D-6) ------------------------------
|
||||
|
||||
|
||||
async def _version_in(
|
||||
maker: async_sessionmaker[AsyncSession],
|
||||
*,
|
||||
status: ConsolidationStatus = ConsolidationStatus.DOCUMENTS_PROCESSING,
|
||||
quarter: PeriodQuarter = PeriodQuarter.Q3,
|
||||
) -> tuple[uuid.UUID, uuid.UUID, uuid.UUID]:
|
||||
"""ГК с ЮЛ-владельцем (ИНН как в результате) и вторым ЮЛ; версия `quarter` 2025; `(entity, version, user)`."""
|
||||
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="ГК Маппинг версии")
|
||||
owner = await make_legal_entity(session, name=OWNER_NAME, inn="7700000000", group=group)
|
||||
other = await make_legal_entity(session, name="ООО Второе", group=group)
|
||||
_, version = await make_consolidation(
|
||||
session, group=group, year=2025, quarter=quarter, members=[owner, other], initiator=initiator, status=status
|
||||
)
|
||||
return owner.id, version.id, initiator.id
|
||||
|
||||
|
||||
async def _load_entity(maker: async_sessionmaker[AsyncSession], entity_id: uuid.UUID) -> LegalEntity:
|
||||
async with maker() as session:
|
||||
entity = await session.get(LegalEntity, entity_id)
|
||||
assert entity is not None
|
||||
return entity
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("job_status", "moved"),
|
||||
[(RecognitionStatus.needs_check, True), (RecognitionStatus.done, False)], # NEEDS_CHECK / RECOGNIZED (чистый)
|
||||
)
|
||||
async def test_mapping_moves_documents_processing_version_only_for_needs_check(
|
||||
committed_sessionmaker: async_sessionmaker[AsyncSession], job_status: RecognitionStatus, moved: bool
|
||||
):
|
||||
"""«Предупреждение маппинга двигает статус» и «Документ RECOGNIZED»: ОСВ 60 за 2025 Q1 (квартальная) подходит версии
|
||||
2025 Q3; `NEEDS_CHECK` → `DATA_CORRECTION`, запись истории с `changed_by` = загрузивший; привязок нет."""
|
||||
maker, storage = committed_sessionmaker, FakeStorage()
|
||||
entity_id, version_id, user_id = await _version_in(maker)
|
||||
file_id, _, _ = await _prepare(
|
||||
maker,
|
||||
storage,
|
||||
_minimal(),
|
||||
entity=await _load_entity(maker, entity_id),
|
||||
job_status=job_status,
|
||||
uploaded_by=user_id,
|
||||
)
|
||||
|
||||
assert await process_once(maker, storage, _settings()) is True
|
||||
|
||||
[document] = await _documents(maker, file_id)
|
||||
assert document.status is (DocumentStatus.NEEDS_CHECK if moved else DocumentStatus.RECOGNIZED)
|
||||
async with maker() as session:
|
||||
version = await session.get(ConsolidationVersion, version_id)
|
||||
assert version is not None
|
||||
history = list(
|
||||
(
|
||||
await session.execute(
|
||||
select(ConsolidationStatusHistory)
|
||||
.where(ConsolidationStatusHistory.version_id == version_id)
|
||||
.order_by(ConsolidationStatusHistory.changed_at)
|
||||
)
|
||||
).scalars()
|
||||
)
|
||||
bindings = (await session.execute(select(func.count()).select_from(ConsolidationDocument))).scalar_one()
|
||||
assert bindings == 0
|
||||
if moved:
|
||||
assert version.status is ConsolidationStatus.DATA_CORRECTION
|
||||
last = history[-1]
|
||||
assert (last.from_status, last.to_status, last.changed_by) == (
|
||||
ConsolidationStatus.DOCUMENTS_PROCESSING,
|
||||
ConsolidationStatus.DATA_CORRECTION,
|
||||
user_id,
|
||||
)
|
||||
else:
|
||||
assert version.status is ConsolidationStatus.DOCUMENTS_PROCESSING
|
||||
assert len(history) == FACTORY_HISTORY # создание и переход фабрики в DOCUMENTS_PROCESSING; новых записей нет
|
||||
|
||||
|
||||
async def test_mapping_does_not_move_initiated_version(committed_sessionmaker: async_sessionmaker[AsyncSession]):
|
||||
"""Только `DOCUMENTS_PROCESSING` реагирует на маппинг (spec): версия `INITIATED` остаётся."""
|
||||
maker, storage = committed_sessionmaker, FakeStorage()
|
||||
entity_id, version_id, user_id = await _version_in(maker, status=ConsolidationStatus.INITIATED)
|
||||
await _prepare(
|
||||
maker,
|
||||
storage,
|
||||
_minimal(),
|
||||
entity=await _load_entity(maker, entity_id),
|
||||
job_status=RecognitionStatus.needs_check,
|
||||
uploaded_by=user_id,
|
||||
)
|
||||
|
||||
assert await process_once(maker, storage, _settings()) is True
|
||||
|
||||
async with maker() as session:
|
||||
version = await session.get(ConsolidationVersion, version_id)
|
||||
assert version is not None
|
||||
assert version.status is ConsolidationStatus.INITIATED
|
||||
|
||||
|
||||
async def test_mapping_without_uploader_skips_transition_without_error(
|
||||
committed_sessionmaker: async_sessionmaker[AsyncSession],
|
||||
):
|
||||
"""`uploaded_by` NULL: маппинг завершается (файл `PROCESSED`), версия не двигается, причина — в логе."""
|
||||
maker, storage = committed_sessionmaker, FakeStorage()
|
||||
entity_id, version_id, _ = await _version_in(maker)
|
||||
file_id, _, _ = await _prepare(
|
||||
maker,
|
||||
storage,
|
||||
_minimal(),
|
||||
entity=await _load_entity(maker, entity_id),
|
||||
job_status=RecognitionStatus.needs_check,
|
||||
uploaded_by=None,
|
||||
)
|
||||
messages: list[str] = []
|
||||
sink = logger.add(lambda message: messages.append(str(message)), level="INFO")
|
||||
try:
|
||||
assert await process_once(maker, storage, _settings()) is True
|
||||
finally:
|
||||
logger.remove(sink)
|
||||
|
||||
assert (await _file(maker, file_id)).status is FileStatus.PROCESSED
|
||||
async with maker() as session:
|
||||
version = await session.get(ConsolidationVersion, version_id)
|
||||
assert version is not None
|
||||
assert version.status is ConsolidationStatus.DOCUMENTS_PROCESSING
|
||||
assert any(str(file_id) in message and "uploaded_by" in message for message in messages)
|
||||
|
||||
|
||||
# --- audit-trail: обмен и маппинг не пишут audit_event (ASSUMPTION(OQ-55)) ----------------------------------------
|
||||
|
||||
|
||||
|
||||
@ -154,9 +154,9 @@
|
||||
- Загрузка: до приёма — `get_version(scope)` (404 `VERSION_NOT_FOUND`) → `status ∈ IN_WORK_STATUSES` (409 `VERSION_NOT_EDITABLE`) → ЮЛ ∈ `members` (422 `ENTITY_NOT_IN_PERIMETER`); после цикла по файлам, если `accepted` и статус `INITIATED` — отдельная транзакция (`session_transaction` как у загрузки файла): `lock_version` → если всё ещё `INITIATED` → `transition(to=DOCUMENTS_PROCESSING, changed_by=user)`.
|
||||
- `on_documents_mapped(session, *, file, documents, now) -> list[uuid.UUID]`: если `file.uploaded_by is None` → `logger.info` и `[]`; версии `DOCUMENTS_PROCESSING` с ЮЛ файла в периметре (`FOR UPDATE` по `id`); для каждой — есть документ `NEEDS_CHECK` с `fits` → `transition(to=DATA_CORRECTION, changed_by=file.uploaded_by)`; привязки не создаёт.
|
||||
|
||||
- [ ] **Step 1: Тесты.** Сценарии `primary-files` «Загрузка в контексте версии» (переход, запись истории), «ЮЛ вне периметра» (422, файлов нет), версия `CANCELLED` → 409, чужая версия → 404; `osv-result-mapping` «Предупреждение маппинга двигает статус», «Документ RECOGNIZED» (без перехода), `uploaded_by` NULL — без перехода и без ошибки.
|
||||
- [ ] **Step 2–4:** падает → реализация → `make check`, `pytest -m integration tests/test_api_files_upload.py tests/test_result_mapping.py -q` → PASS.
|
||||
- [ ] **Step 5: Commit** `feat(consolidation): переходы DOCUMENTS_PROCESSING и DATA_CORRECTION по загрузке и маппингу`.
|
||||
- [x] **Step 1: Тесты.** Сценарии `primary-files` «Загрузка в контексте версии» (переход, запись истории), «ЮЛ вне периметра» (422, файлов нет), версия `CANCELLED` → 409, чужая версия → 404; `osv-result-mapping` «Предупреждение маппинга двигает статус», «Документ RECOGNIZED» (без перехода), `uploaded_by` NULL — без перехода и без ошибки.
|
||||
- [x] **Step 2–4:** падает → реализация → `make check`, `pytest -m integration tests/test_api_files_upload.py tests/test_result_mapping.py -q` → PASS.
|
||||
- [x] **Step 5: Commit** `feat(consolidation): переходы DOCUMENTS_PROCESSING и DATA_CORRECTION по загрузке и маппингу`.
|
||||
|
||||
---
|
||||
|
||||
|
||||
@ -221,3 +221,9 @@ legal_entity_name, account_code, missing_quarters[]}}`. Фильтры `GET …/
|
||||
| 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` той же транзакции |
|
||||
| 37 | 2026-10-06 | Отдельная транзакция перехода после загрузки (`after_upload_to_version`) берётся из `app.state.session_audit.transaction` (та же фабрика `session_transaction`, что у аудита входа; без неё — `src.db.db.session_transaction`); `get_primary_files_service` передаёт её в `PrimaryFilesService(…, transaction)` | отдельной зависимости «фабрика транзакций» в приложении нет, а фикстуры тестов уже подставляют фабрику в `SessionAudit`; не меняем `app.py` и `conftest` | если `SessionAudit` когда-либо уйдёт из `app.state`, сервис молча возьмёт боевой `session_transaction` (в тестах это была бы чужая БД) |
|
||||
| 38 | 2026-10-06 | Переход после загрузки пишет `CONSOLIDATION_STATUS_CHANGE` с IP и user-agent запроса (в отличие от хука подтверждения, №36); `transition` вызывается из сервиса загрузки с `RequestAudit`-контекстом, но в своей транзакции и своим `AuditService(session)` | сервис знает IP/UA запроса, а `transition` их принимает; потери информации нет | нет |
|
||||
| 39 | 2026-10-06 | Версия в статусе `DOCUMENTS_PROCESSING`/`DATA_CORRECTION` при загрузке не меняется, `after_upload_to_version` — no-op; если между проверкой и переходом версию аннулировали/перевели, переход не выполняется (статус перечитывается под `FOR UPDATE`), загрузка при этом остаётся принятой (201) | проверка до приёма и переход после — две разные транзакции, окно между ними неустранимо без блокировки версии, которую спека запрещает | версия, аннулированная в окне, остаётся аннулированной; файл принят и виден в ЮЛ, но к версии не привязан (привязка — только подтверждением) |
|
||||
| 40 | 2026-10-06 | `on_documents_mapped` блокирует через `lock_versions_for_entity` надмножество по `IN_WORK_STATUSES` и фильтрует `DOCUMENTS_PROCESSING` ПОСЛЕ блокировки, а не передаёт `statuses={DOCUMENTS_PROCESSING}`; вызывается в `_map_candidate` ПОСЛЕ `map_file`, то есть после автозамены | при фильтре статуса в блокирующем запросе версия, которую параллельно переводят `INITIATED → DOCUMENTS_PROCESSING`, не ждалась бы; порядок «файл → документы → advisory ГК → версии» сохраняется, автозамена внутри `map_file` берёт advisory ЮЛ → старые файлы → advisory ГК → версии раньше хука; вызов до автозамены держал бы версии и затем ждал старые файлы (встречная блокировка с подтверждением) | блокируется чуть больше версий ГК (все «в работе»), на время короткой транзакции маппинга |
|
||||
| 41 | 2026-10-06 | Хук получает документы файла отдельным `SELECT` по `stored_file_id` после `map_file` (возвращаемое значение `map_file` — предупреждения — не менялось) | сигнатура `map_file` и её тесты не трогаются; документы только что записаны в этой же транзакции | один лишний запрос на файл |
|
||||
| 42 | 2026-10-06 | Переход по маппингу — только если среди документов файла есть `NEEDS_CHECK`, подходящий версии по `fits` (`fits` — D-5: период, год, член); статус файла/документа в других статусах (`RECOGNIZED`) переход не вызывает, привязки не создаются | спека osv-result-mapping | версия с неподходящим по периоду `NEEDS_CHECK` документом остаётся в `DOCUMENTS_PROCESSING` до подтверждения подходящего |
|
||||
|
||||
@ -35,7 +35,7 @@
|
||||
`DocumentService.confirm` (версии `FOR UPDATE` после документа), `supersede_previous` (`unbound_version_ids` в аудите).
|
||||
Тесты: сценарии `documents-api` (привязка, период, `CANCELLED`, подхват, две версии документа), `primary-files`
|
||||
«Заменённый документ был в версии», порядок блокировок (подтверждение ∥ правка периметра).
|
||||
- [ ] 3.2 **Статусы по событиям файлов.** `upload.py`: `consolidation_version_id` (проверки, переход после приёма);
|
||||
- [x] 3.2 **Статусы по событиям файлов.** `upload.py`: `consolidation_version_id` (проверки, переход после приёма);
|
||||
`result_mapping`: хук `on_documents_mapped` (NEEDS_CHECK → `DATA_CORRECTION`, `uploaded_by`, пропуск при NULL).
|
||||
Тесты: сценарии `primary-files` «Загрузка в контексте версии», `osv-result-mapping` «Предупреждение маппинга».
|
||||
- [ ] 3.3 **Чек-лист.** `domain/checklist.py:build(rules, members, documents, files)` (чистая), `GET …/checklist`, ячейки в карточке.
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user