71f596cdb7
- копия схем и примеров контракта v2.0 (запрос остаётся v1.0), модели result/common/message/request; ResultWarning/FileWarning, статус и предупреждения у документов, строки forms v2.0, service.statements_version - statements/pipeline.py: кандидаты адаптера -> normalize_candidate -> to_contract / warning_to_contract - BookParser: статус книги через envelope_status, SHEET_OTHER / FILE_TYPE_MISMATCH / XLS_NO_CODEPAGE как предупреждения файла, OSV_* и OSV_XLSB_NO_OUTLINE в документах ОСВ, тексты иерархии по спеке - воркер, app и CLI получают statements_version из require_statements_version() - старые модели документа формы вынесены в statements/legacy_models.py до удаления старого конвейера (задача 9) - OpenSpec: MODIFIED «Парсер — заменяемый интерфейс» в дельте tabular-recognition-service Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W
248 lines
11 KiB
Python
248 lines
11 KiB
Python
"""Сценарии спеки `statement-parsing` на синтетических файлах (Срез 2b, Task 13).
|
||
|
||
Ожидания выведены из сценариев спеки и из содержимого генератора, а не из текущего вывода парсера:
|
||
файл — вход, оракулом он не служит. Без каталога fixtures тесты пропускаются (маркер `fixtures`).
|
||
"""
|
||
|
||
from datetime import date
|
||
from decimal import Decimal
|
||
from pathlib import Path
|
||
from typing import cast
|
||
|
||
import pytest
|
||
from result_contract import assert_valid_result
|
||
|
||
from src.contract import FileType, OsvDocument, ResultStatus, StatementDocument
|
||
from src.contract.result import StatementLine
|
||
from src.parser import ParseOutcome, get_parser
|
||
from tests.fixtures_helpers import statement_synthetic_skip_reason
|
||
from tests.synthetic.generate_statements import FILE_NAMES
|
||
|
||
pytestmark = pytest.mark.fixtures
|
||
|
||
SYNTHETIC_DIR = Path("statements-xlsx") / "synthetic"
|
||
EXPECTED_STATEMENT_COUNT_TWO = 2
|
||
EXPECTED_COLUMN_COUNT = 2
|
||
EXPECTED_LINES_COUNT = 0
|
||
EXPECTED_DISCLOSURES = 5
|
||
EXPECTED_INCOME_LINES = 10
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _require_synthetic() -> None:
|
||
"""T13.4: fixtures есть, а `statements-xlsx/synthetic` нет/пуст — skip, а не падение."""
|
||
reason = statement_synthetic_skip_reason()
|
||
if reason is not None:
|
||
pytest.skip(reason)
|
||
|
||
|
||
def synthetic_path(fixtures_dir: Path, name: str) -> Path:
|
||
return fixtures_dir / SYNTHETIC_DIR / name
|
||
|
||
|
||
def parse_path(path: Path) -> ParseOutcome:
|
||
return get_parser().parse(path.read_bytes(), cast(FileType, path.suffix.lstrip(".").upper()))
|
||
|
||
|
||
def parse_named(fixtures_dir: Path, name: str) -> ParseOutcome:
|
||
path = synthetic_path(fixtures_dir, name)
|
||
if not path.is_file():
|
||
pytest.fail(f"нет файла {name}: запустите генератор (см. README fixtures)")
|
||
return parse_path(path)
|
||
|
||
|
||
def statement_documents(outcome: ParseOutcome) -> list[StatementDocument]:
|
||
"""Документы форм результата."""
|
||
assert outcome.error is None, outcome.error
|
||
return [doc for doc in outcome.documents if isinstance(doc, StatementDocument)]
|
||
|
||
|
||
def osv_documents(outcome: ParseOutcome) -> list[OsvDocument]:
|
||
"""Документы ОСВ результата."""
|
||
return [doc for doc in outcome.documents if isinstance(doc, OsvDocument)]
|
||
|
||
|
||
def one_statement(outcome: ParseOutcome) -> StatementDocument:
|
||
(doc,) = statement_documents(outcome)
|
||
return doc
|
||
|
||
|
||
def codes_of(doc: StatementDocument) -> dict[str, str]:
|
||
"""Предупреждения документа: код → уровень (контракт 2.0: лежат в самом документе)."""
|
||
return {w.code: w.severity for w in doc.warnings}
|
||
|
||
|
||
def by_code(doc: StatementDocument) -> dict[str, StatementLine]:
|
||
"""Строки по коду показателя."""
|
||
lines = {line.line_code: line for line in doc.lines}
|
||
return lines
|
||
|
||
|
||
def get_line_value(line: StatementLine, position: int = 0) -> Decimal | None:
|
||
"""Значение строки в колонке с порядковым номером `position` (`column_index` — столбец сетки, не порядок)."""
|
||
return line.values[position].amount
|
||
|
||
|
||
# --- Состав набора и общие ожидания ---
|
||
|
||
|
||
def test_synthetic_set_is_complete(fixtures_dir: Path) -> None:
|
||
"""В `fixtures/statements-xlsx/synthetic/` лежат все файлы st01…st05."""
|
||
present = sorted(path.name for path in (fixtures_dir / SYNTHETIC_DIR).glob("st*") if path.is_file())
|
||
assert present == list(FILE_NAMES)
|
||
|
||
|
||
def test_every_synthetic_file_is_listed(statement_synthetic_file: Path) -> None:
|
||
"""Каждый файл каталога соответствует одному из ожидаемых."""
|
||
assert statement_synthetic_file.name in FILE_NAMES
|
||
|
||
|
||
# --- Сценарии спеки ---
|
||
|
||
|
||
def test_st01_both_forms_one_sheet(fixtures_dir: Path) -> None:
|
||
"""st01: ББ и ОФР на одном листе; два документа, один locator, разделение кодов по типам."""
|
||
outcome = parse_named(fixtures_dir, "st01_both_forms_one_sheet.xlsx")
|
||
stmt_docs = statement_documents(outcome)
|
||
assert len(stmt_docs) == EXPECTED_STATEMENT_COUNT_TWO
|
||
|
||
# Оба документа имеют один locator
|
||
locators = {doc.locator for doc in stmt_docs}
|
||
assert len(locators) == 1
|
||
|
||
# Проверяем разделение кодов: 1xxx в BALANCE, 2xxx в INCOME
|
||
balance_codes = set()
|
||
income_codes = set()
|
||
for doc in stmt_docs:
|
||
codes = {line.line_code for line in doc.lines if line.line_code}
|
||
if doc.document_type == "BALANCE":
|
||
balance_codes.update(codes)
|
||
elif doc.document_type == "INCOME":
|
||
income_codes.update(codes)
|
||
|
||
# Проверяем что коды не пересекаются
|
||
assert not (balance_codes & income_codes), f"Пересечение кодов: {balance_codes & income_codes}"
|
||
# Проверяем что есть коды обоих типов
|
||
assert any(code.startswith("1") for code in balance_codes), "Должны быть коды 1xxx в BALANCE"
|
||
assert any(code.startswith("2") for code in income_codes), "Должны быть коды 2xxx в INCOME"
|
||
|
||
# Статус done
|
||
assert outcome.status is ResultStatus.DONE
|
||
|
||
|
||
def test_st02_csv_like_format(fixtures_dir: Path) -> None:
|
||
"""st02: CSV-подобный лист; организация/ИНН найдены, units None, колонки [2024-12-31, 2023-12-31], 1210=-50."""
|
||
outcome = parse_named(fixtures_dir, "st02_csv_like.xlsx")
|
||
doc = one_statement(outcome)
|
||
|
||
# Проверяем организацию и ИНН
|
||
assert doc.header.organization_name == 'ООО "Синтетика"'
|
||
assert doc.header.inn == "7700000001"
|
||
|
||
# Проверяем units = None
|
||
assert doc.header.units is None
|
||
|
||
# Проверяем колонки: две даты
|
||
assert len(doc.columns) == EXPECTED_COLUMN_COUNT
|
||
assert all(col.period_date is not None for col in doc.columns)
|
||
assert [col.period_date for col in doc.columns] == [date(2024, 12, 31), date(2023, 12, 31)]
|
||
# T13.I2: «Изменение, %» не колонка, её значения (5) не читаются, предупреждение — ровно одно
|
||
assert all("Изменение" not in col.label for col in doc.columns)
|
||
assert all(len(line.values) == EXPECTED_COLUMN_COUNT for line in doc.lines)
|
||
skipped = [w for w in doc.warnings if w.code == "COLUMN_SKIPPED"]
|
||
assert [(w.severity, "Изменение, %" in w.message) for w in skipped] == [("info", True)]
|
||
|
||
# Проверяем код 1210 = -50
|
||
lines_by_code = by_code(doc)
|
||
assert get_line_value(lines_by_code["1210"], 0) == Decimal("-50")
|
||
assert get_line_value(lines_by_code["1210"], 1) == Decimal("-40")
|
||
|
||
# T13.I1: данные согласованы со всеми жёсткими соотношениями — статус done, нарушений нет
|
||
assert outcome.status is ResultStatus.DONE
|
||
assert doc.checks is not None and doc.checks.relations_failed == []
|
||
|
||
# Проверяем warning о единицах
|
||
assert codes_of(doc).get("UNITS_NOT_FOUND") == "info", doc.warnings
|
||
|
||
|
||
def test_st03_no_code_column(fixtures_dir: Path) -> None:
|
||
"""st03: форма без колонки кодов; needs_check, warning о коде, lines = []."""
|
||
outcome = parse_named(fixtures_dir, "st03_no_codes.xlsx")
|
||
doc = one_statement(outcome)
|
||
|
||
assert outcome.status is ResultStatus.NEEDS_CHECK
|
||
assert len(doc.lines) == EXPECTED_LINES_COUNT
|
||
|
||
assert codes_of(doc).get("CODE_COLUMN_NOT_FOUND") == "error", doc.warnings
|
||
assert doc.status == "needs_check"
|
||
assert doc.checks is None
|
||
|
||
|
||
def test_st04_mixed_book_osv_and_statement(fixtures_dir: Path) -> None:
|
||
"""st04: книга с листом ОСВ и листом баланса; OsvDocument + StatementDocument, порядок правильный."""
|
||
outcome = parse_named(fixtures_dir, "st04_mixed_book.xlsx")
|
||
|
||
# Проверяем количество документов
|
||
assert len(outcome.documents) == EXPECTED_STATEMENT_COUNT_TWO
|
||
|
||
# Проверяем типы и порядок
|
||
assert isinstance(outcome.documents[0], OsvDocument), "Первый документ должен быть OsvDocument"
|
||
assert isinstance(outcome.documents[1], StatementDocument), "Второй документ должен быть StatementDocument"
|
||
assert outcome.documents[1].document_type == "BALANCE"
|
||
|
||
# Статус done
|
||
assert outcome.status is ResultStatus.DONE
|
||
|
||
|
||
def test_st05_continuation_page(fixtures_dir: Path) -> None:
|
||
"""st05: одинокая страница 2: пустая шапка (`NO_ORGANIZATION_NO_INN`, error) → документ и книга `needs_check`."""
|
||
outcome = parse_named(fixtures_dir, "st05_continuation_only.xlsx")
|
||
doc = one_statement(outcome)
|
||
|
||
assert outcome.status is ResultStatus.NEEDS_CHECK
|
||
|
||
assert doc.status == "needs_check"
|
||
assert codes_of(doc).get("NO_ORGANIZATION_NO_INN") == "error", doc.warnings
|
||
assert doc.header.organization_name is None and doc.header.inn is None
|
||
|
||
# Проверяем что есть коды
|
||
codes = {line.line_code for line in doc.lines if line.line_code}
|
||
assert len(codes) > 0, "должны быть коды"
|
||
|
||
|
||
def test_st06_header_variants_documents_and_header(fixtures_dir: Path) -> None:
|
||
"""st06: старое название ОФР, четыре вида расшифровок и форма только с ОКУД 0710xxx; шапка с «», «Адрес»."""
|
||
outcome = parse_named(fixtures_dir, "st06_header_variants.xlsx")
|
||
documents = statement_documents(outcome)
|
||
|
||
assert [doc.document_type for doc in documents] == ["INCOME", *["DISCLOSURE"] * EXPECTED_DISCLOSURES]
|
||
income = documents[0]
|
||
assert income.header.organization_name == "«Синтетика-Групп»"
|
||
assert income.header.address == "г. Москва, ул. Синтетическая, д. 1"
|
||
assert len(income.lines) == EXPECTED_INCOME_LINES
|
||
assert [column.period_date for column in income.columns] == [date(2024, 12, 31), date(2023, 12, 31)]
|
||
assert all(doc.lines == [] for doc in documents[1:])
|
||
assert income.checks is not None
|
||
assert income.checks.relations_failed == []
|
||
|
||
|
||
def test_st07_single_code_continuation_joins_the_income_form(fixtures_dir: Path) -> None:
|
||
"""st07 (O-9): лист с единственной кодовой строкой 2400 присоединяется к ОФР; без него 2400 терялась бы."""
|
||
outcome = parse_named(fixtures_dir, "st07_single_code_continuation.xlsx")
|
||
document = one_statement(outcome)
|
||
|
||
assert document.document_type == "INCOME"
|
||
assert document.locator == "sheet:ОФР,ОФР_2"
|
||
assert [line.line_code for line in document.lines][-1] == "2400"
|
||
assert document.checks is not None and document.checks.relations_failed == []
|
||
assert outcome.status is ResultStatus.DONE
|
||
|
||
|
||
# --- Валидация контракта ---
|
||
|
||
|
||
def test_all_synthetic_results_are_valid(statement_synthetic_file: Path) -> None:
|
||
"""Все результаты валидны по контракту."""
|
||
outcome = parse_path(statement_synthetic_file)
|
||
assert_valid_result(outcome)
|