1abe212524
I-1: лист с 1–2 кодами после листа формы — продолжение (min_codes_continuation в словаре). M-10: тип формы маркера продолжения запрещает присоединение к документу иного типа. I-2: одиночный месяц не задаёт границы при дате на число/квартале (not_single_month). I-3: название строки из ближайшей текстовой ячейки левее кодов, пустое имя без confidence. M-5: units_thousands и okud_code в словаре, без type: ignore в detect.py. M-8: строка нумерации и нули не считаются строкой без кода с суммами. M-11: предупреждение о дробных значениях — одно на документ. T13.5: _upper_header_row возвращает верхнюю строку при склейке хотя бы одной колонки. M-6/M-7/T15.1/T15.2: остатки statement_markers, описание fractional_share_percent, README, тест воркера. Refs #14 Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EAKRDNzXtRrwmWz9z6PZu3
259 lines
12 KiB
Python
259 lines
12 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 src.parser.statements.statement import SKIPPED_COLUMN_WARNING
|
||
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 warnings_of(outcome: ParseOutcome) -> list[str]:
|
||
assert outcome.metrics is not None
|
||
assert outcome.metrics.warnings is not None
|
||
return outcome.metrics.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, column_index: int = 0) -> Decimal | None:
|
||
"""Получить значение из строки по индексу колонки."""
|
||
for value in line.values:
|
||
if value.column_index == column_index:
|
||
return value.amount
|
||
return None
|
||
|
||
|
||
# --- Состав набора и общие ожидания ---
|
||
|
||
|
||
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 warnings_of(outcome) if SKIPPED_COLUMN_WARNING in w]
|
||
assert skipped == [f"sheet=Лист1 {SKIPPED_COLUMN_WARNING}: Изменение, %"]
|
||
|
||
# Проверяем код 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.relations_failed == []
|
||
|
||
# Проверяем warning о единицах
|
||
warnings = warnings_of(outcome)
|
||
assert any("unit" in w.lower() or "измер" in w.lower() for w in warnings), (
|
||
f"ожидается warning о единицах, получены: {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
|
||
|
||
warnings = warnings_of(outcome)
|
||
assert any("code" in w.lower() or "код" in w.lower() for w in warnings), (
|
||
f"ожидается warning о коде, получены: {warnings}"
|
||
)
|
||
|
||
|
||
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; needs_check, warning о continuation."""
|
||
outcome = parse_named(fixtures_dir, "st05_continuation_only.xlsx")
|
||
doc = one_statement(outcome)
|
||
|
||
assert outcome.status is ResultStatus.NEEDS_CHECK
|
||
|
||
warnings = warnings_of(outcome)
|
||
assert any(
|
||
"continuation" in w.lower() or "страниц" in w.lower() or "страница" in w.lower() or "page" in w.lower()
|
||
for w in warnings
|
||
), f"ожидается warning о continuation, получены: {warnings}"
|
||
|
||
# Проверяем что есть коды
|
||
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)
|