arm-tabular-parser/api/tests/fixtures/test_osv_dkb.py

"""Property-тесты пакета ДКБ `../fixtures/osv-dkb/**` (issue #10): многолистовые книги, статусы `needs_check` как
ожидаемые данные, период каталога, файлы 9–14 МБ без golden.

Инварианты проверяются на каждом файле независимо от снимков. Ожидаемые нарушения контрольных сумм — манифест
`osv_dkb_expectations.yaml` (`dkb_helpers`), а не условия по имени файла: чего в манифесте нет, должно сойтись.
Период документа — по имени каталога (`2025-Q3`, `2024-10_2025-09`).
"""

import time
from pathlib import Path
from typing import cast

import pytest
from dkb_helpers import expected_period, expected_violations, load_expectations, violated_kinds
from fixtures_helpers import GOLDEN_MAX_BYTES, OSV_EXTENSIONS, osv_dkb_files, osv_dkb_golden_files
from golden_helpers import golden_path
from result_contract import assert_valid_result

from src.contract import FileType, OsvDocument, ResultStatus
from src.parser import ParseOutcome, get_parser

pytestmark = pytest.mark.fixtures

EXPECTED_FILES = 25  # пакет ДКБ 06.10.2026 без отчётов «Анализ счета»: 22 файла до порога снимка + 3 большие
EXPECTED_WITHOUT_GOLDEN = 3  # Агрокомплекс 62, 76, 76 (скользящий): 9–14 МБ
MAX_PARSE_SECONDS_SMALL = 5
MAX_PARSE_SECONDS_LARGE = 30
ROLLING_WARNING = "не с начала года"


def _relative(fixtures_dir: Path, file: Path) -> str:
    return file.relative_to(fixtures_dir / "osv-dkb").as_posix()


def _parse(file: Path) -> tuple[ParseOutcome, float]:
    started = time.monotonic()
    outcome = get_parser().parse(file.read_bytes(), cast(FileType, file.suffix.lstrip(".").upper()))
    return outcome, time.monotonic() - started


def _documents(outcome: ParseOutcome) -> list[OsvDocument]:
    assert outcome.error is None, outcome.error
    assert outcome.documents
    assert all(isinstance(document, OsvDocument) for document in outcome.documents)
    return cast(list[OsvDocument], list(outcome.documents))


def test_dkb_package_is_complete(fixtures_dir: Path) -> None:
    files = osv_dkb_files(fixtures_dir)

    assert len(files) == EXPECTED_FILES
    assert all(path.suffix.lower() in OSV_EXTENSIONS for path in files)
    assert len(files) - len(osv_dkb_golden_files(fixtures_dir)) == EXPECTED_WITHOUT_GOLDEN


def test_expectations_point_to_existing_files_and_sheets(fixtures_dir: Path) -> None:
    existing = {_relative(fixtures_dir, path) for path in osv_dkb_files(fixtures_dir)}

    assert set(load_expectations()) <= existing, sorted(set(load_expectations()) - existing)


def test_files_over_threshold_have_no_golden_snapshot(fixtures_dir: Path) -> None:
    """Большие файлы (десятки МБ JSON) снимка не имеют — только property-тесты; снимок у остальных обязателен."""
    for path in osv_dkb_files(fixtures_dir):
        snapshot = golden_path(fixtures_dir, path)
        assert snapshot.is_file() == (path.stat().st_size <= GOLDEN_MAX_BYTES), path.name


def _check_statuses(relative: str, outcome: ParseOutcome, documents: list[OsvDocument]) -> None:
    locators = [document.locator for document in documents]
    assert len(set(locators)) == len(locators)
    assert all(locator.startswith("sheet:") for locator in locators)
    # Статус результата — по нарушениям листов; ожидаемые нарушения — из манифеста
    found = {document.locator: violated_kinds(document) for document in documents}
    assert found == {locator: expected_violations(relative, locator) for locator in locators}, found
    expected_status = ResultStatus.NEEDS_CHECK if any(found.values()) else ResultStatus.DONE
    assert outcome.status is expected_status, outcome.metrics
    assert (relative in load_expectations()) == (expected_status is ResultStatus.NEEDS_CHECK)


def _check_period(directory: str, outcome: ParseOutcome, documents: list[OsvDocument]) -> None:
    expected = expected_period(directory)
    for document in documents:
        header = document.header
        assert header.period_year == expected.year, document.locator
        assert header.period_quarter == expected.quarter, document.locator
        has_rolling_warning = any(
            ROLLING_WARNING in warning for warning in (outcome.metrics.warnings or []) if outcome.metrics
        )
        assert has_rolling_warning is expected.rolling, document.locator
        if expected.rolling:
            assert header.is_cumulative is None
        else:
            assert header.is_cumulative is not None


def _check_structure(outcome: ParseOutcome, documents: list[OsvDocument]) -> None:
    total_data_rows = 0
    for document in documents:
        rows = document.rows
        header = document.header
        assert header.organization_name and header.account_code_raw and header.data_type, document.locator
        by_number = {row.row_number: row for row in rows}
        assert len(by_number) == len(rows)  # номера строк уникальны
        assert [row.row_number for row in rows] == sorted(by_number)
        totals = [row for row in rows if row.group_type == "TOTAL"]
        assert totals and all(row.level == 0 and row.parent_row_number is None for row in totals), document.locator
        for row in rows:
            if row.parent_row_number is None:
                assert row.level == 0
                continue
            parent = by_number[row.parent_row_number]
            assert parent.row_number < row.row_number
            assert parent.level < row.level
            assert parent.group_type != "TOTAL"
            if row.group_type == "ACCOUNT":
                assert row.account_code_raw
        accounts = [row for row in rows if row.group_type == "ACCOUNT"]
        assert accounts, document.locator
        assert all(
            row.account_code_raw and row.account_code_raw.startswith(header.account_code_raw) for row in accounts
        )
        total_data_rows += len(rows) - len(totals)
    assert outcome.metrics is not None
    assert outcome.metrics.rows == total_data_rows


def test_dkb_file_invariants(fixtures_dir: Path, dkb_file: Path) -> None:
    """Один разбор файла — все инварианты: статусы листов по манифесту, период каталога, структура строк, контракт."""
    outcome, seconds = _parse(dkb_file)
    documents = _documents(outcome)

    _check_statuses(_relative(fixtures_dir, dkb_file), outcome, documents)
    _check_period(dkb_file.parent.name, outcome, documents)
    _check_structure(outcome, documents)
    limit = MAX_PARSE_SECONDS_LARGE if dkb_file.stat().st_size > GOLDEN_MAX_BYTES else MAX_PARSE_SECONDS_SMALL
    assert seconds < limit, f"{dkb_file.name}: {seconds:.2f} с"
    assert_valid_result(outcome)