arm-statements/arm_statements/tests/test_loader.py

import re
from decimal import Decimal
from pathlib import Path

import pytest
import yaml

from arm_statements.issues import WarningCode, make_warning
from arm_statements.loader import LABEL_KEYS, Rules, RulesError, load_rules
from arm_statements.model import DocumentType, Edition, FormVariant, Severity

RULES_DIR = Path(__file__).parent.parent / "rules"

SPEC_WARNINGS = {
    "JSON_FAILED": ("error", "file"),
    "TABLE_PARSE_ERROR": ("error", "file"),
    "TABLE_LOST": ("error", "file"),
    "TABLE_SKIPPED": ("info", "file"),
    "CODE_COLUMN_NOT_FOUND": ("error", "document"),
    "SEGMENT_WITHOUT_CODES": ("error", "document"),
    "SHEET_NOT_JOINED": ("error", "document"),
    "CONTINUATION_TYPE_CONFLICT": ("error", "document"),
    "SEGMENT_COLUMNS_MISMATCH": ("error", "document"),
    "COLUMN_WITHOUT_DATE": ("error", "document"),
    "COLUMN_DATE_UNCERTAIN": ("error", "document"),
    "HEADER_ROW_NOT_FOUND": ("error", "document"),
    "TYPE_CONFLICT": ("error", "document"),
    "NO_ORGANIZATION_NO_INN": ("error", "document"),
    "PERIOD_UNKNOWN": ("error", "document"),
    "REPORT_DATE_UNCERTAIN": ("error", "document"),
    "XLSB_SCALE_SUSPECT": ("error", "document"),
    "INN_NOT_FOUND": ("info", "document"),
    "INN_UNCERTAIN": ("info", "document"),
    "ORGANIZATION_NOT_FOUND": ("info", "document"),
    "ORGANIZATION_UNCERTAIN": ("info", "document"),
    "REPORT_DATE_UNPARSED": ("info", "document"),
    "UNITS_NOT_FOUND": ("info", "document"),
    "FILE_TYPE_MISMATCH": ("info", "file"),
    "COLUMN_SKIPPED": ("info", "document"),
    "FRACTIONAL_IN_THOUSANDS": ("info", "document"),
    "XLSB_NO_FORMATS": ("info", "document"),
    "NUMBER_FORMAT_UNPARSED": ("info", "document"),
    "VALUE_OUT_OF_RANGE": ("info", "document"),
    "SHEET_OTHER": ("info", "file"),
    "ROW_SPLIT": ("info", "document"),
    "NUMBERS_OUTSIDE_TABLE": ("info", "document"),
    "XLS_NO_CODEPAGE": ("info", "file"),
}


def test_warning_table_matches_spec(rules: Rules) -> None:
    assert {c: (w.severity.value, w.scope) for c, w in rules.warnings.items()} == SPEC_WARNINGS


def test_simplified_income_relation_is_soft(rules: Rules) -> None:
    (rule,) = rules.checks.relations_for(DocumentType.INCOME, FormVariant.SIMPLIFIED)
    assert rule.target == "2400" and not rule.strict
    assert rule.abs_terms == frozenset({"2120", "2330", "2350", "2410"})
    assert rules.checks.abs_codes(DocumentType.INCOME, FormVariant.FULL) == frozenset(
        {"2120", "2210", "2220", "2330", "2350", "2410"}
    )


def test_code_orders_with_repeats(rules: Rules) -> None:
    fsbu = rules.code_orders.sequence(Edition.FSBU4, DocumentType.BALANCE, FormVariant.FULL)
    assert fsbu.count("1220") == len(("1220", "1220")) and "1120" not in fsbu
    ras = rules.code_orders.sequence(Edition.RAS_66N, DocumentType.BALANCE, FormVariant.FULL)
    assert "1120" in ras and ras.index("1110") < ras.index("1100")
    assert rules.code_orders.editions == (Edition.RAS_66N, Edition.FSBU4)
    assert rules.code_orders.sequence(Edition.FSBU4, DocumentType.DISCLOSURE, FormVariant.FULL) == ()


