arm-tabular-parser/api/tests/synthetic/test_statement_synthetic_files.py

"""Сценарии спеки `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


@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)
    if "1210" in lines_by_code:
        line = lines_by_code["1210"]
        col_0_value = get_line_value(line, 0)
        assert col_0_value == Decimal("-50"), f"1210 в колонке 0 должна быть -50, получено {col_0_value}"

    # Проверяем warning о единицах
    warnings = warnings_of(outcome)
    assert any("unit" in w.lower() or "измер" in w.lower() for w in warnings), (
        f"ожидается warning о единицах, получены: {warnings}"
    )

    # Примечание: warning о "Изменение, %" может быть не выдан из-за особенности двухстрочного заголовка (см. отчёт)


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}"
    )


EXPECTED_LINES_COUNT = 0


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_all_synthetic_results_are_valid(statement_synthetic_file: Path) -> None:
    """Все результаты валидны по контракту."""
    outcome = parse_path(statement_synthetic_file)
    assert_valid_result(outcome)