arm-core/api/tests/test_mapping_statements.py
Andrey Epifancev 02630e830f feat(mapping): §7 и вариант формы при приёме 1.x копией arm_statements (Р-30)
checks_on_intake_1x: один recompute по вставленной форме, запись form_variant и
relation/soft (core); run_checks из приёма убран. P3-36.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-18 21:03:06 +04:00

396 lines
21 KiB
Python
Raw Permalink 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.

"""Маппинг форм ББ/ОФР (spec osv-result-mapping «Цикл маппинга…», «Документ формы ББ/ОФР», «Повтор формы в файле и
прочие документы», «Статус документа формы»; design D-1…D-3, D-6) на golden-снимках fixtures."""
import json
import uuid
from datetime import UTC, datetime
from decimal import Decimal
from pathlib import Path
import pytest
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from src.models.accounting import ChartOfAccount
from src.models.directories import LegalEntity
from src.models.enums import (
DocumentCheckResult,
DocumentIssueKind,
DocumentStatus,
DocumentType,
FileStatus,
IssueSource,
PeriodQuarter,
RecognitionStatus,
StatementFormVariant,
)
from src.models.files import Document, RecognitionJob, StoredFile
from src.models.statements import (
DocumentIssue,
FinancialStatementColumn,
FinancialStatementLine,
FinancialStatementMeta,
FinancialStatementValue,
)
from src.repositories.counterparties import CounterpartyIndex
from src.services.result_mapping import MAPPING_WARNINGS_KEY, map_file
from tests.factories import make_document_issue, make_legal_entity
from tests.fixtures_helpers import GOLDEN_STATEMENTS, OCR_EXPECTED, golden_result, map_golden_snapshot, ocr_result
pytestmark = [pytest.mark.integration, pytest.mark.fixtures]
MEZH = GOLDEN_STATEMENTS / "mezheninovskaya" / "2025-Q3"
NPZ = GOLDEN_STATEMENTS / "npz" / "2025-Q3"
SYNTH = GOLDEN_STATEMENTS / "synthetic"
MEZH_BALANCE_LINES = 37
NPZ_BALANCE_LINES = 132
NPZ_INCOME_LINES = 43
DISCLOSURE_LOCATORS_COMMAS = 4 # пять `DISCLOSURE` снимка st06 → пять locator через запятую
MIN_EXCEL_SNAPSHOTS = 25
MIN_MAPPED_SNAPSHOTS = 27
async def _meta(session: AsyncSession, document: Document) -> FinancialStatementMeta:
stmt = select(FinancialStatementMeta).where(FinancialStatementMeta.document_id == document.id)
return (await session.execute(stmt)).scalar_one()
async def _count(session: AsyncSession, model: type, meta_id: uuid.UUID) -> int:
if model is FinancialStatementValue:
stmt = (
select(func.count())
.select_from(FinancialStatementValue)
.join(FinancialStatementLine, FinancialStatementLine.id == FinancialStatementValue.line_id)
.where(FinancialStatementLine.meta_id == meta_id)
)
else:
stmt = select(func.count()).select_from(model).where(model.meta_id == meta_id) # type: ignore[attr-defined]
return (await session.execute(stmt)).scalar_one()
async def _owner(session: AsyncSession, path: Path) -> LegalEntity:
"""Владелец файла с ИНН шапки формы (MATCH), чтобы чистые формы могли быть RECOGNIZED."""
data = json.loads(path.read_text("utf-8"))
inn = next((d["header"].get("inn") for d in data["documents"] if d.get("header", {}).get("inn")), None)
return await make_legal_entity(session, name="ООО Владелец", inn=inn)
async def test_file_with_forms_only(seeded: AsyncSession, fixtures_dir: Path) -> None:
path = fixtures_dir / MEZH / "1_Форма1_бух_09.json"
owner = await _owner(seeded, path)
documents = await map_golden_snapshot(seeded, entity=owner, path=path, uploaded_by=None)
assert [d.document_type for d in documents] == [DocumentType.BALANCE]
balance = documents[0]
assert (balance.period_year, balance.period_quarter) == (2025, PeriodQuarter.Q3)
assert balance.reporting_date is not None and balance.reporting_date.isoformat() == "2025-09-30"
assert balance.osv_account_code is None and balance.is_cumulative is None
meta = await _meta(seeded, balance)
lines = await _count(seeded, FinancialStatementLine, meta.id)
columns = await _count(seeded, FinancialStatementColumn, meta.id)
assert lines == MEZH_BALANCE_LINES
assert await _count(seeded, FinancialStatementValue, meta.id) == lines * columns
file = await seeded.get(StoredFile, balance.stored_file_id)
assert file is not None and file.status is FileStatus.PROCESSED
async def test_clean_income_recognized(seeded: AsyncSession, fixtures_dir: Path) -> None:
path = fixtures_dir / MEZH / "1_Форма2_бух_09.json"
owner = await _owner(seeded, path)
(income,) = await map_golden_snapshot(seeded, entity=owner, path=path, uploaded_by=None)
assert income.document_type is DocumentType.INCOME
assert income.counterparty_check is DocumentCheckResult.MATCH
assert income.status is DocumentStatus.RECOGNIZED
def _balance_with(tmp_path: Path, source: Path, edit) -> Path: # noqa: ANN001 - правка списка строк снимка
data = json.loads(source.read_text("utf-8"))
data["documents"][0]["lines"] = edit(data["documents"][0]["lines"])
path = tmp_path / "edited.json"
path.write_text(json.dumps(data, ensure_ascii=False), "utf-8")
return path
async def _core_relations(session: AsyncSession, document: Document) -> list[DocumentIssue]:
stmt = select(DocumentIssue).where(
DocumentIssue.document_id == document.id, DocumentIssue.kind == DocumentIssueKind.relation
)
return list((await session.execute(stmt)).scalars())
async def test_1x_balance_with_violation(seeded: AsyncSession, fixtures_dir: Path, tmp_path: Path) -> None:
"""Форма `1.x` с нарушением: `RELATION_FAILED` источника `core` при приёме, документ `NEEDS_CHECK`."""
source = fixtures_dir / MEZH / "1_Форма1_бух_09.json"
owner = await _owner(seeded, source)
def edit(lines: list[dict]) -> list[dict]:
for line in lines:
if line["line_code"] == "1600":
line["values"][0]["amount"] = str(Decimal(line["values"][0]["amount"] or 0) + 1000)
return lines
path = _balance_with(tmp_path, source, edit)
(balance,) = await map_golden_snapshot(seeded, entity=owner, path=path, uploaded_by=None)
assert balance.status is DocumentStatus.NEEDS_CHECK
failed = [i for i in await _core_relations(seeded, balance) if i.code == "RELATION_FAILED"]
assert failed and all(i.source is IssueSource.core for i in failed)
assert any(i.codes == ["1600", "1700"] for i in failed)
async def test_1x_simplified_variant(seeded: AsyncSession, fixtures_dir: Path, tmp_path: Path) -> None:
"""Вариант формы при приёме `1.x` (Р-30): ББ с 1600 и без 1100/1200 — `simplified`."""
source = fixtures_dir / MEZH / "1_Форма1_бух_09.json"
owner = await _owner(seeded, source)
path = _balance_with(tmp_path, source, lambda lines: [x for x in lines if x["line_code"] in {"1600", "1700"}])
(balance,) = await map_golden_snapshot(seeded, entity=owner, path=path, uploaded_by=None)
assert (await _meta(seeded, balance)).form_variant is StatementFormVariant.simplified
async def test_1x_clean_form_has_no_relation_issues(seeded: AsyncSession, fixtures_dir: Path) -> None:
path = fixtures_dir / MEZH / "1_Форма2_бух_09.json"
owner = await _owner(seeded, path)
(income,) = await map_golden_snapshot(seeded, entity=owner, path=path, uploaded_by=None)
assert income.status is DocumentStatus.RECOGNIZED and await _core_relations(seeded, income) == []
async def test_book_with_two_forms(seeded: AsyncSession, fixtures_dir: Path) -> None:
path = next((fixtures_dir / NPZ).glob("*.json"))
owner = await _owner(seeded, path)
documents = await map_golden_snapshot(seeded, entity=owner, path=path, uploaded_by=None)
by_type = {d.document_type: d for d in documents}
assert set(by_type) == {DocumentType.BALANCE, DocumentType.INCOME}
balance_meta = await _meta(seeded, by_type[DocumentType.BALANCE])
income_meta = await _meta(seeded, by_type[DocumentType.INCOME])
assert await _count(seeded, FinancialStatementLine, balance_meta.id) == NPZ_BALANCE_LINES
assert await _count(seeded, FinancialStatementLine, income_meta.id) == NPZ_INCOME_LINES
for document in documents:
assert (document.period_year, document.period_quarter) == (2025, PeriodQuarter.Q3)
async def test_disclosures_merged(seeded: AsyncSession, fixtures_dir: Path) -> None:
owner = await make_legal_entity(seeded, name="ООО Синтетика")
path = fixtures_dir / SYNTH / "st06_header_variants.json"
documents = await map_golden_snapshot(seeded, entity=owner, path=path, uploaded_by=None)
types = [d.document_type for d in documents]
assert types.count(DocumentType.DISCLOSURE) == 1 and DocumentType.INCOME in types
disclosure = next(d for d in documents if d.document_type is DocumentType.DISCLOSURE)
assert disclosure.status is DocumentStatus.RECOGNIZED
assert disclosure.locator.count(",") == DISCLOSURE_LOCATORS_COMMAS
async def test_form_without_codes(seeded: AsyncSession, fixtures_dir: Path) -> None:
owner = await make_legal_entity(seeded, name="ООО Без кодов")
path = fixtures_dir / SYNTH / "st03_no_codes.json"
(balance,) = await map_golden_snapshot(seeded, entity=owner, path=path, uploaded_by=None)
assert balance.status is DocumentStatus.NEEDS_CHECK
meta = await _meta(seeded, balance)
assert meta.inn == "7700000000" and meta.organization_name == 'ООО "Тестовая организация"'
assert await _count(seeded, FinancialStatementLine, meta.id) == 0
# в снимке st03_no_codes колонок нет (`columns = []`): только шапка (расхождение со spec «и колонками»)
assert await _count(seeded, FinancialStatementColumn, meta.id) == 0
async def test_mixed_book_osv_and_balance(seeded: AsyncSession, fixtures_dir: Path) -> None:
owner = await make_legal_entity(seeded, name="ООО Смешанная")
path = fixtures_dir / SYNTH / "st04_mixed_book.json"
documents = await map_golden_snapshot(seeded, entity=owner, path=path, uploaded_by=None)
assert {d.document_type for d in documents} == {DocumentType.OSV, DocumentType.BALANCE}
async def test_mixed_osv_snapshot_with_balance(seeded: AsyncSession, fixtures_dir: Path) -> None:
"""Снимок парсера ОСВ `s12` с листом ББ: лист ОСВ и по одной форме на лист ББ снимка."""
owner = await make_legal_entity(seeded, name="ООО Смешанная ОСВ")
path = fixtures_dir / "golden" / "expected_osv_parse" / "synthetic" / "s12_statement_and_osv.json"
raw = json.loads(path.read_text("utf-8"))
forms = [d for d in raw["documents"] if d["document_type"] in ("BALANCE", "INCOME")]
assert forms, "в снимке s12 больше нет формы: тест вырожден"
documents = await map_golden_snapshot(seeded, entity=owner, path=path, uploaded_by=None)
stored_forms = [d for d in documents if d.document_type in (DocumentType.BALANCE, DocumentType.INCOME)]
assert sorted(d.document_type for d in stored_forms) == sorted(f["document_type"] for f in forms)
assert [d.document_type for d in documents].count(DocumentType.OSV) == sum(
1 for d in raw["documents"] if d["document_type"] == "OSV"
)
for document in stored_forms:
assert (await _meta(seeded, document)).id is not None
async def test_ocr_form_without_report_date(seeded: AsyncSession, fixtures_dir: Path) -> None:
owner = await make_legal_entity(seeded, name="ООО Дорстроймонтаж")
documents = await map_golden_snapshot(
seeded,
entity=owner,
path=fixtures_dir / OCR_EXPECTED / "aep_6.json",
uploaded_by=None,
result_builder=ocr_result,
)
balance = next(d for d in documents if d.document_type is DocumentType.BALANCE)
assert (balance.period_year, balance.period_quarter) == (2025, PeriodQuarter.Q2)
meta = await _meta(seeded, balance)
assert meta.report_date is None and meta.inn is None
current = (
await seeded.execute(
select(FinancialStatementColumn.period_date).where(
FinancialStatementColumn.meta_id == meta.id, FinancialStatementColumn.is_current.is_(True)
)
)
).scalar_one()
assert current is not None and current.isoformat() == "2025-06-30"
assert balance.status is DocumentStatus.NEEDS_CHECK
async def test_ocr_noisy_form_mismatch(seeded: AsyncSession, fixtures_dir: Path) -> None:
owner = await make_legal_entity(seeded, name="ООО Другое", inn="7700000000")
documents = await map_golden_snapshot(
seeded,
entity=owner,
path=fixtures_dir / OCR_EXPECTED / "aep_3.json",
uploaded_by=None,
result_builder=ocr_result,
)
balance = next(d for d in documents if d.document_type is DocumentType.BALANCE)
assert balance.counterparty_check is DocumentCheckResult.MISMATCH
assert balance.status is DocumentStatus.NEEDS_CHECK
async def test_all_statement_snapshots_map(seeded: AsyncSession, fixtures_dir: Path) -> None:
"""Все golden форм (Excel и OCR, кроме ответов-ошибок) маппятся без исключений; у каждой формы значения на все
ячейки."""
owner = await make_legal_entity(seeded, name="ООО Все формы")
excel = sorted((fixtures_dir / GOLDEN_STATEMENTS).rglob("*.json"))
ocr = sorted((fixtures_dir / OCR_EXPECTED).glob("*.json"))
assert len(excel) >= MIN_EXCEL_SNAPSHOTS
mapped = 0
for path in excel:
await map_golden_snapshot(seeded, entity=owner, path=path, uploaded_by=None)
mapped += 1
for path in ocr:
if json.loads(path.read_text("utf-8"))["status"] == "error":
continue
await map_golden_snapshot(seeded, entity=owner, path=path, uploaded_by=None, result_builder=ocr_result)
mapped += 1
assert mapped >= MIN_MAPPED_SNAPSHOTS
metas = (await seeded.execute(select(FinancialStatementMeta.id))).scalars().all()
for meta_id in metas:
lines = await _count(seeded, FinancialStatementLine, meta_id)
columns = await _count(seeded, FinancialStatementColumn, meta_id)
assert await _count(seeded, FinancialStatementValue, meta_id) == lines * columns
async def test_duplicate_form_in_file_stored(seeded: AsyncSession, fixtures_dir: Path, tmp_path: Path) -> None:
"""P3-4 (Р-10, Р-18): вторая форма того же типа и периода хранится; обе `NEEDS_CHECK` (чистая — тоже)."""
source = fixtures_dir / MEZH / "1_Форма2_бух_09.json"
owner = await _owner(seeded, source)
data = json.loads(source.read_text("utf-8"))
data["documents"].append(data["documents"][0] | {"locator": "sheet:повтор"})
path = tmp_path / "duplicate.json"
path.write_text(json.dumps(data, ensure_ascii=False), "utf-8")
documents = await map_golden_snapshot(seeded, entity=owner, path=path, uploaded_by=None)
assert [d.document_type for d in documents] == [DocumentType.INCOME, DocumentType.INCOME]
assert all(d.status is DocumentStatus.NEEDS_CHECK for d in documents) # без повтора тот же снимок — RECOGNIZED
assert sorted(d.file_ordinal for d in documents) == [0, 1]
assert any("другая форма INCOME за тот же период" in w for w in await _mapping_warnings(seeded, documents[0]))
async def test_duplicate_column_index_does_not_fail_mapping(
seeded: AsyncSession, fixtures_dir: Path, tmp_path: Path
) -> None:
"""I-1: повтор `column_index` в результате не роняет маппинг (раньше — IntegrityError и файл `ERROR`): файл
`PROCESSED`, колонка одна, форма `NEEDS_CHECK` с предупреждением."""
source = fixtures_dir / MEZH / "1_Форма2_бух_09.json"
owner = await _owner(seeded, source)
data = json.loads(source.read_text("utf-8"))
columns = data["documents"][0]["columns"]
columns.insert(1, columns[0] | {"label": "дубль"})
path = tmp_path / "duplicate_column.json"
path.write_text(json.dumps(data, ensure_ascii=False), "utf-8")
(income,) = await map_golden_snapshot(seeded, entity=owner, path=path, uploaded_by=None)
assert income.status is DocumentStatus.NEEDS_CHECK # без повтора тот же снимок — RECOGNIZED
meta = await _meta(seeded, income)
stored = (
await seeded.execute(
select(FinancialStatementColumn.column_index, FinancialStatementColumn.label).where(
FinancialStatementColumn.meta_id == meta.id
)
)
).all()
first = columns[0]
assert sorted(i for i, _ in stored) == sorted({c["column_index"] for c in columns})
assert (first["column_index"], first["label"]) in stored and (first["column_index"], "дубль") not in stored
file = await seeded.get(StoredFile, income.stored_file_id)
assert file is not None and file.status is FileStatus.PROCESSED
assert any("повторяется — оставлена первая" in w for w in await _mapping_warnings(seeded, income))
async def test_form_with_needs_check_job_is_needs_check(seeded: AsyncSession, fixtures_dir: Path) -> None:
"""D-6: условие «задание done» — форма чистого снимка при задании `needs_check` не `RECOGNIZED`."""
path = fixtures_dir / MEZH / "1_Форма2_бух_09.json"
owner = await _owner(seeded, path)
(income,) = await map_golden_snapshot(
seeded, entity=owner, path=path, uploaded_by=None, job_status=RecognitionStatus.needs_check
)
assert income.counterparty_check is DocumentCheckResult.MATCH
assert (
income.status is DocumentStatus.NEEDS_CHECK
) # то же при задании done — RECOGNIZED (test_clean_income_recognized)
async def test_repeated_form_mapping_is_idempotent(seeded: AsyncSession, fixtures_dir: Path) -> None:
"""Spec «Повторный маппинг идемпотентен»: файл возвращён в `PROCESSING` и смаплен снова — один набор шапок, колонок,
строк и значений, никаких остатков; формы другого файла не тронуты."""
path = next((fixtures_dir / NPZ).glob("*.json"))
owner = await _owner(seeded, path)
bystander = await make_legal_entity(seeded, name="ООО Сосед")
other = await map_golden_snapshot(
seeded, entity=bystander, path=fixtures_dir / MEZH / "1_Форма1_бух_09.json", uploaded_by=None
)
documents = await map_golden_snapshot(seeded, entity=owner, path=path, uploaded_by=None)
before = await _statement_totals(seeded)
issues_before = (await seeded.execute(select(func.count()).select_from(DocumentIssue))).scalar_one()
file_id = documents[0].stored_file_id
await make_document_issue(seeded, document_id=documents[0].id) # v5.19: остаток прежнего маппинга
file = await seeded.get(StoredFile, file_id)
assert file is not None and file.current_recognition_job_id is not None
job = await seeded.get(RecognitionJob, file.current_recognition_job_id)
assert job is not None
file.status = FileStatus.PROCESSING
await seeded.flush()
chart = {code: key for code, key in await seeded.execute(select(ChartOfAccount.code, ChartOfAccount.id))}
result = golden_result(path, job_id=job.id, file_id=file.id)
await map_file(
seeded,
job=job,
file=file,
result=result,
chart=chart,
index=await CounterpartyIndex.load(seeded),
now=datetime.now(UTC),
)
assert await _statement_totals(seeded) == before
# записи прежнего маппинга (в том числе подложенная) удалены, новые §7 приёма — тот же набор (Р-30)
assert (await seeded.execute(select(func.count()).select_from(DocumentIssue))).scalar_one() == issues_before
again = (await seeded.execute(select(Document).where(Document.stored_file_id == file_id))).scalars().all()
assert {d.document_type for d in again} == {DocumentType.BALANCE, DocumentType.INCOME}
assert {d.id for d in again}.isdisjoint({d.id for d in documents}) # документы пересозданы, не дублированы
assert file.status is FileStatus.PROCESSED
kept = (
await seeded.execute(select(Document).where(Document.stored_file_id == other[0].stored_file_id))
).scalar_one()
assert kept.id == other[0].id
async def _statement_totals(session: AsyncSession) -> dict[str, int]:
"""Число строк во всех таблицах формы: любой остаток или дубль после повторного маппинга меняет итог."""
counts: dict[str, int] = {}
for model in (FinancialStatementMeta, FinancialStatementColumn, FinancialStatementLine, FinancialStatementValue):
counts[model.__tablename__] = (await session.execute(select(func.count()).select_from(model))).scalar_one()
return counts
async def _mapping_warnings(session: AsyncSession, document: Document) -> list[str]:
file = await session.get(StoredFile, document.stored_file_id)
assert file is not None and file.current_recognition_job_id is not None
job = await session.get(RecognitionJob, file.current_recognition_job_id)
assert job is not None
return list(job.metrics[MAPPING_WARNINGS_KEY])