1d821b4d3e
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
205 lines
11 KiB
Python
205 lines
11 KiB
Python
"""Маппинг форм ББ/ОФР (spec osv-result-mapping «Цикл маппинга…», «Документ формы ББ/ОФР», «Повтор формы в файле и
|
||
прочие документы», «Статус документа формы»; design D-1…D-3, D-6) на golden-снимках fixtures."""
|
||
|
||
import json
|
||
import uuid
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from sqlalchemy import func, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from src.models.directories import LegalEntity
|
||
from src.models.enums import DocumentCheckResult, DocumentStatus, DocumentType, FileStatus, PeriodQuarter
|
||
from src.models.files import Document, StoredFile
|
||
from src.models.statements import (
|
||
FinancialStatementColumn,
|
||
FinancialStatementLine,
|
||
FinancialStatementMeta,
|
||
FinancialStatementValue,
|
||
)
|
||
from tests.factories import make_legal_entity
|
||
from tests.fixtures_helpers import GOLDEN_STATEMENTS, OCR_EXPECTED, 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
|
||
|
||
|
||
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 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_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_skipped(seeded: AsyncSession, fixtures_dir: Path, tmp_path: Path) -> None:
|
||
"""R-1, spec «Повтор формы в файле»: вторая форма того же типа не создаётся, первая — `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]
|
||
assert documents[0].status is DocumentStatus.NEEDS_CHECK # без повтора тот же снимок — RECOGNIZED
|