"""Цикл маппинга результата распознавания в `document`/`osv_entry` (spec osv-result-mapping; design D-1, D-2, D-3, D-7, D-9; рулинги R-3, R-4, R-5; Review Focus 1, 2). Интеграционные тесты — на отдельной временной БД (`committed_sessionmaker`: цикл работает в собственных транзакциях), результат лежит в `FakeStorage` под `results/{file_id}/{job_id}/result.json`. Синтетические результаты строятся из примера контракта `contracts/examples/result-osv-minimal.json`; сценарий «Результат смапплен» и замер скорости — на golden-снимках парсера (маркер `fixtures`). """ import asyncio import copy import json import time import uuid from collections.abc import Callable from datetime import UTC, datetime, timedelta from decimal import Decimal from pathlib import Path from typing import Any import pytest from loguru import logger from sqlalchemy import func, select, update from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from src.contract import ack_msg_id, result_msg_id from src.contract.result import RecognitionResult from src.core.config import Settings, get_settings from src.domain.consolidation_periods import reporting_date from src.models.access import AppUser from src.models.accounting import ChartOfAccount from src.models.audit_ex import AuditEvent from src.models.consolidation import ( Consolidation, ConsolidationDocument, ConsolidationStatusHistory, ConsolidationVersion, ) from src.models.directories import LegalEntity from src.models.enums import ( AuditAction, AuditSection, ConsolidationStatus, CounterpartyMatchKind, DocumentCheckResult, DocumentStatus, DocumentType, FileStatus, OsvDataType, OsvGroupType, PeriodQuarter, RecognitionStatus, ) from src.models.files import Document, RecognitionJob, StoredFile from src.models.osv import OsvEntry from src.repositories.counterparties import CounterpartyIndex from src.services import consolidation_documents from src.services import result_mapping as rm from src.services.recognition_exchange import expire_deadlines, handle_message from src.services.result_mapping import ( MAPPING_ATTEMPTS_KEY, MAPPING_ERROR_KEY, MAPPING_WARNINGS_KEY, NO_DOCUMENTS, Candidate, MappingFatal, process_once, run_mapping_loop, supersede_previous, validate_result_key, ) from tests.conftest import FakeStorage 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] MINIMAL_RESULT = API_DIR / "contracts" / "examples" / "result-osv-minimal.json" GOLDEN_OSV_66 = Path("golden") / "expected_osv_parse" / "rb" / "2024-FY" / "ОСВ_66 за 2024.json" 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 # --- данные --------------------------------------------------------------------------------------------------------- def _minimal() -> dict[str, Any]: return json.loads(MINIMAL_RESULT.read_text("utf-8")) def _golden(fixtures_dir: Path, path: Path) -> dict[str, Any]: data = json.loads((fixtures_dir / path).read_text("utf-8")) # у снимка нет job_id/file_id (подставляются) и версии сервиса (в контракте она обязательна) data["service"] = {"version": "golden", **data["service"]} return data def _osv(result: dict[str, Any]) -> dict[str, Any]: return result["documents"][0] def _root(rows: list[dict[str, Any]]) -> dict[str, Any]: return next(row for row in rows if row["level"] == 0 and row["parent_row_number"] is None) def _settings() -> Settings: return get_settings() def _now() -> datetime: return datetime.now(UTC) async def _prepare( # noqa: PLR0913 - каждое поле подготовки задаётся тестом явно maker: async_sessionmaker[AsyncSession], storage: FakeStorage, result: dict[str, Any], *, entity: LegalEntity | None = None, owner_name: str = OWNER_NAME, owner_inn: str | None = None, job_status: RecognitionStatus = RecognitionStatus.done, file_status: FileStatus = FileStatus.PROCESSING, key: str | None = None, uploaded_by: uuid.UUID | None = None, ) -> tuple[uuid.UUID, uuid.UUID, uuid.UUID]: """ЮЛ-владелец, файл `PROCESSING` с текущим заданием `done`, результат в `storage`; `(file_id, job_id, entity_id)`. `job_id`/`file_id` результата подставляются; `key` — подменить ключ `s3_object` результата (OQ-70).""" async with maker() as session, session.begin(): owner = entity or await make_legal_entity(session, name=owner_name, inn=owner_inn) file, job = await make_stored_file( session, entity=owner, status=file_status, job_status=job_status, uploaded_by=uploaded_by ) job.finished_at = _now() job.metrics = copy.deepcopy(result.get("metrics")) # как inbox: метрики сервиса из конверта payload = copy.deepcopy(result) | {"job_id": str(job.id), "file_id": str(file.id)} s3 = await make_result_object(session, job=job, storage=storage, result=payload) if key is not None: storage.objects[key] = storage.objects[s3.key] s3.key = key return file.id, job.id, owner.id async def _file(maker: async_sessionmaker[AsyncSession], file_id: uuid.UUID) -> StoredFile: async with maker() as session: file = await session.get(StoredFile, file_id) assert file is not None return file async def _job(maker: async_sessionmaker[AsyncSession], job_id: uuid.UUID) -> RecognitionJob: async with maker() as session: job = await session.get(RecognitionJob, job_id) assert job is not None return job async def _documents(maker: async_sessionmaker[AsyncSession], file_id: uuid.UUID) -> list[Document]: async with maker() as session: rows = await session.execute(select(Document).where(Document.stored_file_id == file_id)) return list(rows.scalars()) async def _entries(maker: async_sessionmaker[AsyncSession], document_id: uuid.UUID) -> list[OsvEntry]: async with maker() as session: rows = await session.execute( select(OsvEntry).where(OsvEntry.document_id == document_id).order_by(OsvEntry.row_number) ) return list(rows.scalars()) # --- проверка ключа результата (без БД) ---------------------------------------------------------------------------- def _candidate(key: str, bucket: str = BUCKET) -> Candidate: return Candidate(job_id=JOB_ID, file_id=FILE_ID, bucket=bucket, key=key) JOB_ID = uuid.UUID("3f2a9c1e-7b4d-4e0a-9d51-2c8e6a1b0f01") FILE_ID = uuid.UUID("a1c4e7d2-5b36-4f89-8e12-9d0c3b7a4e02") @pytest.mark.parametrize( "key", [f"results/{FILE_ID}/{JOB_ID}/result.json", f"results/{FILE_ID}/{JOB_ID}/sub/result.json"], ) def test_validate_result_key_accepts_job_folder(key: str) -> None: validate_result_key(_candidate(key), BUCKET) @pytest.mark.parametrize( ("key", "bucket"), [ (f"results/{uuid.uuid4()}/{JOB_ID}/result.json", BUCKET), # чужой файл (f"results/{FILE_ID}/{uuid.uuid4()}/result.json", BUCKET), # чужое задание (f"uploads/{FILE_ID}/{JOB_ID}/result.json", BUCKET), # не папка результатов (f"results/{FILE_ID}/{JOB_ID}/", BUCKET), # нет имени объекта (f"results/{FILE_ID}/{JOB_ID}/../{uuid.uuid4()}/result.json", BUCKET), # выход из папки (f"results/{FILE_ID}/{JOB_ID}/result.json", "other-bucket"), # чужой бакет ], ) def test_validate_result_key_rejects_outside_job_folder(key: str, bucket: str) -> None: with pytest.raises(MappingFatal): validate_result_key(_candidate(key, bucket), BUCKET) # --- сценарии спеки ------------------------------------------------------------------------------------------------ @pytest.mark.fixtures async def test_golden_osv_66_mapped( # noqa: PLR0915 - один сценарий спеки целиком committed_sessionmaker: async_sessionmaker[AsyncSession], fixtures_dir: Path ): """«Результат смапплен», «Дерево и счета»: документ 66/2024/Q4/нарастающая, строки без TOTAL, `group_schema`.""" maker, storage = committed_sessionmaker, FakeStorage() golden = _golden(fixtures_dir, GOLDEN_OSV_66) file_id, job_id, entity_id = await _prepare(maker, storage, golden) assert await process_once(maker, storage, _settings()) is True file = await _file(maker, file_id) assert file.status is FileStatus.PROCESSED assert file.updated_at is not None [document] = await _documents(maker, file_id) assert document.document_type is DocumentType.OSV assert document.osv_account_code == "66" assert document.period_year == 2024 # noqa: PLR2004 - год из шапки снимка assert document.period_quarter is PeriodQuarter.Q4 assert document.is_cumulative is True assert document.locator == "sheet:Лист_1" assert document.legal_entity_id == entity_id assert document.status is DocumentStatus.CONFIRMED # done, чистые проверки, шапка = владелец → автоподтверждение assert document.counterparty_check is DocumentCheckResult.MATCH assert document.period_check is DocumentCheckResult.UNKNOWN rows = _osv(golden)["rows"] entries = await _entries(maker, document.id) assert len(entries) == len(rows) - 1 # всё, кроме TOTAL total_rows = {row["row_number"] for row in rows if row["group_type"] == "TOTAL"} assert total_rows and not total_rows & {entry.row_number for entry in entries} # в enum DBML TOTAL нет by_row = {entry.row_number: entry for entry in entries} for row in rows: if row["group_type"] == "TOTAL": continue entry = by_row[row["row_number"]] assert entry.recognition_job_id == job_id assert entry.owner_legal_entity_id == entity_id assert entry.is_modified is False assert entry.level == row["level"] assert entry.row_value == row["row_value"] for name in ("opening_debit", "opening_credit", "turnover_debit", "turnover_credit"): expected = row.get(name) assert getattr(entry, name) == (Decimal(expected) if expected is not None else None) if row["parent_row_number"] is None: assert entry.parent_entry_id is None else: assert entry.parent_entry_id == by_row[row["parent_row_number"]].id async with maker() as session: chart_6603 = (await session.execute(select(ChartOfAccount.id).where(ChartOfAccount.code == "66.03"))).scalar() [yug] = [e for e in entries if e.account_code == "66.03" and e.group_type is OsvGroupType.COUNTERPARTY] assert yug.counterparty_name == "Юг-Финанс / ПЕКАРНЯ СЧАСТЛИВОЧНАЯ" parent = next(e for e in entries if e.id == yug.parent_entry_id) assert parent.account_code == "66.03" assert parent.group_type is OsvGroupType.ACCOUNT assert yug.account_root == "66" assert yug.account_id == chart_6603 is not None job = await _job(maker, job_id) assert job.group_schema == [ { "locator": "sheet:Лист_1", "groups": [ {"type": "ACCOUNT", "name": "Счет", "has_inn": None}, {"type": "COUNTERPARTY", "name": "Контрагенты", "has_inn": None}, ], } ] assert job.metrics[MAPPING_WARNINGS_KEY] == [] assert job.metrics["rows"] == 9 # noqa: PLR2004 - метрики сервиса сохранены рядом с ключами маппинга @pytest.mark.fixtures async def test_mapping_performance_osv_60(committed_sessionmaker: async_sessionmaker[AsyncSession], fixtures_dir: Path): """Цель плана: маппинг самой длинной ОСВ (`ОСВ_60 за 2024`, ~1000 строк) — быстрее 2 с на один проход цикла.""" maker, storage = committed_sessionmaker, FakeStorage() golden = _golden(fixtures_dir, GOLDEN_OSV_60) file_id, _, _ = await _prepare(maker, storage, golden) started = time.perf_counter() assert await process_once(maker, storage, _settings()) is True elapsed = time.perf_counter() - started [document] = await _documents(maker, file_id) rows = _osv(golden)["rows"] entries = await _entries(maker, document.id) total_rows = sum(1 for row in rows if row["group_type"] == "TOTAL") assert len(entries) == len(rows) - total_rows logger.info("Маппинг ОСВ_60 за 2024: {rows} строк за {elapsed:.3f} с", rows=len(entries), elapsed=elapsed) print(f"\nPERF: ОСВ_60 за 2024 — {len(entries)} строк за {elapsed:.3f} с") # noqa: T201 - замер в отчёт assert elapsed < PERF_BUDGET_SECONDS async def test_result_key_outside_job_folder_is_error(committed_sessionmaker: async_sessionmaker[AsyncSession]): """«Ключ результата вне папки задания»: файл `ERROR` с ошибкой маппинга, документов нет, повтора нет.""" maker, storage = committed_sessionmaker, FakeStorage() file_id, job_id, _ = await _prepare( maker, storage, _minimal(), key=f"results/{uuid.uuid4()}/{uuid.uuid4()}/result.json" ) assert await process_once(maker, storage, _settings()) is True file = await _file(maker, file_id) assert file.status is FileStatus.ERROR job = await _job(maker, job_id) assert "results/" in job.metrics[MAPPING_ERROR_KEY] assert MAPPING_ATTEMPTS_KEY not in job.metrics # неповторяемая ошибка — без счётчика попыток assert await _documents(maker, file_id) == [] assert await process_once(maker, storage, _settings()) is False # повтора нет async def test_missing_result_object_is_error(committed_sessionmaker: async_sessionmaker[AsyncSession]): """Объекта результата нет в S3 — неповторяемая ошибка (D-2): файл `ERROR`.""" maker, storage = committed_sessionmaker, FakeStorage() file_id, job_id, _ = await _prepare(maker, storage, _minimal()) storage.objects.clear() assert await process_once(maker, storage, _settings()) is True assert (await _file(maker, file_id)).status is FileStatus.ERROR assert "не найден" in (await _job(maker, job_id)).metrics[MAPPING_ERROR_KEY] async def test_result_with_empty_documents_is_error(committed_sessionmaker: async_sessionmaker[AsyncSession]): """Результат без документов не проходит схему контракта (documents ≥ 1) → файл `ERROR`, не исключение цикла.""" maker, storage = committed_sessionmaker, FakeStorage() result = _minimal() | {"status": "needs_check", "documents": []} file_id, job_id, _ = await _prepare(maker, storage, result, job_status=RecognitionStatus.needs_check) assert await process_once(maker, storage, _settings()) is True assert (await _file(maker, file_id)).status is FileStatus.ERROR assert "схем" in (await _job(maker, job_id)).metrics[MAPPING_ERROR_KEY] assert await _documents(maker, file_id) == [] async def test_map_file_without_documents_is_fatal(seeded: AsyncSession): """`map_file` на результате без документов (схема такого не пропускает — защита) → `NO_DOCUMENTS`.""" entity = await make_legal_entity(seeded) file, job = await make_stored_file( seeded, entity=entity, status=FileStatus.PROCESSING, job_status=RecognitionStatus.done ) result = RecognitionResult.model_construct(documents=[]) index = await CounterpartyIndex.load(seeded) with pytest.raises(MappingFatal, match=NO_DOCUMENTS): await rm.map_file(seeded, job=job, file=file, result=result, chart={}, index=index, now=datetime.now(UTC)) async def test_mapping_failure_retries_then_error( committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch ): """«Сбой маппинга с повтором»: откат (документов нет), `mapping_attempts` 1, 2 — файл `PROCESSING`; третья неудача — `ERROR` с `metrics.mapping_error`.""" maker, storage = committed_sessionmaker, FakeStorage() file_id, job_id, _ = await _prepare(maker, storage, _minimal()) real_map_file = rm.map_file calls: list[int] = [] async def failing_map_file(session: AsyncSession, **kwargs: Any) -> list[str]: await real_map_file(session, **kwargs) # вставка выполнена — и откатывается вместе с транзакцией calls.append(1) raise RuntimeError("подставная ошибка БД при вставке строк") monkeypatch.setattr(rm, "map_file", failing_map_file) for attempt in (1, 2): # R-15: повторяемая неудача — `False`, цикл выдерживает паузу MAPPING_POLL_SECONDS до следующей попытки assert await process_once(maker, storage, _settings()) is False job = await _job(maker, job_id) assert job.metrics[MAPPING_ATTEMPTS_KEY] == attempt assert MAPPING_ERROR_KEY not in job.metrics assert (await _file(maker, file_id)).status is FileStatus.PROCESSING assert await _documents(maker, file_id) == [] # третья неудача переводит файл в ERROR: кандидат покинул очередь, следующий можно брать без паузы — `True` assert await process_once(maker, storage, _settings()) is True job = await _job(maker, job_id) assert job.metrics[MAPPING_ATTEMPTS_KEY] == 3 # noqa: PLR2004 - MAPPING_MAX_ATTEMPTS по умолчанию # M-1: пользователю — категория и число попыток, текст исключения остаётся только в логе assert job.metrics[MAPPING_ERROR_KEY] == "маппинг не удался: ошибка БД/S3 (RuntimeError), попыток 3" assert (await _file(maker, file_id)).status is FileStatus.ERROR assert len(calls) == 3 # noqa: PLR2004 - три попытки assert await process_once(maker, storage, _settings()) is False _INTERNALS = ("[SQL:", "input_value", "http", "sqlalche.me", "parameters") async def test_schema_error_hides_input_values(committed_sessionmaker: async_sessionmaker[AsyncSession]): """M-1: ошибка схемы результата — в `mapping_error` число ошибок и путь/тип первой, без `input_value` и ссылок pydantic (полный текст — только в логе).""" maker, storage = committed_sessionmaker, FakeStorage() result = _minimal() _osv(result)["rows"][1]["opening_debit"] = "секрет http://example.invalid" file_id, job_id, _ = await _prepare(maker, storage, result) assert await process_once(maker, storage, _settings()) is True assert (await _file(maker, file_id)).status is FileStatus.ERROR message = (await _job(maker, job_id)).metrics[MAPPING_ERROR_KEY] assert message.startswith("результат не соответствует схеме recognition-result: ошибок ") assert "первая: documents.0." in message assert "opening_debit" in message assert "секрет" not in message assert not [marker for marker in _INTERNALS if marker in message] async def test_retryable_db_error_hides_sql( committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch ): """M-1: исчерпавшая попытки ошибка БД — в `mapping_error` категория, тип исключения и число попыток, без `[SQL: …]`, параметров и ссылок SQLAlchemy.""" maker, storage = committed_sessionmaker, FakeStorage() file_id, job_id, _ = await _prepare(maker, storage, _minimal()) async def failing_map_file(session: AsyncSession, **kwargs: Any) -> list[str]: raise IntegrityError("INSERT INTO osv_entry (id) VALUES ($1)", {"id": "секрет"}, Exception("fk violation")) monkeypatch.setattr(rm, "map_file", failing_map_file) settings = _settings().model_copy(update={"MAPPING_MAX_ATTEMPTS": 1}) assert await process_once(maker, storage, settings) is True assert (await _file(maker, file_id)).status is FileStatus.ERROR message = (await _job(maker, job_id)).metrics[MAPPING_ERROR_KEY] assert message == "маппинг не удался: ошибка БД/S3 (IntegrityError), попыток 1" assert not [marker for marker in _INTERNALS if marker in message] async def test_child_before_parent_across_batches( committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch ): """T7.2: строки вставляются родитель раньше потомка независимо от порядка строк в результате — потомок перед родителем на границе пачки не нарушает self-FK `parent_entry_id`.""" maker, storage = committed_sessionmaker, FakeStorage() result = _minimal() rows = _osv(result)["rows"] rows.reverse() # потомок (level 1) раньше корня (level 0) monkeypatch.setattr(rm, "ENTRY_BATCH_SIZE", 1) # каждая строка — отдельная пачка file_id, _, _ = await _prepare(maker, storage, result) assert await process_once(maker, storage, _settings()) is True assert (await _file(maker, file_id)).status is FileStatus.PROCESSED [document] = await _documents(maker, file_id) entries = await _entries(maker, document.id) root, child = entries assert child.parent_entry_id == root.id async def test_repeated_mapping_is_idempotent(committed_sessionmaker: async_sessionmaker[AsyncSession]): """«Повторный маппинг идемпотентен»: остатки прерванной вставки (документ и половина строк) удаляются, у файла ровно один набор документов и строк.""" maker, storage = committed_sessionmaker, FakeStorage() result = _minimal() rows = _osv(result)["rows"] rows.extend( rows[-1] | {"row_number": number, "counterparty_name": f"ООО Лютик-{number}", "counterparty_inn": None} for number in range(3, 7) ) file_id, job_id, entity_id = await _prepare(maker, storage, result) async with maker() as session, session.begin(): file = await session.get(StoredFile, file_id) job = await session.get(RecognitionJob, job_id) assert file is not None and job is not None stale = await make_document(session, file=file, job=job, account="60", status=DocumentStatus.NEEDS_CHECK) for row in rows[: len(rows) // 2]: session.add( OsvEntry( recognition_job_id=job_id, document_id=stale.id, owner_legal_entity_id=entity_id, row_number=row["row_number"], level=row["level"], row_value=row["row_value"], ) ) assert await process_once(maker, storage, _settings()) is True [document] = await _documents(maker, file_id) assert document.id != stale.id entries = await _entries(maker, document.id) assert [entry.row_number for entry in entries] == [row["row_number"] for row in rows] async with maker() as session: total = ( await session.execute( select(func.count()) .select_from(OsvEntry) .join(Document, Document.id == OsvEntry.document_id) .where(Document.stored_file_id == file_id) ) ).scalar_one() assert total == len(rows) async def test_two_sheets_of_same_account(committed_sessionmaker: async_sessionmaker[AsyncSession]): """«Два листа одного счёта»: один документ (первый лист), `NEEDS_CHECK`, предупреждение называет оба `locator`.""" maker, storage = committed_sessionmaker, FakeStorage() result = _minimal() first = _osv(result) second = copy.deepcopy(first) | {"locator": "sheet:Лист_2"} second["header"]["account_code_raw"] = "60" result["documents"] = [first, second] file_id, job_id, _ = await _prepare(maker, storage, result, owner_inn="7700000000") assert await process_once(maker, storage, _settings()) is True [document] = await _documents(maker, file_id) assert document.locator == "sheet:Лист_1" assert document.osv_account_code == "60" assert document.status is DocumentStatus.NEEDS_CHECK assert len(await _entries(maker, document.id)) == len(first["rows"]) job = await _job(maker, job_id) assert job.metrics[MAPPING_WARNINGS_KEY] == ["sheet=sheet:Лист_2: счёт 60 уже есть в листе sheet:Лист_1"] assert (await _file(maker, file_id)).status is FileStatus.PROCESSED async def test_header_of_another_organization(committed_sessionmaker: async_sessionmaker[AsyncSession]): """«Шапка другой организации»: файл ЮЛ «Пекарня Счастливочная», в шапке `АО "Русский бисквит"` → `MISMATCH`, `NEEDS_CHECK`.""" maker, storage = committed_sessionmaker, FakeStorage() result = _minimal() _osv(result)["header"] |= {"organization_name": 'АО "Русский бисквит"', "organization_inn": None} file_id, _, _ = await _prepare(maker, storage, result, owner_name="ООО «Пекарня Счастливочная»") assert await process_once(maker, storage, _settings()) is True [document] = await _documents(maker, file_id) assert document.counterparty_check is DocumentCheckResult.MISMATCH assert document.status is DocumentStatus.NEEDS_CHECK async def _q1_document( maker: async_sessionmaker[AsyncSession], entity_id: uuid.UUID, *, opening_credit: Decimal, file_status: FileStatus, ) -> None: """Документ ОСВ 60 за 2024 Q1 другого файла того же ЮЛ с корнем (level 0) и строкой-потомком.""" async with maker() as session, session.begin(): entity = await session.get(LegalEntity, entity_id) assert entity is not None file, job = await make_stored_file( session, entity=entity, status=file_status, job_status=RecognitionStatus.done ) document = await make_document(session, file=file, job=job, account="60", year=2024, quarter=PeriodQuarter.Q1) common = {"recognition_job_id": job.id, "document_id": document.id, "owner_legal_entity_id": entity_id} root = OsvEntry( **common, row_number=5, level=0, row_value="60", opening_debit=None, opening_credit=opening_credit ) session.add(root) await session.flush() # строка уровня 1 с меньшим номером не должна считаться корнем session.add(OsvEntry(**common, row_number=1, level=1, parent_entry_id=root.id, opening_credit=Decimal(1))) @pytest.mark.parametrize( ("q1_opening", "q1_file_status", "expected"), [ (Decimal("1000.00"), FileStatus.PROCESSED, True), # сальдо на начало совпало с Q1 (Decimal("1000.01"), FileStatus.PROCESSED, True), # в пределах ±0.01 (Decimal("999.00"), FileStatus.PROCESSED, False), # не совпало — по тексту (Decimal("1000.00"), FileStatus.SUPERSEDED, False), # Q1 из заменённого файла не учитывается ], ) async def test_cumulative_by_q1_data( committed_sessionmaker: async_sessionmaker[AsyncSession], q1_opening: Decimal, q1_file_status: FileStatus, expected: bool, ): """«Нарастающая по данным»: ОСВ 60 «за 3 квартал 2024» с `opening_*` корня = Q1 → `is_cumulative = true` и предупреждение «нарастающая по данным Q1» (ASSUMPTION(OQ-02)).""" maker, storage = committed_sessionmaker, FakeStorage() async with maker() as session, session.begin(): entity = await make_legal_entity(session, name=OWNER_NAME, inn="7700000000") await _q1_document(maker, entity.id, opening_credit=q1_opening, file_status=q1_file_status) result = _minimal() _osv(result)["header"] |= { "period_text": "3 квартал 2024 г.", "period_year": 2024, "period_quarter": "Q3", "is_cumulative": False, } assert _root(_osv(result)["rows"])["opening_credit"] == "1000.00" file_id, job_id, _ = await _prepare(maker, storage, result, entity=entity) assert await process_once(maker, storage, _settings()) is True [document] = await _documents(maker, file_id) assert document.period_quarter is PeriodQuarter.Q3 assert document.is_cumulative is expected warnings = (await _job(maker, job_id)).metrics[MAPPING_WARNINGS_KEY] if expected: assert len(warnings) == 1 assert "нарастающая по данным Q1" in warnings[0] assert "sheet:Лист_1" in warnings[0] assert document.status is DocumentStatus.NEEDS_CHECK else: assert warnings == [] assert document.status is DocumentStatus.CONFIRMED # чистый → автоподтверждение (OQ-157) @pytest.mark.parametrize( ("job_status", "expected"), [(RecognitionStatus.done, DocumentStatus.CONFIRMED), (RecognitionStatus.needs_check, DocumentStatus.NEEDS_CHECK)], ) async def test_clean_result( committed_sessionmaker: async_sessionmaker[AsyncSession], job_status: RecognitionStatus, expected: DocumentStatus, ): """«Чистый результат»: задание `done`, шапка = владелец (ИНН), нарушений нет → `CONFIRMED` (автоподтверждение, OQ-157), `MATCH`, `UNKNOWN`; задание `needs_check` → `NEEDS_CHECK`.""" maker, storage = committed_sessionmaker, FakeStorage() file_id, _, _ = await _prepare(maker, storage, _minimal(), owner_inn="7700000000", job_status=job_status) assert await process_once(maker, storage, _settings()) is True [document] = await _documents(maker, file_id) assert document.status is expected assert document.counterparty_check is DocumentCheckResult.MATCH assert document.period_check is DocumentCheckResult.UNKNOWN assert document.osv_account_code == "60" assert document.period_quarter is PeriodQuarter.Q1 assert document.is_cumulative is False assert (await _file(maker, file_id)).status is FileStatus.PROCESSED async def test_counterparty_resolved_in_entries(committed_sessionmaker: async_sessionmaker[AsyncSession]): """R-3: `counterparty_id` — только у `COUNTERPARTY` с `data_type ∈ {BU, SUM}`; `NU` хранится как пришёл, без резолва; строка-счёт без контрагента.""" maker, storage = committed_sessionmaker, FakeStorage() async with maker() as session, session.begin(): lyutik = await make_legal_entity(session, name="ООО «Лютик»", inn="7711111111") result = _minimal() rows = _osv(result)["rows"] rows.append(rows[1] | {"row_number": 3, "data_type": "NU"}) file_id, _, _ = await _prepare(maker, storage, result) assert await process_once(maker, storage, _settings()) is True [document] = await _documents(maker, file_id) account, bu, nu = await _entries(maker, document.id) assert account.group_type is OsvGroupType.ACCOUNT assert account.counterparty_id is None assert bu.counterparty_id == lyutik.id assert bu.counterparty_inn == "7711111111" # v5.16: способ сопоставления хранится ровно у сопоставленных строк (здесь — по ИНН) assert (bu.counterparty_match_kind, account.counterparty_match_kind, nu.counterparty_match_kind) == ( CounterpartyMatchKind.INN, None, None, ) assert nu.data_type is OsvDataType.NU assert nu.counterparty_id is None async def test_result_of_non_current_job_is_ignored(committed_sessionmaker: async_sessionmaker[AsyncSession]): """Review Focus 1: результат задания, которое уже не текущее для файла, не выбирается и файл не трогает.""" maker, storage = committed_sessionmaker, FakeStorage() file_id, old_job_id, _ = await _prepare(maker, storage, _minimal()) async with maker() as session, session.begin(): file = await session.get(StoredFile, file_id) assert file is not None new_job = RecognitionJob( file_id=file_id, pipeline=(await session.get(RecognitionJob, old_job_id)).pipeline, # type: ignore[union-attr] request_subject="files.tabular_uploaded", schema_version="1.0", status=RecognitionStatus.processing, attempt=2, ) session.add(new_job) await session.flush() file.current_recognition_job_id = new_job.id before = await _file(maker, file_id) assert await process_once(maker, storage, _settings()) is False after = await _file(maker, file_id) assert (after.status, after.updated_at) == (FileStatus.PROCESSING, before.updated_at) assert await _documents(maker, file_id) == [] assert (await _job(maker, old_job_id)).group_schema is None async def test_job_replaced_while_reading_result( committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch ): """Review Focus 1, гонка: пока цикл читал S3, у файла сменилось текущее задание → транзакция маппинга ничего не меняет (задание и файл перечитываются под блокировкой).""" maker, storage = committed_sessionmaker, FakeStorage() file_id, job_id, _ = await _prepare(maker, storage, _minimal()) real_get_bytes = storage.get_bytes async def get_bytes_and_replace_job(key: str) -> bytes: async with maker() as session, session.begin(): file = await session.get(StoredFile, file_id) job = await session.get(RecognitionJob, job_id) assert file is not None and job is not None new_job = RecognitionJob( file_id=file_id, pipeline=job.pipeline, request_subject=job.request_subject, schema_version="1.0", status=RecognitionStatus.queued, attempt=2, ) session.add(new_job) await session.flush() file.current_recognition_job_id = new_job.id return await real_get_bytes(key) monkeypatch.setattr(storage, "get_bytes", get_bytes_and_replace_job) assert await process_once(maker, storage, _settings()) is True file = await _file(maker, file_id) assert file.status is FileStatus.PROCESSING assert await _documents(maker, file_id) == [] job = await _job(maker, job_id) assert job.group_schema is None assert MAPPING_ATTEMPTS_KEY not in (job.metrics or {}) async def test_run_mapping_loop_processes_two_candidates(committed_sessionmaker: async_sessionmaker[AsyncSession]): """`run_mapping_loop`: два кандидата подряд без паузы (`sleep(0)`), затем пауза `MAPPING_POLL_SECONDS`.""" maker, storage = committed_sessionmaker, FakeStorage() first, _, _ = await _prepare(maker, storage, _minimal()) second, _, _ = await _prepare(maker, storage, _minimal()) settings = _settings() sleeps: list[float] = [] async def fake_sleep(seconds: float) -> None: sleeps.append(seconds) if len(sleeps) == 3: # noqa: PLR2004 - два кандидата и один пустой проход raise asyncio.CancelledError with pytest.raises(asyncio.CancelledError): await run_mapping_loop(maker, storage, settings, sleep=fake_sleep) assert sleeps == [0, 0, settings.MAPPING_POLL_SECONDS] assert (await _file(maker, first)).status is FileStatus.PROCESSED assert (await _file(maker, second)).status is FileStatus.PROCESSED async def test_run_mapping_loop_pauses_after_retryable_failure( committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch ): """R-15: после повторяемой неудачи цикл спит `MAPPING_POLL_SECONDS` (попытки не сгорают за миллисекунды), после успешного маппинга — `sleep(0)`.""" maker, storage = committed_sessionmaker, FakeStorage() file_id, job_id, _ = await _prepare(maker, storage, _minimal()) real_map_file = rm.map_file failures = [RuntimeError("кратковременный сбой БД")] async def flaky_map_file(session: AsyncSession, **kwargs: Any) -> list[str]: if failures: raise failures.pop() return await real_map_file(session, **kwargs) monkeypatch.setattr(rm, "map_file", flaky_map_file) settings = _settings() sleeps: list[float] = [] async def fake_sleep(seconds: float) -> None: sleeps.append(seconds) if len(sleeps) == 3: # noqa: PLR2004 - неудача, успех, пустой проход raise asyncio.CancelledError with pytest.raises(asyncio.CancelledError): await run_mapping_loop(maker, storage, settings, sleep=fake_sleep) assert sleeps == [settings.MAPPING_POLL_SECONDS, 0, settings.MAPPING_POLL_SECONDS] assert (await _file(maker, file_id)).status is FileStatus.PROCESSED assert (await _job(maker, job_id)).metrics[MAPPING_ATTEMPTS_KEY] == 1 async def test_run_mapping_loop_survives_errors(monkeypatch: pytest.MonkeyPatch): """Ошибка прохода логируется, цикл не умирает и ждёт `MAPPING_POLL_SECONDS`.""" outcomes: list[Callable[[], bool]] = [lambda: (_ for _ in ()).throw(RuntimeError("сбой")), lambda: True] sleeps: list[float] = [] async def fake_process_once(*args: Any, **kwargs: Any) -> bool: return outcomes.pop(0)() async def fake_sleep(seconds: float) -> None: sleeps.append(seconds) if not outcomes: raise asyncio.CancelledError monkeypatch.setattr(rm, "process_once", fake_process_once) settings = _settings() records: list[Any] = [] handler_id = logger.add(lambda message: records.append(message.record), level="ERROR") try: with pytest.raises(asyncio.CancelledError): await run_mapping_loop(None, FakeStorage(), settings, sleep=fake_sleep) # type: ignore[arg-type] finally: logger.remove(handler_id) assert sleeps == [settings.MAPPING_POLL_SECONDS, 0] assert len(records) == 1 # --- автозамена прежних версий (spec primary-files «Автозамена прежних версий», audit-trail; design D-10) ---------- OldDocument = tuple[str, int, PeriodQuarter, DocumentStatus] # счёт, год, квартал, статус документа прежнего файла OSV_60_2025_Q1: OldDocument = ("60", 2025, PeriodQuarter.Q1, DocumentStatus.CONFIRMED) async def _old_file( # noqa: PLR0913 - каждое поле прежнего файла задаётся тестом явно maker: async_sessionmaker[AsyncSession], entity_id: uuid.UUID, documents: list[OldDocument], *, version_no: int = 1, status: FileStatus = FileStatus.PROCESSED, superseded_by: uuid.UUID | None = None, ) -> uuid.UUID: """Прежний файл ЮЛ с документами ОСВ и строкой ОСВ на каждый документ (история должна сохраниться).""" async with maker() as session, session.begin(): entity = await session.get(LegalEntity, entity_id) assert entity is not None file, job = await make_stored_file( session, entity=entity, status=status, job_status=RecognitionStatus.done, original_name="old.xls" ) file.version_no = version_no file.superseded_by_file_id = superseded_by for account, year, quarter, document_status in documents: document = await make_document( session, file=file, job=job, account=account, year=year, quarter=quarter, status=document_status ) session.add( OsvEntry( recognition_job_id=job.id, document_id=document.id, owner_legal_entity_id=entity_id, row_number=1, level=0, row_value=account, ) ) return file.id async def _entity( maker: async_sessionmaker[AsyncSession], name: str = OWNER_NAME, inn: str = "7700000000" ) -> uuid.UUID: async with maker() as session, session.begin(): return (await make_legal_entity(session, name=name, inn=inn)).id async def _any_user(maker: async_sessionmaker[AsyncSession]) -> uuid.UUID: async with maker() as session: return (await session.execute(select(AppUser.id).order_by(AppUser.email).limit(1))).scalar_one() async def _replace_events(maker: async_sessionmaker[AsyncSession]) -> list[AuditEvent]: async with maker() as session: rows = await session.execute( select(AuditEvent).where(AuditEvent.action == AuditAction.FILE_REPLACE).order_by(AuditEvent.occurred_at) ) return list(rows.scalars()) async def _active_files(maker: async_sessionmaker[AsyncSession], entity_id: uuid.UUID) -> set[uuid.UUID]: """Файлы ЮЛ, которые список файлов показывает по умолчанию (без `include_superseded`).""" async with maker() as session: rows = await session.execute( select(StoredFile.id).where( StoredFile.legal_entity_id == entity_id, StoredFile.status != FileStatus.SUPERSEDED ) ) return set(rows.scalars()) @pytest.mark.parametrize( ("old_status", "confirmed"), [(DocumentStatus.CONFIRMED, "confirmed=true"), (DocumentStatus.RECOGNIZED, "confirmed=false")], ) async def test_new_osv_60_supersedes_previous_file( committed_sessionmaker: async_sessionmaker[AsyncSession], old_status: DocumentStatus, confirmed: str ): """«Новая ОСВ 60 за 1 кв. 2025»: прежний файл (документ `CONFIRMED` тоже, решение владельца 05.10.2026) — `SUPERSEDED`, `superseded_by_file_id` = новый, у нового `version_no = 2`; `FILE_REPLACE` — системное событие (`user_id` NULL, инициатор загрузки — в `description`); документы и строки прежнего файла сохранены.""" maker, storage = committed_sessionmaker, FakeStorage() entity_id = await _entity(maker) uploader = await _any_user(maker) old_id = await _old_file(maker, entity_id, [("60", 2025, PeriodQuarter.Q1, old_status)]) async with maker() as session: entity = await session.get(LegalEntity, entity_id) new_id, _, _ = await _prepare(maker, storage, _minimal(), entity=entity, uploaded_by=uploader) assert await process_once(maker, storage, _settings()) is True old, new = await _file(maker, old_id), await _file(maker, new_id) assert old.status is FileStatus.SUPERSEDED assert old.superseded_by_file_id == new_id assert old.updated_at is not None assert (new.status, new.version_no, new.superseded_by_file_id) == (FileStatus.PROCESSED, 2, None) assert await _active_files(maker, entity_id) == {new_id} [old_document] = await _documents(maker, old_id) # история: документ и строки прежнего файла на месте assert old_document.status is old_status assert len(await _entries(maker, old_document.id)) == 1 [event] = await _replace_events(maker) 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, "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 assert confirmed in event.description @pytest.mark.parametrize( "document", [ ("62", 2025, PeriodQuarter.Q1, DocumentStatus.CONFIRMED), # другой счёт ("60", 2024, PeriodQuarter.Q1, DocumentStatus.CONFIRMED), # тот же счёт, другой год ("60", 2025, PeriodQuarter.Q2, DocumentStatus.CONFIRMED), # тот же счёт, другой квартал ], ) async def test_other_document_is_not_superseded( committed_sessionmaker: async_sessionmaker[AsyncSession], document: OldDocument ): """Два разных счёта (или периода) не заменяют друг друга: прежний файл `PROCESSED`, у нового `version_no = 1`.""" maker, storage = committed_sessionmaker, FakeStorage() entity_id = await _entity(maker) old_id = await _old_file(maker, entity_id, [document]) async with maker() as session: entity = await session.get(LegalEntity, entity_id) new_id, _, _ = await _prepare(maker, storage, _minimal(), entity=entity) assert await process_once(maker, storage, _settings()) is True old, new = await _file(maker, old_id), await _file(maker, new_id) assert (old.status, old.superseded_by_file_id) == (FileStatus.PROCESSED, None) assert (new.status, new.version_no) == (FileStatus.PROCESSED, 1) assert await _replace_events(maker) == [] async def test_other_legal_entity_is_not_superseded(committed_sessionmaker: async_sessionmaker[AsyncSession]): """Тот же документ у другого ЮЛ не затрагивается: замена — только в пределах ЮЛ-владельца.""" maker, storage = committed_sessionmaker, FakeStorage() other_id = await _entity(maker, name="ООО «Другое ЮЛ»", inn="7799999999") old_id = await _old_file(maker, other_id, [OSV_60_2025_Q1]) new_id, _, _ = await _prepare(maker, storage, _minimal(), owner_inn="7700000000") assert await process_once(maker, storage, _settings()) is True assert (await _file(maker, old_id)).status is FileStatus.PROCESSED assert (await _file(maker, new_id)).version_no == 1 assert await _replace_events(maker) == [] async def test_supersede_is_idempotent( committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch ): """Сбой после автозамены откатывает и замену, и аудит; повтор маппинга даёт ровно одно `FILE_REPLACE`. Повторный вызов `supersede_previous` для того же файла ничего не находит (заменённый уже `SUPERSEDED`).""" maker, storage = committed_sessionmaker, FakeStorage() entity_id = await _entity(maker) old_id = await _old_file(maker, entity_id, [OSV_60_2025_Q1]) async with maker() as session: entity = await session.get(LegalEntity, entity_id) new_id, _, _ = await _prepare(maker, storage, _minimal(), entity=entity) real_map_file = rm.map_file failures = [RuntimeError("сбой БД после автозамены")] async def flaky_map_file(session: AsyncSession, **kwargs: Any) -> list[str]: warnings = await real_map_file(session, **kwargs) if failures: raise failures.pop() return warnings monkeypatch.setattr(rm, "map_file", flaky_map_file) assert await process_once(maker, storage, _settings()) is False # повторяемая неудача: всё откачено assert (await _file(maker, old_id)).status is FileStatus.PROCESSED assert await _replace_events(maker) == [] assert await process_once(maker, storage, _settings()) is True assert (await _file(maker, old_id)).status is FileStatus.SUPERSEDED assert len(await _replace_events(maker)) == 1 async with maker() as session, session.begin(): new = await session.get(StoredFile, new_id) assert new is not None documents = list((await session.execute(select(Document).where(Document.stored_file_id == new_id))).scalars()) replaced = await supersede_previous( session, new_file=new, new_documents=documents, uploader_id=None, now=_now() ) assert replaced == [] assert len(await _replace_events(maker)) == 1 assert (await _file(maker, new_id)).version_no == 2 # noqa: PLR2004 - версия не растёт от повторного вызова async def test_version_chain(committed_sessionmaker: async_sessionmaker[AsyncSession]): """Цепочка v1 → v2 → v3: третий файл заменяет только действующий v2 и получает `version_no = 3`; v1 остаётся заменённым файлом v2.""" maker, storage = committed_sessionmaker, FakeStorage() entity_id = await _entity(maker) v2_id = await _old_file(maker, entity_id, [OSV_60_2025_Q1], version_no=2) v1_id = await _old_file( maker, entity_id, [OSV_60_2025_Q1], version_no=1, status=FileStatus.SUPERSEDED, superseded_by=v2_id ) async with maker() as session: entity = await session.get(LegalEntity, entity_id) v3_id, _, _ = await _prepare(maker, storage, _minimal(), entity=entity) assert await process_once(maker, storage, _settings()) is True v1, v2, v3 = await _file(maker, v1_id), await _file(maker, v2_id), await _file(maker, v3_id) assert (v1.status, v1.superseded_by_file_id) == (FileStatus.SUPERSEDED, v2_id) assert (v2.status, v2.superseded_by_file_id) == (FileStatus.SUPERSEDED, v3_id) 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, "unbound_version_ids": []} async def _consolidation_id( maker: async_sessionmaker[AsyncSession], entity_id: uuid.UUID, quarter: PeriodQuarter ) -> uuid.UUID: """Консолидация ГК этого ЮЛ за 2025/quarter (без версий: для наследования они не нужны). ЮЛ без ГК получает ГК (консолидация требует группу) и привязывается к ней.""" async with maker() as session, session.begin(): entity = await session.get(LegalEntity, entity_id) assert entity is not None if entity.company_group_id is None: group = await make_company_group(session, name="ГК Наследование") entity.company_group_id = group.id consolidation = Consolidation( company_group_id=entity.company_group_id, year=2025, quarter=quarter, reporting_date=reporting_date(2025, quarter), created_at=_now(), ) session.add(consolidation) await session.flush() return consolidation.id async def _set_consolidation( maker: async_sessionmaker[AsyncSession], file_id: uuid.UUID, consolidation_id: uuid.UUID | None ) -> None: async with maker() as session, session.begin(): await session.execute( update(StoredFile).where(StoredFile.id == file_id).values(consolidation_id=consolidation_id) ) async def test_supersede_inherits_consolidation_from_first_with_value( committed_sessionmaker: async_sessionmaker[AsyncSession], ) -> None: """Спека «Новый файл без контекста наследует консолидацию» + Review Focus 5: два заменяемых файла, у первого по `id` `consolidation_id` пуст, у второго задан — новый получает значение второго.""" maker, storage = committed_sessionmaker, FakeStorage() entity_id = await _entity(maker) c1 = await _consolidation_id(maker, entity_id, PeriodQuarter.Q1) first_id = await _old_file(maker, entity_id, [OSV_60_2025_Q1]) second_id = await _old_file(maker, entity_id, [OSV_60_2025_Q1]) empty_id, filled_id = sorted([first_id, second_id]) await _set_consolidation(maker, filled_id, c1) async with maker() as session: entity = await session.get(LegalEntity, entity_id) new_id, _, _ = await _prepare(maker, storage, _minimal(), entity=entity) assert await process_once(maker, storage, _settings()) is True assert (await _file(maker, new_id)).consolidation_id == c1 assert (await _file(maker, empty_id)).status is FileStatus.SUPERSEDED assert (await _file(maker, filled_id)).status is FileStatus.SUPERSEDED async def test_supersede_keeps_own_consolidation(committed_sessionmaker: async_sessionmaker[AsyncSession]) -> None: """Спека «Новый файл с контекстом сохраняет свой»: старый в C1, новый загружен в C2 — остаётся C2.""" maker, storage = committed_sessionmaker, FakeStorage() entity_id = await _entity(maker) c1 = await _consolidation_id(maker, entity_id, PeriodQuarter.Q1) c2 = await _consolidation_id(maker, entity_id, PeriodQuarter.Q2) old_id = await _old_file(maker, entity_id, [OSV_60_2025_Q1]) await _set_consolidation(maker, old_id, c1) async with maker() as session: entity = await session.get(LegalEntity, entity_id) new_id, _, _ = await _prepare(maker, storage, _minimal(), entity=entity) await _set_consolidation(maker, new_id, c2) assert await process_once(maker, storage, _settings()) is True assert (await _file(maker, new_id)).consolidation_id == c2 assert (await _file(maker, old_id)).status is FileStatus.SUPERSEDED async def test_one_event_per_replaced_file(committed_sessionmaker: async_sessionmaker[AsyncSession]): """Новый файл с ОСВ 60 и 62 заменяет прежний файл с теми же двумя документами одним событием `FILE_REPLACE`, в `description` — оба документа.""" maker, storage = committed_sessionmaker, FakeStorage() entity_id = await _entity(maker) old_id = await _old_file( maker, entity_id, [OSV_60_2025_Q1, ("62", 2025, PeriodQuarter.Q1, DocumentStatus.NEEDS_CHECK)], ) result = _minimal() second = copy.deepcopy(_osv(result)) | {"locator": "sheet:Лист_2"} second["header"]["account_code_raw"] = "62" result["documents"].append(second) async with maker() as session: entity = await session.get(LegalEntity, entity_id) new_id, _, _ = await _prepare(maker, storage, result, entity=entity) assert await process_once(maker, storage, _settings()) is True assert (await _file(maker, old_id)).superseded_by_file_id == new_id assert len(await _documents(maker, new_id)) == 2 # noqa: PLR2004 - два листа, два счёта [event] = await _replace_events(maker) assert event.object_id == old_id assert event.description is not None assert "OSV 60 2025 Q1 (CONFIRMED)" in event.description assert "OSV 62 2025 Q1 (NEEDS_CHECK)" in event.description assert "confirmed=true" in event.description async def test_new_file_supersedes_several_files(committed_sessionmaker: async_sessionmaker[AsyncSession]): """Один новый файл (ОСВ 60, 62, 66) заменяет три разных прежних файла: событие `FILE_REPLACE` на каждый заменённый файл, у нового `version_no` = МАКСИМУМ заменённых + 1. Версии раздаются по возрастанию `id` как 1, 3, 2: максимум — в середине порядка блокировки, поэтому ни первый, ни последний заблокированный файл его не дают.""" maker, storage = committed_sessionmaker, FakeStorage() entity_id = await _entity(maker) old_ids = [ await _old_file(maker, entity_id, [(account, 2025, PeriodQuarter.Q1, DocumentStatus.RECOGNIZED)]) for account in ("60", "62", "66") ] ordered = sorted(old_ids) versions = dict(zip(ordered, (1, 3, 2), strict=True)) async with maker() as session, session.begin(): for file_id, version_no in versions.items(): file = await session.get(StoredFile, file_id) assert file is not None file.version_no = version_no result = _minimal() for number, account in enumerate(("62", "66"), start=2): sheet = copy.deepcopy(_osv(result)) | {"locator": f"sheet:Лист_{number}"} sheet["header"]["account_code_raw"] = account result["documents"].append(sheet) async with maker() as session: entity = await session.get(LegalEntity, entity_id) new_id, _, _ = await _prepare(maker, storage, result, entity=entity) assert await process_once(maker, storage, _settings()) is True expected_version = 4 # max(1, 3, 2) + 1 for old_id in old_ids: old = await _file(maker, old_id) assert (old.status, old.superseded_by_file_id) == (FileStatus.SUPERSEDED, new_id) assert (await _file(maker, new_id)).version_no == expected_version events = await _replace_events(maker) 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, "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)]} # --- OQ-154: предупреждение о документах заменённого файла, которых нет в новом ----------------------------------- STATEMENT_MINIMAL = API_DIR / "contracts" / "examples" / "result-statement-minimal.json" def _statement_result(*document_types: str) -> dict[str, Any]: """Результат с формами `document_types` за 30.09.2025 (Q3 2025), копии `result-statement-minimal.json`.""" result = json.loads(STATEMENT_MINIMAL.read_text("utf-8")) template = result["documents"][0] result["documents"] = [] for number, document_type in enumerate(document_types, start=1): document = copy.deepcopy(template) | {"document_type": document_type, "locator": f"pages:{number}"} result["documents"].append(document) return result async def _mapping_warnings(maker: async_sessionmaker[AsyncSession], job_id: uuid.UUID) -> list[str]: job = await _job(maker, job_id) assert isinstance(job.metrics, dict) warnings: list[str] = job.metrics[MAPPING_WARNINGS_KEY] return warnings async def test_replace_warns_about_dropped_confirmed_balance( committed_sessionmaker: async_sessionmaker[AsyncSession], ): """OQ-154: книга «ББ + ОФР» за Q3 2025, ББ подтверждён и привязан к версии «в работе»; отдельный файл только с ОФР того же периода заменяет всю книгу, а в `mapping_warnings` нового задания — имя прежнего файла, выпавший `BALANCE 2025 Q3 (CONFIRMED)` и версия, с которой снята привязка. Замена и статус документа ОФР не меняются.""" maker, storage = committed_sessionmaker, FakeStorage() book_id, book_job_id, entity_id = await _prepare( maker, storage, _statement_result("BALANCE", "INCOME"), owner_inn="7700000000" ) assert await process_once(maker, storage, _settings()) is True assert await _mapping_warnings(maker, book_job_id) == [] # книга ничего не заменяла: предупреждений нет async with maker() as session, session.begin(): initiator = (await session.execute(select(AppUser).order_by(AppUser.email).limit(1))).scalar_one() entity = await session.get(LegalEntity, entity_id) assert entity is not None group = await make_company_group(session, name="ГК Книга") entity.company_group_id = group.id _, version = await make_consolidation( session, group=group, year=2025, quarter=PeriodQuarter.Q3, members=[entity], initiator=initiator, status=ConsolidationStatus.DATA_CORRECTION, ) documents = { doc.document_type: doc for doc in (await session.execute(select(Document).where(Document.stored_file_id == book_id))).scalars() } assert set(documents) == {DocumentType.BALANCE, DocumentType.INCOME} documents[DocumentType.BALANCE].status = DocumentStatus.CONFIRMED session.add( ConsolidationDocument(consolidation_version_id=version.id, document_id=documents[DocumentType.BALANCE].id) ) version_id = version.id async with maker() as session: entity = await session.get(LegalEntity, entity_id) new_id, new_job_id, _ = await _prepare(maker, storage, _statement_result("INCOME"), entity=entity) assert await process_once(maker, storage, _settings()) is True [warning] = await _mapping_warnings(maker, new_job_id) book_name = (await _file(maker, book_id)).original_name assert warning.startswith(f"автозамена: файл «{book_name}» (версия 1) заменён этим файлом") assert "BALANCE 2025 Q3 (CONFIRMED)" in warning assert "INCOME" not in warning # ОФР есть в новом файле: не выпал assert "ГК Книга" in warning and "2025 Q3, версия 1" in warning book = await _file(maker, book_id) assert (book.status, book.superseded_by_file_id) == (FileStatus.SUPERSEDED, new_id) # поведение замены прежнее async with maker() as session: bound = ( await session.execute( select(ConsolidationDocument).where(ConsolidationDocument.consolidation_version_id == version_id) ) ).all() assert bound == [] [event] = await _replace_events(maker) assert event.after_value is not None and event.after_value["unbound_version_ids"] == [str(version_id)] async def test_replace_without_dropped_documents_has_no_warning( committed_sessionmaker: async_sessionmaker[AsyncSession], ): """Новый файл содержит ключи всех документов прежнего — предупреждения нет (обычная автозамена).""" maker, storage = committed_sessionmaker, FakeStorage() entity_id = await _entity(maker) await _old_file(maker, entity_id, [OSV_60_2025_Q1]) async with maker() as session: entity = await session.get(LegalEntity, entity_id) _, new_job_id, _ = await _prepare(maker, storage, _minimal(), entity=entity) assert await process_once(maker, storage, _settings()) is True assert await _mapping_warnings(maker, new_job_id) == [] async def test_supersede_previous_reports_dropped_documents_without_changing_replace( committed_sessionmaker: async_sessionmaker[AsyncSession], ): """`supersede_previous(warnings=…)`: по предупреждению на заменённый файл; выпавшие — и подтверждённый, и нет, и документ другого периода; версия «не в работе» в предупреждении не фигурирует (привязка не снималась).""" 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) consolidation, in_work = await make_consolidation( session, group=group, year=2025, quarter=PeriodQuarter.Q1, members=[entity], initiator=initiator, status=ConsolidationStatus.DATA_CORRECTION, ) cancelled = await make_version( session, consolidation=consolidation, version_no=2, members=[entity], initiator=initiator, status=ConsolidationStatus.CANCELLED, ) old_file, old_job = await make_stored_file( session, entity=entity, status=FileStatus.PROCESSED, job_status=RecognitionStatus.done ) await make_document(session, file=old_file, job=old_job, account="60", status=DocumentStatus.CONFIRMED) old_62 = await make_document(session, file=old_file, job=old_job, account="62", status=DocumentStatus.CONFIRMED) old_66 = await make_document( session, file=old_file, job=old_job, account="66", status=DocumentStatus.NEEDS_CHECK ) for version in (in_work, cancelled): for document in (old_62, old_66): 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") new_file_id, in_work_id = new_file.id, in_work.id warnings: list[str] = [] 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(), warnings=warnings ) assert len(replaced) == 1 [warning] = warnings assert "OSV 62 2025 Q1 (CONFIRMED)" in warning and "OSV 66 2025 Q1 (NEEDS_CHECK)" in warning assert "OSV 60" not in warning # совпал по ключу — не выпал assert "«ГК Выпавшие» 2025 Q1, версия 1" in warning assert "версия 2" not in warning.split("привязки сняты", 1)[1] # CANCELLED-версия не затронута [event] = await _replace_events(maker) assert event.after_value is not None and event.after_value["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 / CONFIRMED (чистый) ) 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` = загрузивший, привязок нет; чистый документ → `CONFIRMED`, привязка и тот же переход (OQ-157).""" 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.CONFIRMED) async with maker() as session: version = await session.get(ConsolidationVersion, version_id) assert version is not None history = await _history_of(maker, version_id) bindings = (await session.execute(select(func.count()).select_from(ConsolidationDocument))).scalar_one() if moved: assert bindings == 0 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: # чистый документ подтверждён системой (OQ-157): привязан, версия перешла от загрузившего с комментарием assert bindings == 1 assert version.status is ConsolidationStatus.DATA_CORRECTION last = history[-1] assert (last.from_status, last.to_status, last.changed_by, last.comment) == ( ConsolidationStatus.DOCUMENTS_PROCESSING, ConsolidationStatus.DATA_CORRECTION, user_id, "автоподтверждение документа OSV 60", ) 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 _history_of( maker: async_sessionmaker[AsyncSession], version_id: uuid.UUID ) -> list[ConsolidationStatusHistory]: async with maker() as session: stmt = ( select(ConsolidationStatusHistory) .where(ConsolidationStatusHistory.version_id == version_id) .order_by(ConsolidationStatusHistory.changed_at) ) return list((await session.execute(stmt)).scalars()) async def _bound_ids(maker: async_sessionmaker[AsyncSession], version_id: uuid.UUID) -> set[uuid.UUID]: async with maker() as session: stmt = select(ConsolidationDocument.document_id).where( ConsolidationDocument.consolidation_version_id == version_id ) return set((await session.execute(stmt)).scalars()) async def test_clean_document_is_auto_confirmed_bound_and_moves_version( committed_sessionmaker: async_sessionmaker[AsyncSession], ): """osv-result-mapping «Привязка и переход версии»: чистая ОСВ 60 за 2025 Q1 → `CONFIRMED` системой, привязана к версии 2025 Q3 (`DOCUMENTS_PROCESSING`), переход в `DATA_CORRECTION` от загрузившего с комментарием.""" 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), uploaded_by=user_id ) assert await process_once(maker, storage, _settings()) is True [document] = await _documents(maker, file_id) assert (document.status, document.confirmed_by) == (DocumentStatus.CONFIRMED, None) assert document.confirmed_at is not None assert await _bound_ids(maker, version_id) == {document.id} last = (await _history_of(maker, version_id))[-1] assert (last.to_status, last.changed_by, last.comment) == ( ConsolidationStatus.DATA_CORRECTION, user_id, "автоподтверждение документа OSV 60", ) async with maker() as session: events = ( ( await session.execute( select(AuditEvent).where( AuditEvent.action == AuditAction.DOCUMENT_CONFIRM, AuditEvent.object_id == document.id ) ) ) .scalars() .all() ) assert [(event.user_id, event.description) for event in events] == [(None, "автоподтверждение: без замечаний")] async def test_mixed_file_confirms_clean_and_moves_version_once( committed_sessionmaker: async_sessionmaker[AsyncSession], ): """Review Focus 1: в файле чистый и `NEEDS_CHECK` документы — версия получает ровно один переход.""" maker, storage = committed_sessionmaker, FakeStorage() entity_id, version_id, user_id = await _version_in(maker) result = _minimal() second = copy.deepcopy(result["documents"][0]) second["header"]["account_code_raw"] = "62" second["locator"] = "sheet:Лист_2" second["checks"] = {"total_row_matches_root": False} # нарушение проверок → NEEDS_CHECK result["documents"].append(second) file_id, _, _ = await _prepare( maker, storage, result, entity=await _load_entity(maker, entity_id), uploaded_by=user_id ) assert await process_once(maker, storage, _settings()) is True statuses = sorted((d.osv_account_code, d.status) for d in await _documents(maker, file_id)) assert statuses == [("60", DocumentStatus.CONFIRMED), ("62", DocumentStatus.NEEDS_CHECK)] history = await _history_of(maker, version_id) assert [h.to_status for h in history].count(ConsolidationStatus.DATA_CORRECTION) == 1 # переход сделал именно автоподтверждение (T4.2): при перестановке с on_documents_mapped комментарий иной assert history[-1].to_status is ConsolidationStatus.DATA_CORRECTION assert history[-1].comment == "автоподтверждение документа OSV 60" clean = next(d for d in await _documents(maker, file_id) if d.osv_account_code == "60") assert await _bound_ids(maker, version_id) == {clean.id} # NEEDS_CHECK-документ 62 не привязан async def test_supersede_rebinds_auto_confirmed_document(committed_sessionmaker: async_sessionmaker[AsyncSession]): """Review Focus 2: новая чистая версия файла того же ключа заменяет привязку старой.""" maker, storage = committed_sessionmaker, FakeStorage() entity_id, version_id, user_id = await _version_in(maker) entity = await _load_entity(maker, entity_id) first_id, _, _ = await _prepare(maker, storage, _minimal(), entity=entity, uploaded_by=user_id) assert await process_once(maker, storage, _settings()) is True [first] = await _documents(maker, first_id) assert await _bound_ids(maker, version_id) == {first.id} second_id, _, _ = await _prepare(maker, storage, _minimal(), entity=entity, uploaded_by=user_id) assert await process_once(maker, storage, _settings()) is True [second] = await _documents(maker, second_id) assert second.status is DocumentStatus.CONFIRMED assert await _bound_ids(maker, version_id) == {second.id} assert (await _file(maker, first_id)).status is FileStatus.SUPERSEDED async def test_auto_confirm_rolls_back_with_mapping( committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch ): """Review Focus 3: сбой после автоподтверждения откатывает и его; повтор — одно событие `DOCUMENT_CONFIRM`.""" maker, storage = committed_sessionmaker, FakeStorage() entity_id, _, user_id = await _version_in(maker) file_id, job_id, _ = await _prepare( maker, storage, _minimal(), entity=await _load_entity(maker, entity_id), uploaded_by=user_id ) original = consolidation_documents.on_documents_mapped calls = {"n": 0} async def flaky(*args: Any, **kwargs: Any) -> list[uuid.UUID]: calls["n"] += 1 if calls["n"] == 1: raise RuntimeError("сбой после автоподтверждения") return await original(*args, **kwargs) monkeypatch.setattr(consolidation_documents, "on_documents_mapped", flaky) assert await process_once(maker, storage, _settings()) is False # повторяемая неудача (R-15) assert await _documents(maker, file_id) == [] # маппинг вместе с автоподтверждением откатился целиком assert await process_once(maker, storage, _settings()) is True [document] = await _documents(maker, file_id) assert document.status is DocumentStatus.CONFIRMED async with maker() as session: count = ( await session.execute( select(func.count()).select_from(AuditEvent).where(AuditEvent.action == AuditAction.DOCUMENT_CONFIRM) ) ).scalar_one() assert count == 1 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)) ---------------------------------------- async def _audit_count(maker: async_sessionmaker[AsyncSession]) -> int: async with maker() as session: return (await session.execute(select(func.count()).select_from(AuditEvent))).scalar_one() async def test_exchange_and_mapping_write_no_audit_events(committed_sessionmaker: async_sessionmaker[AsyncSession]): """spec audit-trail «События файлов и документов»: смена статусов задания и файла (ack, результат, маппинг) и автозамена без заменяемых файлов `audit_event` не пишут: путь `UPLOADED → PROCESSING → PROCESSED` для документа `NEEDS_CHECK` (у владельца нет ИНН) оставляет таблицу аудита неизменной. Автоподтверждение чистого документа пишет `DOCUMENT_CONFIRM` — это проверяет отдельный тест.""" maker, storage = committed_sessionmaker, FakeStorage() file_id, job_id, _ = await _prepare( maker, storage, _minimal(), job_status=RecognitionStatus.published, file_status=FileStatus.UPLOADED ) before = await _audit_count(maker) await handle_message( fake_msg(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1)), session_factory=maker ) assert (await _file(maker, file_id)).status is FileStatus.PROCESSING await handle_message(fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)), session_factory=maker) assert (await _job(maker, job_id)).status is RecognitionStatus.done assert await process_once(maker, storage, _settings()) is True assert (await _file(maker, file_id)).status is FileStatus.PROCESSED [document] = await _documents(maker, file_id) assert document.status is DocumentStatus.NEEDS_CHECK assert await _audit_count(maker) == before async def test_watchdog_timeout_writes_no_audit_event(committed_sessionmaker: async_sessionmaker[AsyncSession]): """Путь ошибки: вахтёр переводит задание в `timeout`, файл — в `ERROR`, `audit_event` не пишется.""" maker, storage = committed_sessionmaker, FakeStorage() file_id, job_id, _ = await _prepare(maker, storage, _minimal(), job_status=RecognitionStatus.processing) async with maker() as session, session.begin(): job = await session.get(RecognitionJob, job_id) assert job is not None job.reply_deadline = _now() - timedelta(minutes=1) before = await _audit_count(maker) assert await expire_deadlines(maker, now=_now) >= 1 assert (await _file(maker, file_id)).status is FileStatus.ERROR assert (await _job(maker, job_id)).status is RecognitionStatus.timeout assert await _audit_count(maker) == before