126 lines
6.0 KiB
Python
126 lines
6.0 KiB
Python
"""Review Focus плана 5: правила пакета arm_statements на реальных ответах ИС (`fixtures/ocr`).
|
||
|
||
Утверждения не ослабляются и не подгоняются под файл (Р-31/P5-17): расхождение — находка для пакета."""
|
||
|
||
from decimal import Decimal
|
||
from typing import Any
|
||
|
||
import pytest
|
||
from arm_statements import default_rules
|
||
|
||
from src.contract import ErrorCode, ResultStatus
|
||
from src.contract.result import StatementDocument, StatementLine, StatementValue
|
||
from src.normalizer import NormalizeOutcome, OcrNormalizer
|
||
from tests.golden_update import JOB, REAL_FIXTURES, real_set
|
||
|
||
pytestmark = [
|
||
pytest.mark.fixtures,
|
||
pytest.mark.skipif(not REAL_FIXTURES.is_dir(), reason="fixtures недоступен: положите репозиторий fixtures рядом"),
|
||
]
|
||
|
||
NORMALIZER = OcrNormalizer(default_rules(), version="0" * 40)
|
||
|
||
|
||
def run(name: str) -> NormalizeOutcome:
|
||
return NORMALIZER.normalize(real_set(name))
|
||
|
||
|
||
def lines(doc: StatementDocument) -> dict[str, StatementLine]:
|
||
refs = [ln.ref for ln in doc.lines]
|
||
assert len(refs) == len(set(refs)) and [ln.row_index for ln in doc.lines] == list(range(len(doc.lines)))
|
||
return {ln.ref: ln for ln in doc.lines}
|
||
|
||
|
||
def value(line: StatementLine, column_index: int) -> StatementValue:
|
||
return next(v for v in line.values if v.column_index == column_index)
|
||
|
||
|
||
def _holders(out: NormalizeOutcome) -> list[Any]:
|
||
"""Что удерживает `done`: значения и строки с needs_review, строки не ok/detail, предупреждения error."""
|
||
found: list[Any] = []
|
||
for doc in out.documents:
|
||
for ln in doc.lines:
|
||
if ln.line_status not in ("ok", "detail"):
|
||
found.append((doc.locator, ln.ref, "line_status", ln.line_status, ln.line_code_text))
|
||
for v in ln.values:
|
||
if v.needs_review:
|
||
found.append((doc.locator, ln.ref, v.column_index, v.text, v.confidence))
|
||
if ln.needs_review:
|
||
found.append((doc.locator, ln.ref, "line", ln.line_code_text, None))
|
||
found.extend(("warning", w.code, w.severity, w.locator, w.message) for w in out.warnings if w.severity == "error")
|
||
return found
|
||
|
||
|
||
def test_aep_3_junk_codes_and_dirty_amounts() -> None: # Review Focus 1, 4
|
||
out = run("aep_3")
|
||
assert out.status is ResultStatus.NEEDS_CHECK
|
||
bal, inc = out.documents
|
||
assert [(d.document_type, d.locator, d.status) for d in out.documents] == [
|
||
("BALANCE", f"file:{JOB}.json;table:0", "needs_check"),
|
||
("INCOME", f"file:{JOB}.json;table:1", "needs_check"),
|
||
]
|
||
assert [c.column_index for c in bal.columns] == [2, 4, 5]
|
||
b = lines(bal)
|
||
assert (b["row:0"].line_status, b["row:0"].line_code_text) == ("missing", "НИ")
|
||
assert (b["row:2"].line_status, b["row:2"].parent_ref) == ("detail", "row:1")
|
||
assert (b["row:3"].line_status, b["row:3"].line_code) == ("ok", "1140")
|
||
assert {b["row:12"].line_status, b["row:31"].line_status} == {"unreadable"}
|
||
assert (b["row:17"].line_status, b["row:17"].parent_ref) == ("detail", "row:16")
|
||
i = lines(inc)
|
||
revenue = value(i["row:0"], 2)
|
||
assert i["row:0"].line_status == "missing"
|
||
assert (revenue.amount, revenue.string_value, revenue.is_recognized) == (None, "211(1 |", False)
|
||
cost = value(i["row:1"], 2)
|
||
assert (i["row:1"].line_code, cost.amount, cost.is_recognized) == ("2120", Decimal("-7744536"), False)
|
||
assert value(i["row:8"], 2).amount is None
|
||
[failed] = [w for w in out.warnings if w.code == "JSON_FAILED"]
|
||
assert (failed.severity, failed.locator) == ("error", f"file:{JOB}_ofr.json")
|
||
assert failed.message == "INVALID_DOCUMENT: Не удалось определить тип документа."
|
||
|
||
|
||
def test_aep_6_garbage_code_and_repeats() -> None: # Review Focus 2
|
||
bal, inc = run("aep_6").documents
|
||
b = lines(bal)
|
||
assert (b["row:6"].line_status, b["row:6"].line_code, b["row:6"].line_code_text) == (
|
||
"unreadable",
|
||
None,
|
||
"44 30\nI |",
|
||
)
|
||
assert "row:6#1" not in b and all(v.amount is None for v in b["row:6"].values)
|
||
assert [ln.line_code for ln in bal.lines].count("1220") == 2
|
||
assert [ln.line_code for ln in bal.lines].count("1520") == 2
|
||
i = lines(inc)
|
||
assert [ln.line_code for ln in inc.lines].count("2410") == 2 and i["row:4"].line_status == "detail"
|
||
assert bal.status == "needs_check"
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("name", "row", "amounts"),
|
||
[
|
||
("БО 1 кв 2025", 73, (Decimal("43274"), Decimal("198622"))),
|
||
("Бухгалтерская отчетность 1 кв.25", 93, (Decimal("917775"), Decimal("2371333"))),
|
||
],
|
||
)
|
||
def test_q1_2025_split_repeats_and_done(name: str, row: int, amounts: tuple[Decimal, Decimal]) -> None: # RF 3
|
||
out = run(name)
|
||
bal, _ = out.documents
|
||
b = lines(bal)
|
||
first, second = b[f"row:{row}#1"], b[f"row:{row}#2"]
|
||
assert (first.line_code, second.line_code) == ("1500", "1700")
|
||
assert (value(first, 2).amount, value(second, 2).amount) == amounts
|
||
assert any(w.code == "ROW_SPLIT" and w.severity == "info" for w in bal.warnings)
|
||
codes = [ln.line_code for ln in bal.lines]
|
||
assert codes.count("1220") == 2 and codes.count("1520") == 2
|
||
# Р-23: расшифровки не держат документ; расхождение — находка для пакета (P5-17), не правка шлюза
|
||
assert [d.status for d in out.documents] == ["done", "done"], _holders(out)
|
||
assert out.status is ResultStatus.DONE
|
||
|
||
|
||
@pytest.mark.parametrize("name", ["aep_5", "aep_8_form4"])
|
||
def test_all_json_failed_is_error(name: str) -> None: # Review Focus 4, 5
|
||
out = run(name)
|
||
assert (out.status, out.documents) == (ResultStatus.ERROR, [])
|
||
assert out.error is not None and out.error.code is ErrorCode.INVALID_DOCUMENT
|
||
assert out.error.message == "Не удалось определить тип документа."
|
||
assert [(w.code, w.locator) for w in out.warnings] == [("JSON_FAILED", f"file:{JOB}.json")]
|