2fce78080b
I-1: plan_sheets проверяет дедлайн на каждом листе (регрессия против main). I-2: тесты исключения пакета на кандидате (RECOGNITION_FAILED с предупреждениями файла), TIMEOUT в адаптере и конвейере, RulesError в get_parser. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W
1024 lines
48 KiB
Python
1024 lines
48 KiB
Python
"""Оркестратор `BookParser` (design D-6, D-8; spec «Листы ОСВ находятся по содержимому», «Контрольные суммы определяют
|
||
статус», «Форматы, лимиты, время», «Словари — данные, не код»).
|
||
|
||
Книги собираются openpyxl в памяти: лист описывается строками сверху вниз, как в `gridbuilder` (`M` — объединённая
|
||
ячейка, `L` — outline level строки первым элементом). Позиции колонок произвольные — парсер находит их по тексту.
|
||
"""
|
||
|
||
import datetime
|
||
import io
|
||
import itertools
|
||
import shutil
|
||
import time
|
||
import traceback
|
||
from collections.abc import Callable, Iterator, Sequence
|
||
from decimal import Decimal
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import openpyxl
|
||
import pytest
|
||
from gridbuilder import L, M, sheet_from_rows
|
||
from loguru import logger
|
||
from result_contract import assert_valid_result
|
||
|
||
from arm_statements import Rules as StatementRules
|
||
from arm_statements import default_rules
|
||
from src.contract import ErrorCode, FileWarning, OsvDocument, ResultStatus, StatementDocument
|
||
from src.parser import ParseOutcome, hierarchy
|
||
from src.parser.book import BookParser
|
||
from src.parser.classify import SUM_FIELDS
|
||
from src.parser.deadline import Deadline
|
||
from src.parser.errors import ParseTimeoutError
|
||
from src.parser.hierarchy import BodyRow
|
||
from src.parser.osv import OsvParser, _unique_notes
|
||
from src.parser.osv_warnings import OsvNote
|
||
from src.parser.readers.xls import NO_CODEPAGE_WARNING
|
||
from src.parser.rules import DEFAULT_RULES_DIR, HEADERS_FILE, Rules, load_rules
|
||
from src.parser.statements.pipeline import StatementPipeline
|
||
from tests.synthetic.xls_writer import write_xls
|
||
from tests.synthetic.xlsb_writer import write_xlsb
|
||
|
||
OPENING = "Сальдо на начало периода"
|
||
TURNOVER = "Обороты за период"
|
||
CLOSING = "Сальдо на конец периода"
|
||
TITLE = "Оборотно-сальдовая ведомость по счету 60 за 1 квартал 2025 г."
|
||
MAX_BYTES = 50 * 1024 * 1024
|
||
TIMEOUT_SECONDS = 120
|
||
# Строки мини-ОСВ: шапка 6 строк, блок заголовка 2 строки → данные с 9-й
|
||
ACCOUNT_ROW, SUPPLIER_ROW, CONTRACTOR_ROW, TOTAL_ROW = 9, 10, 11, 12
|
||
DATA_ROWS = 3
|
||
# Часы-заглушка: каждый вызов +100 с; лимит 150 с истекает на второй проверке дедлайна
|
||
CLOCK_STEP = 100
|
||
SHORT_TIMEOUT = 150
|
||
|
||
Cells = list[object]
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def rules() -> Rules:
|
||
return load_rules()
|
||
|
||
|
||
def xlsx(sheets: dict[str, Sequence[Cells]]) -> bytes:
|
||
"""Книга XLSX: имя листа → строки (значение, `M` — merged-диапазон, `L` — outline level первым элементом)."""
|
||
workbook = openpyxl.Workbook()
|
||
default = workbook.active
|
||
assert default is not None
|
||
workbook.remove(default)
|
||
for name, rows in sheets.items():
|
||
sheet = workbook.create_sheet(name)
|
||
for row_index, cells in enumerate(rows, start=1):
|
||
col = 1
|
||
for value in cells:
|
||
if isinstance(value, L):
|
||
sheet.row_dimensions[row_index].outline_level = value.level
|
||
continue
|
||
if isinstance(value, M):
|
||
sheet.cell(row_index, col, value.value)
|
||
sheet.merge_cells(
|
||
start_row=row_index,
|
||
start_column=col,
|
||
end_row=row_index + value.height - 1,
|
||
end_column=col + value.width - 1,
|
||
)
|
||
col += value.width
|
||
continue
|
||
if value is not None:
|
||
sheet.cell(row_index, col, value)
|
||
col += 1
|
||
buffer = io.BytesIO()
|
||
workbook.save(buffer)
|
||
return buffer.getvalue()
|
||
|
||
|
||
def head(opening: str = OPENING, title: str = TITLE) -> list[Cells]:
|
||
"""Шапка варианта 1 и блок заголовка «Счет | начало | обороты | конец», группировки Счет / Контрагенты (с ИНН)."""
|
||
return [
|
||
['АО "Ромашка"'], # 1
|
||
[title], # 2
|
||
[], # 3
|
||
["Выводимые данные: БУ (данные бухгалтерского учета)"], # 4
|
||
["Параметры:", None, "Валюта БУ, НУ: RUB"], # 5
|
||
[], # 6
|
||
["Счет", None, M(opening, 2), M(TURNOVER, 2), M(CLOSING, 2)], # 7
|
||
["Контрагенты", "ИНН", "Дебет", "Кредит", "Дебет", "Кредит", "Дебет", "Кредит"], # 8
|
||
]
|
||
|
||
|
||
def data(level: int, text: str, sums: Sequence[object], inn: object = None) -> Cells:
|
||
return [L(level), text, inn, *sums]
|
||
|
||
|
||
ROOT = (100, None, 50, 30, 120, None)
|
||
SUPPLIER = (60, None, 20, 10, 70, None)
|
||
CONTRACTOR = (40, None, 30, 20, 50, None)
|
||
|
||
|
||
def mini_osv(supplier: Sequence[object] = SUPPLIER, *, total: bool = True, opening: str = OPENING) -> list[Cells]:
|
||
rows = [
|
||
*head(opening),
|
||
data(0, "60", ROOT), # 9
|
||
data(1, "ООО Поставщик", supplier, inn="7701234567"), # 10
|
||
data(1, "ООО Подрядчик", CONTRACTOR, inn="7707654321"), # 11
|
||
]
|
||
if total:
|
||
rows.append(data(0, "Итого", ROOT)) # 12
|
||
return rows
|
||
|
||
|
||
def statement() -> list[Cells]:
|
||
return [["Бухгалтерский баланс"], ["на 31 марта 2025 г."], ["Актив", "Код", "На 31 марта 2025 г."]]
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def statement_rules() -> StatementRules:
|
||
return default_rules()
|
||
|
||
|
||
def book(
|
||
rules: Rules,
|
||
*,
|
||
max_bytes: int = MAX_BYTES,
|
||
timeout_seconds: float = TIMEOUT_SECONDS,
|
||
clock: Callable[[], float] = time.monotonic,
|
||
) -> BookParser:
|
||
statement_rules = default_rules()
|
||
return BookParser(
|
||
rules,
|
||
osv=OsvParser(rules),
|
||
statements=StatementPipeline(statement_rules),
|
||
statement_rules=statement_rules,
|
||
max_bytes=max_bytes,
|
||
timeout_seconds=timeout_seconds,
|
||
clock=clock,
|
||
)
|
||
|
||
|
||
def only_document(outcome: ParseOutcome) -> OsvDocument:
|
||
(document,) = outcome.documents
|
||
assert isinstance(document, OsvDocument)
|
||
return document
|
||
|
||
|
||
def file_warnings_of(outcome: ParseOutcome) -> list[FileWarning]:
|
||
"""Предупреждения файла (`metrics.warnings`)."""
|
||
assert outcome.metrics is not None
|
||
assert outcome.metrics.warnings is not None
|
||
return outcome.metrics.warnings
|
||
|
||
|
||
def warnings_of(outcome: ParseOutcome) -> list[str]:
|
||
"""Сообщения предупреждений документов ОСВ (контракт 2.0: лежат в самих документах), по порядку документов."""
|
||
return [
|
||
warning.message
|
||
for document in outcome.documents
|
||
if isinstance(document, OsvDocument)
|
||
for warning in document.warnings
|
||
]
|
||
|
||
|
||
def file_codes_of(outcome: ParseOutcome) -> list[str]:
|
||
return [warning.code for warning in file_warnings_of(outcome)]
|
||
|
||
|
||
# --- Штатный разбор ---
|
||
|
||
|
||
def test_mini_osv_is_done_with_full_fields(rules: Rules) -> None:
|
||
outcome = book(rules).parse(xlsx({"ОСВ 60": mini_osv()}), "XLSX")
|
||
|
||
assert outcome.status is ResultStatus.DONE
|
||
assert outcome.error is None
|
||
document = only_document(outcome)
|
||
assert document.document_type == "OSV"
|
||
assert document.locator == "sheet:ОСВ 60"
|
||
header = document.header
|
||
assert (header.organization_name, header.account_code_raw, header.period_text) == (
|
||
'АО "Ромашка"',
|
||
"60",
|
||
"1 квартал 2025 г.",
|
||
)
|
||
assert (header.period_year, header.period_quarter, header.is_cumulative) == (2025, "Q1", True)
|
||
assert (header.data_type, header.currency, header.header_variant) == ("BU", "RUB", 1)
|
||
assert [(group.type, group.name, group.has_inn) for group in document.group_schema] == [
|
||
("ACCOUNT", "Счет", None),
|
||
("COUNTERPARTY", "Контрагенты", True),
|
||
]
|
||
assert document.columns == {
|
||
"opening_debit": "C",
|
||
"opening_credit": "D",
|
||
"turnover_debit": "E",
|
||
"turnover_credit": "F",
|
||
"closing_debit": "G",
|
||
"closing_credit": "H",
|
||
"inn": "B",
|
||
}
|
||
rows = document.rows
|
||
assert [(row.row_number, row.level, row.parent_row_number, row.group_type) for row in rows] == [
|
||
(ACCOUNT_ROW, 0, None, "ACCOUNT"),
|
||
(SUPPLIER_ROW, 1, ACCOUNT_ROW, "COUNTERPARTY"),
|
||
(CONTRACTOR_ROW, 1, ACCOUNT_ROW, "COUNTERPARTY"),
|
||
(TOTAL_ROW, 0, None, "TOTAL"),
|
||
]
|
||
supplier = rows[1]
|
||
assert (supplier.row_value, supplier.counterparty_name, supplier.counterparty_inn) == (
|
||
"ООО Поставщик",
|
||
"ООО Поставщик",
|
||
"7701234567",
|
||
)
|
||
assert (supplier.group_name, supplier.account_code_raw, supplier.data_type) == ("Контрагенты", "60", "BU")
|
||
sums = [getattr(supplier, name) for name in SUM_FIELDS]
|
||
assert sums == [Decimal(60), None, Decimal(20), Decimal(10), Decimal(70), None]
|
||
assert document.checks is not None
|
||
assert document.checks.total_row_matches_root is True
|
||
assert document.checks.balance_identity_violations == []
|
||
assert document.checks.parent_sum_violations == []
|
||
assert outcome.metrics is not None
|
||
assert outcome.metrics.rows == DATA_ROWS
|
||
assert outcome.metrics.duration_ms is not None
|
||
assert outcome.metrics.duration_ms >= 0
|
||
assert warnings_of(outcome) == ["иерархия строк — по группировке листа (outline)"]
|
||
assert (document.status, outcome.metrics.warnings) == ("done", [])
|
||
payload = assert_valid_result(outcome)
|
||
assert payload["schema_version"] == "2.0"
|
||
|
||
|
||
def test_spoiled_sum_needs_check_with_row_numbers(rules: Rules) -> None:
|
||
"""Spec «Испорченная сумма»: `closing_debit` контрагента +1 → тождество в его строке и сумма родителя."""
|
||
spoiled = (60, None, 20, 10, 71, None)
|
||
|
||
outcome = book(rules).parse(xlsx({"ОСВ 60": mini_osv(spoiled)}), "XLSX")
|
||
|
||
assert outcome.status is ResultStatus.NEEDS_CHECK
|
||
checks = only_document(outcome).checks
|
||
assert checks is not None
|
||
assert checks.balance_identity_violations == [SUPPLIER_ROW]
|
||
assert checks.parent_sum_violations == [ACCOUNT_ROW]
|
||
assert checks.total_row_matches_root is True
|
||
warnings = warnings_of(outcome)
|
||
assert warnings[0] == "иерархия строк — по группировке листа (outline)"
|
||
assert any(f"строки {SUPPLIER_ROW}" in warning for warning in warnings), warnings
|
||
assert any(f"строки {ACCOUNT_ROW}" in warning for warning in warnings), warnings
|
||
assert_valid_result(outcome)
|
||
|
||
|
||
def test_period_bounds_reach_osv_header(rules: Rules) -> None:
|
||
"""Контракт v1.2: границы периода и признак «с начала года» заполнены парсером ОСВ."""
|
||
header = only_document(book(rules).parse(xlsx({"ОСВ 60": mini_osv()}), "XLSX")).header
|
||
|
||
assert (header.period_start, header.period_end, header.is_from_year_start) == (
|
||
datetime.date(2025, 1, 1),
|
||
datetime.date(2025, 3, 31),
|
||
True,
|
||
)
|
||
|
||
|
||
def test_period_bounds_reach_osv_header_not_from_year_start(rules: Rules) -> None:
|
||
title = TITLE.replace("1 квартал", "3 квартал")
|
||
rows = [*head(title=title), *mini_osv()[len(head()) :]]
|
||
|
||
header = only_document(book(rules).parse(xlsx({"ОСВ 60": rows}), "XLSX")).header
|
||
|
||
assert (header.period_start, header.period_end, header.is_from_year_start) == (
|
||
datetime.date(2025, 7, 1),
|
||
datetime.date(2025, 9, 30),
|
||
False,
|
||
)
|
||
|
||
|
||
def test_missing_total_needs_check(rules: Rules) -> None:
|
||
"""R-20: нет «Итого» — сверку с корнем выполнить нельзя: `false`, предупреждение, `needs_check`."""
|
||
outcome = book(rules).parse(xlsx({"ОСВ 60": mini_osv(total=False)}), "XLSX")
|
||
|
||
assert outcome.status is ResultStatus.NEEDS_CHECK
|
||
checks = only_document(outcome).checks
|
||
assert checks is not None
|
||
assert checks.total_row_matches_root is False
|
||
assert "строка «Итого» не найдена" in warnings_of(outcome)
|
||
|
||
|
||
def test_empty_osv_is_done_with_total_only(rules: Rules) -> None:
|
||
"""Spec «Пустая ОСВ» (R-13): заголовок и сразу «Итого» без сумм."""
|
||
outcome = book(rules).parse(xlsx({"ОСВ 66": [*head(), [None, "Итого"]]}), "XLSX")
|
||
|
||
assert outcome.status is ResultStatus.DONE
|
||
document = only_document(outcome)
|
||
assert [(row.group_type, row.level, row.parent_row_number) for row in document.rows] == [("TOTAL", 0, None)]
|
||
assert document.checks is not None
|
||
assert document.checks.total_row_matches_root is True
|
||
assert outcome.metrics is not None
|
||
assert outcome.metrics.rows == 0
|
||
assert warnings_of(outcome) == ["иерархия строк — плоский список (trivial)", "нет строк данных"]
|
||
assert_valid_result(outcome)
|
||
|
||
|
||
def test_number_warnings_carry_row_numbers(rules: Rules) -> None:
|
||
rows = mini_osv()
|
||
rows[CONTRACTOR_ROW - 1] = data(1, "ООО Подрядчик", (40, None, 30, 20, 50, "н/д"), inn="7707654321")
|
||
|
||
outcome = book(rules).parse(xlsx({"ОСВ 60": rows}), "XLSX")
|
||
|
||
assert outcome.status is ResultStatus.DONE
|
||
assert any(warning.startswith(f"строка {CONTRACTOR_ROW}:") for warning in warnings_of(outcome))
|
||
|
||
|
||
# --- Шапка без меток и код/наименование счёта в двух колонках (issue #11 п. 4, OQ-134) ---
|
||
|
||
NO_LABEL_ORG_SUMS = (None, 600, 100, None, None, 500)
|
||
NO_LABEL_SUPPLIER = (None, 400, None, None, None, 400)
|
||
NO_LABEL_CONTRACTOR = (None, 200, 100, None, None, 100)
|
||
|
||
|
||
def code_name_osv(*, total_label: str | None = None) -> list[Cells]:
|
||
"""ОСВ 63 Новоросметалл: «Бухгалтерский | период» отдельной строкой, организация — в названии ведомости, колонки
|
||
«Счет: Код / Наименование», контрагенты — в колонке правее кода (уровень — сдвиг колонки), «Итого» без подписи.
|
||
"""
|
||
blank = [None] * 3
|
||
sums_head = ["Дебет", "Кредит", "Дебет", "Кредит", "Дебет", "Кредит"]
|
||
|
||
def line(label_cells: Sequence[object], sums: Sequence[object]) -> Cells:
|
||
return [*label_cells, *sums]
|
||
|
||
return [
|
||
["Оборотно-сальдовая ведомость НОВОРОСМЕТАЛЛ ООО"], # 1
|
||
[], # 2
|
||
["Бухгалтерский", *([None] * 10), "9 месяцев 2025 г."], # 3
|
||
["Счет", *([None] * 5), M(OPENING, 2), M(TURNOVER, 2), M(CLOSING, 2)], # 4
|
||
["Код", *blank, "Наименование", None, *sums_head], # 5
|
||
line([63, *blank, "Резервы по сомнительным долгам", None], NO_LABEL_ORG_SUMS), # 6
|
||
line([None, "ООО Поставщик", None, None, None, None], NO_LABEL_SUPPLIER), # 7
|
||
line([None, "ООО Подрядчик", None, None, None, None], NO_LABEL_CONTRACTOR), # 8
|
||
line([total_label, None, None, None, None, None], NO_LABEL_ORG_SUMS), # 9
|
||
]
|
||
|
||
|
||
def test_header_without_labels_and_code_name_columns_are_found(rules: Rules) -> None:
|
||
outcome = book(rules).parse(xlsx({"Лист1": code_name_osv()}), "XLSX")
|
||
|
||
document = only_document(outcome)
|
||
header = document.header
|
||
assert header.organization_name == "НОВОРОСМЕТАЛЛ ООО"
|
||
assert header.account_code_raw == "63"
|
||
assert (header.period_text, header.period_year, header.period_quarter, header.is_cumulative) == (
|
||
"9 месяцев 2025 г.",
|
||
2025,
|
||
"Q3",
|
||
True,
|
||
)
|
||
assert header.data_type == "BU"
|
||
warnings = warnings_of(outcome)
|
||
assert not any("период" in warning and "не найден" in warning for warning in warnings), warnings
|
||
assert not any("тип выводимых данных" in warning for warning in warnings), warnings
|
||
|
||
|
||
def test_code_name_columns_rows_are_account_and_children_by_label_column(rules: Rules) -> None:
|
||
outcome = book(rules).parse(xlsx({"Лист1": code_name_osv(total_label="Итого")}), "XLSX")
|
||
|
||
document = only_document(outcome)
|
||
assert [(row.row_value, row.level, row.group_type) for row in document.rows] == [
|
||
("63", 0, "ACCOUNT"),
|
||
("ООО Поставщик", 1, "OTHER"),
|
||
("ООО Подрядчик", 1, "OTHER"),
|
||
("Итого", 0, "TOTAL"),
|
||
]
|
||
assert document.rows[0].extra == {"Наименование": "Резервы по сомнительным долгам"}
|
||
assert outcome.status is ResultStatus.DONE
|
||
|
||
|
||
def test_total_row_without_caption_is_the_last_row_of_sums(rules: Rules) -> None:
|
||
outcome = book(rules).parse(xlsx({"Лист1": code_name_osv()}), "XLSX")
|
||
|
||
document = only_document(outcome)
|
||
assert document.rows[-1].group_type == "TOTAL"
|
||
assert document.checks is not None
|
||
assert document.checks.total_row_matches_root is True
|
||
assert outcome.status is ResultStatus.DONE
|
||
assert any("без подписи" in warning for warning in warnings_of(outcome)), warnings_of(outcome)
|
||
|
||
|
||
# --- Несколько листов ---
|
||
|
||
|
||
def test_statement_sheet_goes_to_real_pipeline_next_to_osv(rules: Rules) -> None:
|
||
outcome = book(rules).parse(xlsx({"Баланс": statement(), "ОСВ 60": mini_osv()}), "XLSX")
|
||
|
||
assert [type(document).__name__ for document in outcome.documents] == ["OsvDocument", "StatementDocument"]
|
||
assert outcome.documents[0].locator == "sheet:ОСВ 60" and outcome.documents[1].locator == "sheet:Баланс"
|
||
assert "иерархия строк — по группировке листа (outline)" in warnings_of(outcome)
|
||
assert_valid_result(outcome)
|
||
|
||
|
||
def test_two_osv_sheets_and_other_sheet(rules: Rules) -> None:
|
||
"""Spec «Два листа ОСВ и посторонний лист»: два документа в порядке листов, пропуск «Справки» в warnings."""
|
||
other: list[Cells] = [["Справка о дебиторской задолженности"], ["Контрагент", "Сумма"], ["ООО Лютик", 5]]
|
||
content = xlsx({"ОСВ 60": mini_osv(), "Справка": other, "ОСВ 62": mini_osv()})
|
||
|
||
outcome = book(rules).parse(content, "XLSX")
|
||
|
||
assert outcome.status is ResultStatus.DONE
|
||
assert [document.locator for document in outcome.documents] == ["sheet:ОСВ 60", "sheet:ОСВ 62"]
|
||
assert outcome.metrics is not None
|
||
assert outcome.metrics.rows == 2 * DATA_ROWS
|
||
for document in outcome.documents:
|
||
assert isinstance(document, OsvDocument)
|
||
assert [w.code for w in document.warnings] == ["OSV_HIERARCHY_BY_OUTLINE"]
|
||
(other_warning,) = file_warnings_of(outcome)
|
||
assert (other_warning.code, other_warning.severity, other_warning.locator) == (
|
||
"SHEET_OTHER",
|
||
"info",
|
||
"sheet:Справка",
|
||
)
|
||
assert other_warning.message == "лист без признаков ОСВ и формы"
|
||
assert_valid_result(outcome)
|
||
|
||
|
||
def test_one_spoiled_sheet_makes_whole_result_needs_check(rules: Rules) -> None:
|
||
spoiled = (60, None, 20, 10, 71, None)
|
||
content = xlsx({"ОСВ 60": mini_osv(), "ОСВ 62": mini_osv(spoiled)})
|
||
|
||
outcome = book(rules).parse(content, "XLSX")
|
||
|
||
assert outcome.status is ResultStatus.NEEDS_CHECK
|
||
assert len(outcome.documents) == 2 # noqa: PLR2004 - два листа ОСВ
|
||
|
||
|
||
# --- Ошибки (карта D-6) ---
|
||
|
||
|
||
def test_book_without_osv_is_invalid_document(rules: Rules) -> None:
|
||
"""Spec «Нет ОСВ»: сообщение называет число просмотренных листов."""
|
||
outcome = book(rules).parse(
|
||
xlsx({"Справка": [["Наименование", "Сумма"], ["Аренда", 100]], "Данные": [["a", "b"], [1, 2]]}), "XLSX"
|
||
)
|
||
|
||
assert outcome.status is ResultStatus.ERROR
|
||
assert outcome.error is not None
|
||
assert outcome.error.code is ErrorCode.INVALID_DOCUMENT
|
||
assert "листов просмотрено 2" in outcome.error.message
|
||
assert outcome.documents == ()
|
||
|
||
|
||
def test_osv_sheet_without_table_header_is_invalid_format(rules: Rules) -> None:
|
||
rows: list[Cells] = [['АО "Ромашка"'], [TITLE], [], ["Счет", "Сумма"], ["60", 100]]
|
||
|
||
outcome = book(rules).parse(xlsx({"ОСВ 60": rows}), "XLSX")
|
||
|
||
assert outcome.status is ResultStatus.ERROR
|
||
assert outcome.error is not None
|
||
assert outcome.error.code is ErrorCode.INVALID_FORMAT
|
||
assert "«ОСВ 60»" in outcome.error.message
|
||
assert "заголовок таблицы" in outcome.error.message
|
||
assert outcome.error.message.count("ОСВ 60") == 1
|
||
|
||
|
||
def test_undetermined_hierarchy_is_invalid_format_with_sheet_and_step(
|
||
rules: Rules, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
def nothing(rows: Sequence[BodyRow], deadline: Deadline | None = None) -> None:
|
||
return None
|
||
|
||
monkeypatch.setattr(hierarchy, "STRATEGIES", (("nothing", nothing),))
|
||
|
||
outcome = book(rules).parse(xlsx({"ОСВ 60": mini_osv()}), "XLSX")
|
||
|
||
assert outcome.error is not None
|
||
assert outcome.error.code is ErrorCode.INVALID_FORMAT
|
||
assert "«ОСВ 60»" in outcome.error.message
|
||
assert "иерархия" in outcome.error.message
|
||
|
||
|
||
def test_size_over_limit_is_unsupported_format_before_reading(rules: Rules) -> None:
|
||
"""Spec «Превышен лимит»: на байт больше лимита — `UNSUPPORTED_FORMAT`, ровно лимит — разбор."""
|
||
content = xlsx({"ОСВ 60": mini_osv()})
|
||
|
||
over = book(rules, max_bytes=len(content) - 1).parse(content, "XLSX")
|
||
exact = book(rules, max_bytes=len(content)).parse(content, "XLSX")
|
||
|
||
assert over.status is ResultStatus.ERROR
|
||
assert over.error is not None
|
||
assert over.error.code is ErrorCode.UNSUPPORTED_FORMAT
|
||
assert "лимит" in over.error.message
|
||
assert exact.status is ResultStatus.DONE
|
||
|
||
|
||
def test_tiny_limit_rejects_any_file(rules: Rules) -> None:
|
||
outcome = book(rules, max_bytes=10).parse(b"PK\x03\x04" + b"x" * 7, "XLSX")
|
||
|
||
assert outcome.error is not None
|
||
assert outcome.error.code is ErrorCode.UNSUPPORTED_FORMAT
|
||
|
||
|
||
def test_unknown_signature_is_unsupported_format(rules: Rules) -> None:
|
||
outcome = book(rules).parse(b"%PDF-1.7 not a workbook", "XLSX")
|
||
|
||
assert outcome.error is not None
|
||
assert outcome.error.code is ErrorCode.UNSUPPORTED_FORMAT
|
||
|
||
|
||
def test_broken_zip_is_unreadable(rules: Rules) -> None:
|
||
outcome = book(rules).parse(b"PK\x03\x04 broken archive", "XLSX")
|
||
|
||
assert outcome.error is not None
|
||
assert outcome.error.code is ErrorCode.UNREADABLE
|
||
|
||
|
||
def test_deadline_exceeded_is_timeout(rules: Rules) -> None:
|
||
clock = itertools.count(step=CLOCK_STEP)
|
||
|
||
outcome = book(rules, timeout_seconds=SHORT_TIMEOUT, clock=lambda: float(next(clock))).parse(
|
||
xlsx({"ОСВ 60": mini_osv()}), "XLSX"
|
||
)
|
||
|
||
assert outcome.status is ResultStatus.ERROR
|
||
assert outcome.error is not None
|
||
assert outcome.error.code is ErrorCode.TIMEOUT
|
||
assert "превышен лимит" in outcome.error.message
|
||
|
||
|
||
def test_sheet_kind_classification_is_interrupted_by_deadline(rules: Rules) -> None:
|
||
"""I-1: классификация листов (`plan_sheets`) проверяет дедлайн на каждом листе, книга из «прочего» — TIMEOUT."""
|
||
notes: list[Cells] = [["Примечания"], ["Текст без кодов и сумм"]]
|
||
clock = itertools.count(step=CLOCK_STEP)
|
||
content = xlsx({"Лист 1": notes, "Лист 2": notes, "Лист 3": notes})
|
||
|
||
outcome = book(rules, timeout_seconds=SHORT_TIMEOUT, clock=lambda: float(next(clock))).parse(content, "XLSX")
|
||
|
||
assert outcome.error is not None
|
||
assert outcome.error.code is ErrorCode.TIMEOUT
|
||
assert "вид листа" in outcome.error.message
|
||
|
||
|
||
def test_package_exception_on_candidate_is_recognition_failed_with_collected_warnings(
|
||
rules: Rules, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""I-2(а), P4-11: исключение пакета на кандидате — ошибка книги, предупреждения файла до него сохраняются."""
|
||
|
||
def broken(candidate: object, statement_rules: object) -> object:
|
||
raise ValueError("сбой пакета\nвторая строка")
|
||
|
||
monkeypatch.setattr("src.parser.statements.pipeline.normalize_candidate", broken)
|
||
notes: list[Cells] = [["Примечания"], ["Текст без кодов и сумм"]]
|
||
|
||
outcome = book(rules).parse(xlsx({"Примечания": notes, "ББ": BALANCE_ROWS}), "XLSX")
|
||
|
||
assert outcome.status is ResultStatus.ERROR
|
||
assert outcome.error is not None
|
||
assert outcome.error.code is ErrorCode.RECOGNITION_FAILED
|
||
assert outcome.error.message == "ValueError: сбой пакета вторая строка"
|
||
assert [(w.code, w.locator) for w in file_warnings_of(outcome)] == [("SHEET_OTHER", "sheet:Примечания")]
|
||
|
||
|
||
def test_deadline_on_forms_path_is_timeout_in_adapter(rules: Rules) -> None:
|
||
"""I-2(б): дедлайн истекает на сборке кандидатов адаптера (после классификации листа)."""
|
||
ticks = itertools.chain([0.0] * 3, itertools.repeat(1e9)) # старт, создание дедлайна, вид листа
|
||
|
||
outcome = book(rules, timeout_seconds=100, clock=lambda: next(ticks)).parse(xlsx({"ББ": BALANCE_ROWS}), "XLSX")
|
||
|
||
assert outcome.error is not None
|
||
assert outcome.error.code is ErrorCode.TIMEOUT
|
||
assert "форма" in outcome.error.message
|
||
|
||
|
||
def test_deadline_on_forms_path_is_timeout_in_pipeline(rules: Rules) -> None:
|
||
"""I-2(б): дедлайн истекает перед нормализацией кандидата в `StatementPipeline`."""
|
||
ticks = itertools.chain([0.0] * 4, itertools.repeat(1e9)) # + проверка адаптера
|
||
|
||
outcome = book(rules, timeout_seconds=100, clock=lambda: next(ticks)).parse(xlsx({"ББ": BALANCE_ROWS}), "XLSX")
|
||
|
||
assert outcome.error is not None
|
||
assert outcome.error.code is ErrorCode.TIMEOUT
|
||
assert "форма «sheet:ББ»" in outcome.error.message
|
||
|
||
|
||
def test_unexpected_exception_is_recognition_failed_in_one_line(rules: Rules, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
def broken(rows: Sequence[BodyRow], deadline: Deadline | None = None) -> dict[int, int] | None:
|
||
raise ValueError("сбой стратегии\nвторая строка")
|
||
|
||
monkeypatch.setattr(hierarchy, "STRATEGIES", (("broken", broken),))
|
||
|
||
outcome = book(rules).parse(xlsx({"ОСВ 60": mini_osv()}), "XLSX")
|
||
|
||
assert outcome.status is ResultStatus.ERROR
|
||
assert outcome.error is not None
|
||
assert outcome.error.code is ErrorCode.RECOGNITION_FAILED
|
||
assert outcome.error.message == "ValueError: сбой стратегии вторая строка"
|
||
|
||
|
||
@pytest.fixture
|
||
def log_records() -> Iterator[list[Any]]:
|
||
"""Записи loguru, выпущенные за время теста."""
|
||
records: list[Any] = []
|
||
handler_id = logger.add(lambda message: records.append(message.record), level="DEBUG")
|
||
yield records
|
||
logger.remove(handler_id)
|
||
|
||
|
||
def test_unexpected_exception_is_logged_with_traceback(
|
||
rules: Rules, monkeypatch: pytest.MonkeyPatch, log_records: list[Any]
|
||
) -> None:
|
||
"""I-2: ответ прежний (RECOGNITION_FAILED одной строкой), но стек непредвиденной ошибки уходит в лог."""
|
||
|
||
def broken(rows: Sequence[BodyRow], deadline: Deadline | None = None) -> dict[int, int] | None:
|
||
raise ValueError("сбой стратегии\nвторая строка")
|
||
|
||
monkeypatch.setattr(hierarchy, "STRATEGIES", (("broken", broken),))
|
||
|
||
outcome = book(rules).parse(xlsx({"ОСВ 60": mini_osv()}), "XLSX")
|
||
|
||
assert outcome.error is not None
|
||
assert outcome.error.code is ErrorCode.RECOGNITION_FAILED
|
||
assert outcome.error.message == "ValueError: сбой стратегии вторая строка"
|
||
(record,) = [record for record in log_records if record["level"].name == "ERROR"]
|
||
assert "Непредвиденная ошибка разбора" in record["message"]
|
||
assert "ValueError: сбой стратегии вторая строка" in record["message"]
|
||
exception = record["exception"]
|
||
assert exception is not None
|
||
assert exception.type is ValueError
|
||
assert exception.traceback is not None
|
||
assert "broken" in "".join(traceback.format_tb(exception.traceback))
|
||
|
||
|
||
def test_unreadable_book_is_logged_as_warning_without_traceback(rules: Rules, log_records: list[Any]) -> None:
|
||
outcome = book(rules).parse(b"PK\x03\x04 broken archive", "XLSX")
|
||
|
||
assert outcome.error is not None
|
||
assert outcome.error.code is ErrorCode.UNREADABLE
|
||
(record,) = [record for record in log_records if record["level"].name == "WARNING"]
|
||
assert outcome.error.message in record["message"]
|
||
assert record["exception"] is None
|
||
assert not [record for record in log_records if record["level"].name == "ERROR"]
|
||
|
||
|
||
def test_expected_errors_other_than_unreadable_are_not_logged(rules: Rules, log_records: list[Any]) -> None:
|
||
outcome = book(rules).parse(b"%PDF-1.7", "XLSX")
|
||
|
||
assert outcome.error is not None
|
||
assert outcome.error.code is ErrorCode.UNSUPPORTED_FORMAT
|
||
assert not [record for record in log_records if record["level"].no >= logger.level("WARNING").no]
|
||
|
||
|
||
def test_header_search_is_interrupted_by_deadline(rules: Rules) -> None:
|
||
"""R-23 (в): дедлайн из `parse_sheet` доходит до поиска заголовка таблицы — цикла по всему листу."""
|
||
sheet = sheet_from_rows([[TITLE], *([["строка"]] * 1000)])
|
||
deadline = Deadline(1, clock=itertools.chain([0.0], itertools.repeat(1e9)).__next__)
|
||
|
||
with pytest.raises(ParseTimeoutError, match="поиск заголовка таблицы"):
|
||
OsvParser(rules).parse_sheet(sheet, deadline)
|
||
|
||
|
||
# --- Словари — данные (R-8, spec «Новый синоним без релиза кода») ---
|
||
|
||
NEW_OPENING = "Нач. остаток"
|
||
NEW_SYNONYM_LINE = " - 'нач\\.\\s*остаток'\n"
|
||
|
||
|
||
def rules_with_new_opening_synonym(tmp_path: Path) -> Path:
|
||
"""Копия штатного каталога словарей, в `osv_headers.yaml` синоним «Нач. остаток» добавлен к группе opening."""
|
||
copy = tmp_path / "rules"
|
||
shutil.copytree(DEFAULT_RULES_DIR, copy)
|
||
headers = copy / HEADERS_FILE
|
||
text = headers.read_text(encoding="utf-8")
|
||
anchor = " opening:\n"
|
||
assert text.count(anchor) == 1
|
||
headers.write_text(text.replace(anchor, anchor + NEW_SYNONYM_LINE), encoding="utf-8")
|
||
return copy
|
||
|
||
|
||
def test_new_synonym_parses_without_code_changes(tmp_path: Path, rules: Rules) -> None:
|
||
content = xlsx({"ОСВ 60": mini_osv(opening=NEW_OPENING)})
|
||
|
||
outcome = book(load_rules(rules_with_new_opening_synonym(tmp_path))).parse(content, "XLSX")
|
||
|
||
assert outcome.status is ResultStatus.DONE
|
||
document = only_document(outcome)
|
||
assert document.columns is not None
|
||
assert (document.columns["opening_debit"], document.columns["opening_credit"]) == ("C", "D")
|
||
assert document.rows[0].opening_debit == Decimal(100)
|
||
assert_valid_result(outcome)
|
||
|
||
|
||
def test_default_dictionary_does_not_know_the_new_synonym(rules: Rules) -> None:
|
||
"""До правки словаря «Нач. остаток» не распознаётся: заголовок находится по «оборотам» и «концу» (начало
|
||
необязательно, R-12), колонки начала уходят в `extra`, `opening_*` = null, и тождество сальдо не сходится в
|
||
строках с ненулевым началом → `needs_check` (фактическое поведение, зафиксировано тестом).
|
||
"""
|
||
outcome = book(rules).parse(xlsx({"ОСВ 60": mini_osv(opening=NEW_OPENING)}), "XLSX")
|
||
|
||
assert outcome.status is ResultStatus.NEEDS_CHECK
|
||
document = only_document(outcome)
|
||
assert document.columns is not None
|
||
assert "opening_debit" not in document.columns
|
||
assert all(row.opening_debit is None and row.opening_credit is None for row in document.rows)
|
||
assert document.rows[0].extra is not None
|
||
assert "100" in document.rows[0].extra.values()
|
||
checks = document.checks
|
||
assert checks is not None
|
||
assert checks.balance_identity_violations == [ACCOUNT_ROW, SUPPLIER_ROW, CONTRACTOR_ROW, TOTAL_ROW]
|
||
|
||
|
||
# --- Формы (оркестрация книги, общий код arm_statements) ---
|
||
|
||
|
||
def test_statement_sheets_go_to_pipeline_in_sheet_order(rules: Rules) -> None:
|
||
content = xlsx({"ББ": BALANCE_ROWS, "Лист_1": mini_osv(), "ОФР": INCOME_ROWS})
|
||
|
||
outcome = book(rules).parse(content, "XLSX")
|
||
|
||
assert [type(document).__name__ for document in outcome.documents] == [
|
||
"OsvDocument",
|
||
"StatementDocument",
|
||
"StatementDocument",
|
||
]
|
||
assert [document.locator for document in outcome.documents] == ["sheet:Лист_1", "sheet:ББ", "sheet:ОФР"]
|
||
assert_valid_result(outcome)
|
||
|
||
|
||
def test_osv_document_has_status_and_osv_warnings(rules: Rules) -> None:
|
||
"""Spec «Предупреждение ОСВ»: статус документа и `OSV_*` в самом документе, а не в `metrics.warnings`."""
|
||
spoiled = (60, None, 20, 10, 71, None)
|
||
|
||
outcome = book(rules).parse(xlsx({"ОСВ 60": mini_osv(spoiled)}), "XLSX")
|
||
|
||
document = only_document(outcome)
|
||
assert document.status == "needs_check"
|
||
assert all(w.severity == "info" and w.code.startswith("OSV_") for w in document.warnings)
|
||
assert "OSV_BALANCE_IDENTITY" in {w.code for w in document.warnings}
|
||
assert outcome.metrics is not None and outcome.metrics.warnings == []
|
||
assert not any(w.message.startswith("sheet=") for w in file_warnings_of(outcome))
|
||
assert outcome.status is ResultStatus.NEEDS_CHECK
|
||
|
||
|
||
def test_osv_notes_are_unique_by_code_and_message(rules: Rules) -> None:
|
||
"""T6.1: повтор считается по паре (код, текст): одинаковый текст под разными кодами остаётся."""
|
||
notes = [OsvNote("A", "текст"), OsvNote("B", "текст"), OsvNote("A", "текст")]
|
||
assert _unique_notes(notes) == [OsvNote("A", "текст"), OsvNote("B", "текст")]
|
||
|
||
|
||
def test_other_sheet_is_file_warning(rules: Rules) -> None:
|
||
"""Spec «Лист «прочее»»: документ формы не создаётся, предупреждение файла с `locator`."""
|
||
notes: list[Cells] = [["Примечания"], ["Текст без кодов и сумм"]]
|
||
|
||
outcome = book(rules).parse(xlsx({"ББ": BALANCE_ROWS, "Примечания": notes}), "XLSX")
|
||
|
||
assert len(outcome.documents) == 1
|
||
(warning,) = file_warnings_of(outcome)
|
||
assert (warning.code, warning.severity, warning.locator) == ("SHEET_OTHER", "info", "sheet:Примечания")
|
||
assert warning.message == "лист без признаков ОСВ и формы"
|
||
|
||
|
||
def test_file_type_mismatch_is_info_file_warning(rules: Rules) -> None:
|
||
outcome = book(rules).parse(xlsx({"ОСВ 60": mini_osv()}), "XLS")
|
||
|
||
(warning,) = file_warnings_of(outcome)
|
||
assert (warning.code, warning.severity, warning.locator) == ("FILE_TYPE_MISMATCH", "info", None)
|
||
assert warning.message == "file_type=XLS, но содержимое OOXML: прочитано как XLSX"
|
||
assert outcome.status is ResultStatus.DONE
|
||
|
||
|
||
def _xls_cells(rows: Sequence[Cells]) -> dict[tuple[int, int], object]:
|
||
return {position: cell.value for position, cell in sheet_from_rows(rows, "x").cells.items()}
|
||
|
||
|
||
def test_xls_without_codepage_is_file_warning(rules: Rules) -> None:
|
||
"""R-2: заметка читателя — предупреждение файла `XLS_NO_CODEPAGE` (info, без locator), кода OSV_ для неё нет."""
|
||
content = write_xls({"ОСВ 60": _xls_cells(mini_osv()), "ББ": _xls_cells(BALANCE_ROWS)}, codepage=None)
|
||
|
||
outcome = book(rules).parse(content, "XLS")
|
||
|
||
assert [(w.code, w.severity, w.locator) for w in file_warnings_of(outcome)] == [("XLS_NO_CODEPAGE", "info", None)]
|
||
assert file_warnings_of(outcome)[0].message == NO_CODEPAGE_WARNING
|
||
osv = [d for d in outcome.documents if isinstance(d, OsvDocument)]
|
||
assert osv and all("OSV_XLS_NO_CODEPAGE" not in {w.code for w in d.warnings} for d in osv)
|
||
assert_valid_result(outcome)
|
||
|
||
|
||
def test_xls_without_codepage_forms_only_book(rules: Rules) -> None:
|
||
content = write_xls({"ББ": _xls_cells(BALANCE_ROWS)}, codepage=None)
|
||
|
||
outcome = book(rules).parse(content, "XLS")
|
||
|
||
assert "XLS_NO_CODEPAGE" in file_codes_of(outcome)
|
||
|
||
|
||
def test_book_without_osv_and_forms_is_invalid_document(rules: Rules) -> None:
|
||
outcome = book(rules).parse(xlsx({"Справка": [["Наименование", "Сумма"], ["Аренда", 100]]}), "XLSX")
|
||
|
||
assert outcome.status is ResultStatus.ERROR
|
||
assert outcome.error is not None
|
||
assert outcome.error.code is ErrorCode.INVALID_DOCUMENT
|
||
assert "нет листов ОСВ или форм" in outcome.error.message
|
||
|
||
|
||
def test_documents_order_osv_then_forms(rules: Rules) -> None:
|
||
outcome = book(rules).parse(xlsx({"ББ": BALANCE_ROWS, "Лист_1": mini_osv()}), "XLSX")
|
||
|
||
assert [type(document).__name__ for document in outcome.documents] == ["OsvDocument", "StatementDocument"]
|
||
|
||
|
||
def test_consistent_forms_book_is_done(rules: Rules) -> None:
|
||
"""ББ и ОФР с сошедшимися соотношениями: оба документа `done`, предупреждений нет, книга `done`."""
|
||
outcome = book(rules).parse(xlsx({"ББ": GOOD_BALANCE_ROWS, "ОФР": GOOD_INCOME_ROWS}), "XLSX")
|
||
|
||
assert [(d.document_type, d.status) for d in outcome.documents] == [("BALANCE", "done"), ("INCOME", "done")]
|
||
assert outcome.status is ResultStatus.DONE
|
||
assert_valid_result(outcome)
|
||
|
||
|
||
def test_statement_error_warning_makes_book_needs_check(rules: Rules) -> None:
|
||
"""Книга `done` ⇔ все документы `done` и нет предупреждений файла `error` (спека «Статус книги»)."""
|
||
broken = [row[:] for row in GOOD_BALANCE_ROWS]
|
||
broken[-1][2] = 999 # 1600 ≠ 1100 + 1200 → жёсткое соотношение
|
||
|
||
outcome = book(rules).parse(xlsx({"ББ": broken}), "XLSX")
|
||
|
||
(document,) = outcome.documents
|
||
assert document.status == "needs_check"
|
||
assert outcome.status is ResultStatus.NEEDS_CHECK
|
||
|
||
|
||
def test_disclosures_only_book_is_done(rules: Rules) -> None:
|
||
"""Spec «Книга только из пояснений»: `DISCLOSURE` всегда `done`, статус книги от него не зависит."""
|
||
content = xlsx(
|
||
{
|
||
"0710003": [["Форма по ОКУД", "0710003"], ["Расшифровка"]],
|
||
"Пояснения": [["Пояснения к бухгалтерскому балансу и отчету о финансовых результатах"], ["Текст"]],
|
||
}
|
||
)
|
||
|
||
outcome = book(rules).parse(content, "XLSX")
|
||
|
||
assert [(d.document_type, d.status) for d in outcome.documents] == [("DISCLOSURE", "done")] * 2
|
||
assert outcome.status is ResultStatus.DONE
|
||
|
||
|
||
def test_disclosure_title_with_balance_codes_is_balance_with_type_conflict(rules: Rules) -> None:
|
||
rows: list[Cells] = [["Пояснения к бухгалтерскому балансу"], *BALANCE_ROWS[1:]]
|
||
|
||
outcome = book(rules).parse(xlsx({"Пояснения": rows}), "XLSX")
|
||
|
||
(document,) = outcome.documents
|
||
assert isinstance(document, StatementDocument)
|
||
assert document.document_type == "BALANCE"
|
||
conflicts = [w for w in document.warnings if w.code == "TYPE_CONFLICT"]
|
||
assert [w.severity for w in conflicts] == ["error"]
|
||
assert document.status == "needs_check"
|
||
|
||
|
||
def _xlsb_book(rows_by_sheet: dict[str, Sequence[Cells]]) -> bytes:
|
||
return write_xlsb({name: _xls_cells(rows) for name, rows in rows_by_sheet.items()})
|
||
|
||
|
||
def test_xlsb_form_scale_suspect(rules: Rules) -> None:
|
||
"""Книга xlsb, единица тыс. руб. и ≥ 40 % дробных значений: масштаб подозрителен (error) → `needs_check`."""
|
||
fractional = [row[:] for row in BALANCE_ROWS]
|
||
for row in fractional[4:]:
|
||
row[2] = Decimal(row[2]) + Decimal("0.5") # 100.5, 50.5, ...
|
||
fractional.insert(3, ["Единица измерения", "тыс. руб."])
|
||
|
||
outcome = book(rules).parse(_xlsb_book({"ББ": fractional}), "XLSB")
|
||
|
||
(document,) = outcome.documents
|
||
codes = {w.code: w.severity for w in document.warnings}
|
||
assert codes.get("XLSB_SCALE_SUSPECT") == "error", codes
|
||
assert document.status == "needs_check"
|
||
assert ("XLSB_NO_FORMATS", "info") in {(w.code, w.severity) for w in file_warnings_of(outcome)} | {
|
||
(w.code, w.severity) for w in document.warnings
|
||
}
|
||
|
||
|
||
def test_xlsb_form_without_fractions_has_only_no_formats(rules: Rules) -> None:
|
||
whole = [row[:] for row in GOOD_BALANCE_ROWS]
|
||
whole.insert(3, ["Единица измерения", "тыс. руб."])
|
||
whole.insert(3, ["ИНН", "7700000000"])
|
||
|
||
outcome = book(rules).parse(_xlsb_book({"ББ": whole}), "XLSB")
|
||
|
||
(document,) = outcome.documents
|
||
assert [w.code for w in document.warnings] == ["XLSB_NO_FORMATS"]
|
||
assert document.status == "done"
|
||
assert outcome.status is ResultStatus.DONE
|
||
|
||
|
||
def test_xlsb_book_marks_osv_with_outline_note_only(rules: Rules) -> None:
|
||
"""P4-13: заметка «outline недоступен» — в каждом документе ОСВ книги xlsb; формы её не получают."""
|
||
outcome = book(rules).parse(_xlsb_book({"ББ": BALANCE_ROWS, "ОСВ 60": mini_osv()}), "XLSB")
|
||
|
||
osv, form = outcome.documents
|
||
assert isinstance(osv, OsvDocument) and isinstance(form, StatementDocument)
|
||
assert "OSV_XLSB_NO_OUTLINE" in {w.code for w in osv.warnings}
|
||
assert "OSV_XLSB_NO_OUTLINE" not in {w.code for w in form.warnings}
|
||
|
||
|
||
def test_xlsx_book_has_no_xlsb_notes(rules: Rules) -> None:
|
||
outcome = book(rules).parse(xlsx({"ББ": BALANCE_ROWS, "ОСВ 60": mini_osv()}), "XLSX")
|
||
|
||
assert not any("XLSB" in w.code for document in outcome.documents for w in document.warnings)
|
||
assert outcome.metrics is not None and outcome.metrics.warnings == []
|
||
|
||
|
||
# --- N-2/N-3: лист с одной кодовой строкой — продолжение только сразу после листа формы ---
|
||
|
||
BALANCE_ROWS: list[Cells] = [
|
||
["Бухгалтерский баланс"],
|
||
["на 31 марта 2025 г."],
|
||
["Организация", "ООО Ромашка"],
|
||
["Наименование показателя", "Код", "На 31 марта 2025 г."],
|
||
["Основные средства", "1150", 100],
|
||
["Итого по разделу I", "1100", 100],
|
||
["Запасы", "1210", 50],
|
||
["Итого по разделу II", "1200", 50],
|
||
["БАЛАНС", "1600", 150],
|
||
]
|
||
GOOD_BALANCE_ROWS: list[Cells] = [
|
||
*BALANCE_ROWS,
|
||
["Нераспределенная прибыль", "1370", 150],
|
||
["Итого по разделу III", "1300", 150],
|
||
["БАЛАНС", "1700", 150],
|
||
]
|
||
GOOD_INCOME_ROWS: list[Cells] = [
|
||
["Отчет о финансовых результатах"],
|
||
["за 1 квартал 2025 г."],
|
||
["Организация", "ООО Ромашка"],
|
||
["Наименование показателя", "Код", "За 1 квартал 2025 г."],
|
||
["Выручка", "2110", 1000],
|
||
["Себестоимость продаж", "2120", -600],
|
||
["Валовая прибыль (убыток)", "2100", 400],
|
||
["Прибыль (убыток) от продаж", "2200", 400],
|
||
["Прибыль (убыток) до налогообложения", "2300", 400],
|
||
["Чистая прибыль (убыток)", "2400", 400],
|
||
]
|
||
INCOME_ROWS: list[Cells] = [
|
||
["Отчет о финансовых результатах"],
|
||
["за 1 квартал 2025 г."],
|
||
["Организация", "ООО Ромашка"],
|
||
["Наименование показателя", "Код", "За 1 квартал 2025 г."],
|
||
["Выручка", "2110", 1000],
|
||
["Себестоимость продаж", "2120", -600],
|
||
["Валовая прибыль (убыток)", "2100", 400],
|
||
]
|
||
|
||
|
||
def test_single_code_sheet_after_osv_sheet_is_not_a_continuation(rules: Rules) -> None:
|
||
"""Баланс, лист ОСВ, затем лист с одной строкой 1700: между формой и листом стоит ОСВ — лист «прочее»."""
|
||
data_ = xlsx({"ББ": BALANCE_ROWS, "Лист_1": mini_osv(), "Хвост": [["БАЛАНС", "1700", 150]]})
|
||
|
||
outcome = book(rules).parse(data_, "XLSX")
|
||
|
||
assert [(w.code, w.locator) for w in file_warnings_of(outcome)] == [("SHEET_OTHER", "sheet:Хвост")]
|
||
statements = [document for document in outcome.documents if not isinstance(document, OsvDocument)]
|
||
assert len(statements) == 1 and "1700" not in [line.line_code for line in statements[0].lines]
|
||
|
||
|
||
def test_lone_single_code_sheet_is_not_a_form(rules: Rules) -> None:
|
||
outcome = book(rules).parse(xlsx({"Справка": [["Основные средства", "1150", 100]]}), "XLSX")
|
||
|
||
assert outcome.status is ResultStatus.ERROR and outcome.error is not None
|
||
assert outcome.error.code is ErrorCode.INVALID_DOCUMENT
|
||
|
||
|
||
def test_single_code_sheet_right_after_form_joins(rules: Rules) -> None:
|
||
outcome = book(rules).parse(xlsx({"ББ": BALANCE_ROWS, "ББ_2": [["БАЛАНС", "1700", 150]]}), "XLSX")
|
||
|
||
(document,) = outcome.documents
|
||
assert not isinstance(document, OsvDocument)
|
||
assert document.locator == "sheet:ББ,ББ_2" and [line.line_code for line in document.lines][-1] == "1700"
|
||
|
||
|
||
def test_file_warnings_before_osv_error_are_kept(rules: Rules) -> None:
|
||
"""Ошибка листа ОСВ: ответ несёт предупреждения файла, собранные до неё (P4-14); «форма не разобрана» не пишется."""
|
||
broken: list[Cells] = [['АО "Ромашка"'], [TITLE], [], ["Счет", "Сумма"], ["60", 100]]
|
||
notes: list[Cells] = [["Примечания"], ["Текст без кодов и сумм"]]
|
||
|
||
outcome = book(rules).parse(xlsx({"Примечания": notes, "ББ": BALANCE_ROWS, "ОСВ 60": broken}), "XLSX")
|
||
|
||
assert outcome.status is ResultStatus.ERROR
|
||
assert [(w.code, w.locator) for w in file_warnings_of(outcome)] == [("SHEET_OTHER", "sheet:Примечания")]
|
||
assert not any("форма не разобрана" in w.message for w in file_warnings_of(outcome))
|
||
|
||
|
||
def test_file_warnings_follow_sheet_order(rules: Rules) -> None:
|
||
"""Предупреждения идут по порядку листов: «Справка» (до ББ) и «Хвост» (после ОФР) не уезжают относительно форм."""
|
||
notes: list[Cells] = [["Примечания"], ["Текст без кодов и сумм"]]
|
||
tail: list[Cells] = [["Хвост"], ["Текст без кодов и сумм"]]
|
||
unjoined = [["БАЛАНС", "1700", 150]] # одна кодовая строка после листа «прочее» — не продолжение
|
||
content = xlsx(
|
||
{
|
||
"Справка": notes,
|
||
"ББ": BALANCE_ROWS,
|
||
"Лист_1": mini_osv(),
|
||
"Одиночка": unjoined,
|
||
"ОФР": INCOME_ROWS,
|
||
"Хвост": tail,
|
||
}
|
||
)
|
||
|
||
outcome = book(rules).parse(content, "XLS") # xlsx под видом XLS → FILE_TYPE_MISMATCH первым
|
||
|
||
assert [(w.code, w.locator) for w in file_warnings_of(outcome)] == [
|
||
("FILE_TYPE_MISMATCH", None),
|
||
("SHEET_OTHER", "sheet:Справка"),
|
||
("SHEET_OTHER", "sheet:Одиночка"),
|
||
("SHEET_OTHER", "sheet:Хвост"),
|
||
]
|