"""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 # Пакет ДКБ 06.10.2026 без отчётов «Анализ счета» (25) + ОСВ 60 Новопетровской (#32): 23 до порога снимка + 3 большие EXPECTED_FILES = 26 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)