RSHB/arm-core
9
0
Fork 0
Code Issues 25 Pull Requests 1 Actions Packages Projects Releases Wiki Activity
arm-core/api/tests/test_result_mapping.py
Andrey Epifancev 2b8fafc3b3 feat(mapping): ОСВ по контракту 2.0 (Р-28) и выбор маппера по MAJOR 
Приём файла результата выбирает модель по MAJOR (parse_result); нечитаемый JSON и неизвестный
MAJOR — неповторяемая ошибка схемы (T5.2). Статус ОСВ 2.0 — osv_status_v2, предупреждения листа
(OSV_*) — в document_issue. Решения P3-46…P3-49.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-18 21:57:38 +04:00

2037 lines
100 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Цикл маппинга результата распознавания в `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/v1/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.models.statements import (
DocumentIssue,
FinancialStatementLine,
FinancialStatementMeta,
FinancialStatementValue,
)
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" / "v1" / "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" / "v1" / "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
# --- v5.19: поля 1.x, file_ordinal, вторая форма, MAPPING_WARNING (план 3, A9) ---------------------------------------
def _osv_sheet(account: str, locator: str) -> dict[str, Any]:
sheet = copy.deepcopy(_osv(_minimal()))
sheet["header"]["account_code_raw"] = account
sheet["locator"] = locator
return sheet
def _form(document_type: str, locator: str, *, report_date: str = "2025-09-30") -> dict[str, Any]:
form = copy.deepcopy(json.loads(STATEMENT_MINIMAL.read_text("utf-8"))["documents"][0])
form["document_type"] = document_type
form["locator"] = locator
form["header"]["report_date"] = report_date
return form
async def _map_dict(
session: AsyncSession, documents: list[dict[str, Any]], *, owner_inn: str = "7700000000"
) -> tuple[StoredFile, list[Document]]:
"""`map_file` на результате 1.x из словарей `documents`; документы файла по `file_ordinal`."""
entity = await make_legal_entity(session, name=OWNER_NAME, inn=owner_inn)
file, job = await make_stored_file(
session, entity=entity, status=FileStatus.PROCESSING, job_status=RecognitionStatus.done
)
payload = _minimal() | {"job_id": str(job.id), "file_id": str(file.id), "documents": documents}
result = RecognitionResult.model_validate(payload)
chart = {code: key for code, key in await session.execute(select(ChartOfAccount.code, ChartOfAccount.id))}
await rm.map_file(
session, job=job, file=file, result=result, chart=chart, index=await CounterpartyIndex.load(session), now=_now()
)
stored = (
(
await session.execute(
select(Document).where(Document.stored_file_id == file.id).order_by(Document.file_ordinal)
)
)
.scalars()
.all()
)
return file, list(stored)
async def _issues(session: AsyncSession, document: Document) -> list[DocumentIssue]:
stmt = select(DocumentIssue).where(DocumentIssue.document_id == document.id)
return list((await session.execute(stmt)).scalars())
async def test_two_osv_sheets_file_ordinals(seeded: AsyncSession) -> None:
file, documents = await _map_dict(seeded, [_osv_sheet("60", "sheet:60"), _osv_sheet("62", "sheet:62")])
assert [(d.osv_account_code, d.file_ordinal) for d in documents] == [("60", 0), ("62", 1)]
await seeded.refresh(file, ["status"])
assert file.status is FileStatus.PROCESSED
async def test_pdf_balance_income_1x_ordinals(seeded: AsyncSession) -> None:
_, documents = await _map_dict(seeded, [_form("BALANCE", "pages:1"), _form("INCOME", "pages:2")])
assert [(d.document_type, d.file_ordinal) for d in documents] == [
(DocumentType.BALANCE, 0),
(DocumentType.INCOME, 1),
]
async def test_duplicate_osv_sheet_takes_no_ordinal(seeded: AsyncSession) -> None:
_, documents = await _map_dict(
seeded, [_osv_sheet("60", "sheet:a"), _osv_sheet("60", "sheet:b"), _form("BALANCE", "sheet:ББ")]
)
assert [(d.document_type, d.file_ordinal) for d in documents] == [(DocumentType.OSV, 0), (DocumentType.BALANCE, 2)]
async def test_disclosure_ordinal_is_first_in_documents(seeded: AsyncSession) -> None:
"""P3-6: объединённый `DISCLOSURE` — индекс первого такого документа в `documents[]`."""
bare = {"header": {}, "columns": [], "lines": []}
disclosure = {"document_type": "DISCLOSURE", "locator": "pages:3", **bare}
other = {"document_type": "OTHER", "locator": "pages:9", **bare}
_, documents = await _map_dict(
seeded, [_form("BALANCE", "pages:1"), disclosure, other, disclosure | {"locator": "pages:4"}]
)
assert {(d.document_type, d.file_ordinal) for d in documents} == {
(DocumentType.BALANCE, 0),
(DocumentType.DISCLOSURE, 1),
(DocumentType.OTHER, 2),
}
async def test_1x_line_fields(seeded: AsyncSession) -> None:
form = _form("BALANCE", "sheet:ББ")
form["lines"] = [
{
"line_code": code,
"line_name": f"строка {code}",
"values": [
{
"column_index": 0,
"amount": "10",
"string_value": "10,0",
"confidence": confidence,
"is_recognized": True,
}
],
}
for code, confidence in (("1100", 100), ("1200", 80), ("1600", 100))
]
_, [balance] = await _map_dict(seeded, [form])
meta = (
await seeded.execute(select(FinancialStatementMeta).where(FinancialStatementMeta.document_id == balance.id))
).scalar_one()
lines = (
(
await seeded.execute(
select(FinancialStatementLine)
.where(FinancialStatementLine.meta_id == meta.id)
.order_by(FinancialStatementLine.row_index)
)
)
.scalars()
.all()
)
assert [line.row_index for line in lines] == [0, 1, 2]
assert all(line.line_status.value == "ok" and line.source_ref is None for line in lines)
assert all(line.line_code_original_value == line.line_code and line.needs_review is False for line in lines)
values = {
line.line_code: (
await seeded.execute(select(FinancialStatementValue).where(FinancialStatementValue.line_id == line.id))
)
.scalars()
.all()
for line in lines
}
flags = {code: {v.needs_review for v in cells if v.confidence is not None} for code, cells in values.items()}
assert flags == {"1100": {False}, "1200": {True}, "1600": {False}}
assert {v.source_text for v in values["1100"] if v.string_value} == {"10,0"}
async def test_1x_duplicate_code_keeps_first(seeded: AsyncSession) -> None:
form = _form("INCOME", "sheet:ОФР")
form["lines"] = [
{"line_code": "2110", "line_name": "а", "values": []},
{"line_code": "2410", "line_name": "б", "values": []},
{"line_code": "2410", "line_name": "повтор", "values": []},
]
_, [income] = await _map_dict(seeded, [form])
meta = (
await seeded.execute(select(FinancialStatementMeta).where(FinancialStatementMeta.document_id == income.id))
).scalar_one()
rows = (
await seeded.execute(
select(FinancialStatementLine.row_index, FinancialStatementLine.line_code)
.where(FinancialStatementLine.meta_id == meta.id)
.order_by(FinancialStatementLine.row_index)
)
).all()
assert [tuple(row) for row in rows] == [(0, "2110"), (1, "2410")]
messages = [issue.message for issue in await _issues(seeded, income) if issue.code == "MAPPING_WARNING"]
assert any("2410" in message and "повторяется" in message for message in messages)
async def test_1x_second_form_same_period_stored(seeded: AsyncSession) -> None:
"""Review Focus 1, P3-4 (Р-10/Р-18): две формы одного типа и периода хранятся, обе `NEEDS_CHECK`."""
file, documents = await _map_dict(seeded, [_form("BALANCE", "sheet:а"), _form("BALANCE", "sheet:б")])
assert [d.file_ordinal for d in documents] == [0, 1]
assert all(d.status is DocumentStatus.NEEDS_CHECK for d in documents)
for document in documents:
messages = [i.message for i in await _issues(seeded, document) if i.code == "MAPPING_WARNING"]
assert "в файле есть другая форма BALANCE за тот же период" in messages
issue = next(i for i in await _issues(seeded, document) if i.code == "MAPPING_WARNING")
assert (issue.kind.value, issue.severity.value, issue.source.value) == ("warning", "info", "core")
async def test_1x_second_form_other_period(seeded: AsyncSession) -> None:
_, documents = await _map_dict(
seeded, [_form("BALANCE", "sheet:а"), _form("BALANCE", "sheet:б", report_date="2024-12-31")]
)
assert [(d.period_year, d.period_quarter) for d in documents] == [
(2025, PeriodQuarter.Q3),
(2024, PeriodQuarter.Q4),
]
for document in documents:
assert not any("другая форма" in i.message for i in await _issues(seeded, document))
async def test_osv_orphans_mapping_warning_issue(seeded: AsyncSession) -> None:
sheet = _osv_sheet("60", "sheet:60")
sheet["rows"] = [*sheet["rows"], sheet["rows"][1] | {"row_number": 40, "parent_row_number": 99}]
file, [document] = await _map_dict(seeded, [sheet])
issues = await _issues(seeded, document)
[orphans] = [i for i in issues if i.code == "MAPPING_WARNING"]
assert "без родителя" in orphans.message and "40" in orphans.message
assert (orphans.kind.value, orphans.severity.value, orphans.source.value) == ("warning", "info", "core")
job = await seeded.get(RecognitionJob, file.current_recognition_job_id)
assert job is not None and any("без родителя" in w for w in job.metrics[MAPPING_WARNINGS_KEY])
# --- приём файла результата: выбор по MAJOR, нечитаемый JSON (B5) ---------------------------------------------------
def _result_key(file_id: uuid.UUID, job_id: uuid.UUID) -> str:
return f"results/{file_id}/{job_id}/result.json"
async def test_result_major_3_is_fatal(committed_sessionmaker: async_sessionmaker[AsyncSession]) -> None:
"""P3-46: неизвестный MAJOR (`3.0`) — неповторяемая ошибка схемы: файл `ERROR`, без документов и повтора."""
maker, storage = committed_sessionmaker, FakeStorage()
file_id, job_id, _ = await _prepare(maker, storage, _minimal() | {"schema_version": "3.0"})
assert await process_once(maker, storage, _settings()) is True
assert (await _file(maker, file_id)).status is FileStatus.ERROR
job = await _job(maker, job_id)
assert job.metrics[MAPPING_ERROR_KEY].startswith("результат не соответствует схеме")
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_result_1x_still_mapped(committed_sessionmaker: async_sessionmaker[AsyncSession]) -> None:
"""MAJOR 1 (`1.2`) — маппер `1.x`: документ ОСВ создан, файл `PROCESSED`."""
maker, storage = committed_sessionmaker, FakeStorage()
file_id, _, _ = await _prepare(maker, storage, _minimal() | {"schema_version": "1.2"}, owner_inn="7700000000")
assert await process_once(maker, storage, _settings()) is True
assert (await _file(maker, file_id)).status is FileStatus.PROCESSED
assert len(await _documents(maker, file_id)) == 1
@pytest.mark.parametrize("content", [b"{not json", b"", b"\xff\xfe\x00", b"[1, 2]", b"null"])
async def test_unreadable_result_json_is_fatal(
committed_sessionmaker: async_sessionmaker[AsyncSession], content: bytes
) -> None:
"""T5.2: невалидный JSON / не объект — `MappingFatal` («результат не соответствует схеме…»): файл `ERROR`,
без повтора, без текста исключения парсера в `mapping_error`."""
maker, storage = committed_sessionmaker, FakeStorage()
file_id, job_id, _ = await _prepare(maker, storage, _minimal())
storage.objects[_result_key(file_id, job_id)] = content
assert await process_once(maker, storage, _settings()) is True
assert (await _file(maker, file_id)).status is FileStatus.ERROR
job = await _job(maker, job_id)
assert job.metrics[MAPPING_ERROR_KEY].startswith("результат не соответствует схеме recognition-result")
assert "Expecting" not in job.metrics[MAPPING_ERROR_KEY]
assert MAPPING_ATTEMPTS_KEY not in job.metrics
assert await process_once(maker, storage, _settings()) is False
Reference in New Issue View Git Blame Copy Permalink