def test_public_rule_types(rules: Rules) -> None:
    assert isinstance(rules.checks.tolerance, Decimal) and rules.checks.tolerance == Decimal("1")
    assert all(isinstance(e, Edition) for e in rules.code_orders.editions)
    fsbu = rules.code_orders.sequence(Edition.FSBU4, DocumentType.INCOME, FormVariant.SIMPLIFIED)
    assert isinstance(fsbu, tuple) and all(isinstance(c, str) for c in fsbu)
    assert (rules.code_ranges.min_codes_continuation, rules.code_ranges.min_codes) == (1, 3)
    assert tuple(rules.headers.labels) == LABEL_KEYS


def test_warning_code_enum_matches_dictionary(rules: Rules) -> None:
    assert {c.value for c in WarningCode} == set(rules.warnings) == set(SPEC_WARNINGS)
    assert WarningCode.XLS_NO_CODEPAGE == "XLS_NO_CODEPAGE"


def test_no_code_to_name_mapping_anywhere() -> None:
    code = re.compile(r"^\d{4,5}$")

    def walk(node: object) -> None:
        if isinstance(node, dict):
            for key, value in node.items():
                assert not (code.match(str(key)) and isinstance(value, str) and re.search(r"[А-Яа-яЁё]", value))
                walk(value)
        elif isinstance(node, list):
            for item in node:
                walk(item)

    for path in RULES_DIR.glob("*.yaml"):
        walk(yaml.safe_load(path.read_text(encoding="utf-8")))


def test_broken_repeat_count_rejected(tmp_path: Path) -> None:
    for path in RULES_DIR.glob("*.yaml"):
        (tmp_path / path.name).write_bytes(path.read_bytes())
    orders = tmp_path / "statement_code_orders.yaml"
    orders.write_text(
        orders.read_text(encoding="utf-8").replace('- "1520"   # «Обязательство', '# "1520"'), encoding="utf-8"
    )
    with pytest.raises(RulesError, match="statement_code_orders"):
        load_rules(tmp_path)


def test_unknown_severity_rejected(tmp_path: Path) -> None:
    for path in RULES_DIR.glob("*.yaml"):
        (tmp_path / path.name).write_bytes(path.read_bytes())
    target = tmp_path / "statement_warnings.yaml"
    target.write_text(
        target.read_text(encoding="utf-8").replace("severity: info", "severity: warn", 1), encoding="utf-8"
    )
    with pytest.raises(RulesError, match="statement_warnings"):
        load_rules(tmp_path)


def test_make_warning_takes_severity_from_dictionary(rules: Rules) -> None:
    warning = make_warning(WarningCode.COLUMN_SKIPPED, "колонка «Изменение, %» пропущена", rules, column_index=5)
    assert (warning.code, warning.severity, warning.column_index) == ("COLUMN_SKIPPED", Severity.INFO, 5)
    assert make_warning("JSON_FAILED", "x", rules).severity is Severity.ERROR


def test_make_warning_unknown_code_raises(rules: Rules) -> None:
    with pytest.raises(KeyError, match="NOT_A_CODE"):
        make_warning("NOT_A_CODE", "x", rules)


def test_label_keys_are_checked(tmp_path: Path) -> None:
    for path in RULES_DIR.glob("*.yaml"):
        (tmp_path / path.name).write_bytes(path.read_bytes())
    target = tmp_path / "statement_headers.yaml"
    target.write_text(target.read_text(encoding="utf-8").replace("  okpo:", "  okpo_x:", 1), encoding="utf-8")
    with pytest.raises(RulesError, match="statement_headers"):
        load_rules(tmp_path)


def test_absolute_period_kinds_come_from_dictionary(rules: Rules) -> None:
    expected = [
        "range_end",
        "day_month_year",
        "quarter",
        "half",
        "months",
        "month_range",
        "year",
    ]
    data = yaml.safe_load((RULES_DIR / "statement_periods.yaml").read_text(encoding="utf-8"))
    assert [p["kind"] for p in data["absolute"]] == expected
    assert [p.kind for p in rules.periods.absolute] == expected


def test_repeats_for_unknown_form_rejected(tmp_path: Path) -> None:
    for path in RULES_DIR.glob("*.yaml"):
        (tmp_path / path.name).write_bytes(path.read_bytes())
    orders = tmp_path / "statement_code_orders.yaml"
    orders.write_text(
        orders.read_text(encoding="utf-8").replace("    INCOME: []", "    DISCLOSURE: []", 1), encoding="utf-8"
    )
    with pytest.raises(RulesError, match="repeats"):
        load_rules(tmp_path)