028512603a
T2.1: аудит lines[] включает итоговую смену parent_row_index после пересчёта. T1.1: тест паритета редакции приёма и пересчёта для строки с confidence < 100 и code_fixed. T1.2: edition_codes пропускает строку ok без кода. T2.4: общий issue_column_index для ответа формы и detail отказа подтверждения. T2.6: удалено дублирование needs_review в statement_line_edits. T2.8: неизвестный relative:* не даёт 500 (адаптер stored_form и _resolve_relative_columns). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
173 lines
8.8 KiB
Python
173 lines
8.8 KiB
Python
"""Адаптер `StatementsEngine` поверх реального пакета `arm_statements` (часть Б, B1)."""
|
||
|
||
import datetime
|
||
import json
|
||
from decimal import Decimal
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from arm_statements import DocStatus, DocumentType, Edition, FormVariant, PeriodKind, default_rules, statements_version
|
||
from src.contract import StatementDocumentV2
|
||
from src.domain.stored_form import build_stored_form, edition_codes
|
||
from src.models.enums import StatementLineStatus
|
||
from src.services.statements_engine import ArmStatementsEngine, default_engine
|
||
from tests.fake_engine import FakeEngine, empty_result
|
||
from tests.test_domain_stored_form import make_column, make_document, make_line, make_meta, make_value
|
||
|
||
EXAMPLES = Path(__file__).resolve().parent.parent / "contracts" / "examples"
|
||
ENGINE = default_engine()
|
||
|
||
|
||
def _documents(name: str) -> list[dict[str, object]]:
|
||
raw = json.loads((EXAMPLES / name).read_text(encoding="utf-8"))
|
||
return [d for d in raw["documents"] if d["document_type"] in ("BALANCE", "INCOME", "DISCLOSURE")]
|
||
|
||
|
||
@pytest.mark.parametrize("name", ["result-statement-minimal.json", "result-statement-excel.json"])
|
||
def test_contract_document_status_matches_examples(name: str) -> None:
|
||
documents = _documents(name)
|
||
assert documents
|
||
for doc in documents:
|
||
assert ENGINE.contract_document_status(doc) == DocStatus(doc["status"])
|
||
if name == "result-statement-minimal.json":
|
||
assert ENGINE.contract_document_status(documents[0]) is DocStatus.NEEDS_CHECK # строка unreadable
|
||
|
||
|
||
def test_edition_codes_and_engine_edition() -> None:
|
||
doc = StatementDocumentV2.model_validate(_documents("result-statement-minimal.json")[0])
|
||
pairs = edition_codes(doc)
|
||
effective = datetime.date(2025, 9, 30)
|
||
edition = ENGINE.edition(DocumentType.BALANCE, pairs, effective, FormVariant.FULL)
|
||
assert isinstance(edition, Edition)
|
||
|
||
# та же форма, собранная из этого результата в строки БД: редакция пересчёта совпадает с редакцией приёма
|
||
column = make_column()
|
||
lines = []
|
||
for ln in doc.lines:
|
||
line = make_line(
|
||
ln.line_code,
|
||
row_index=ln.row_index,
|
||
status=StatementLineStatus(ln.line_status),
|
||
confidence=None if ln.line_code_confidence is None else str(ln.line_code_confidence),
|
||
)
|
||
lines.append((line, [make_value(line, column, v.amount, confidence=str(v.confidence)) for v in ln.values]))
|
||
form = build_stored_form(make_document(effective), make_meta(organization_name="ООО"), [column], lines, [])
|
||
assert ENGINE.recompute(form).edition == edition
|
||
|
||
|
||
@pytest.mark.parametrize(("modified", "confirmed"), [(False, False), (True, False), (False, True)])
|
||
def test_edition_parity_for_unsure_fixed_code(modified: bool, confirmed: bool) -> None:
|
||
"""T1.1: строка `ok` с `line_code_confidence < 100` и `code_fixed` (назначенная / подтверждённая) — редакция
|
||
приёма (`edition_codes` + `engine.edition`) совпадает с редакцией пересчёта из БД."""
|
||
doc = StatementDocumentV2.model_validate(_documents("result-statement-minimal.json")[0])
|
||
first_ok = next(ln for ln in doc.lines if ln.line_status == "ok")
|
||
doc = doc.model_copy(
|
||
update={
|
||
"lines": [
|
||
ln.model_copy(update={"line_code_confidence": Decimal(50)}) if ln is first_ok else ln
|
||
for ln in doc.lines
|
||
]
|
||
}
|
||
)
|
||
unsure = next(ln for ln in doc.lines if ln.line_status == "ok")
|
||
effective = datetime.date(2025, 9, 30)
|
||
edition = ENGINE.edition(DocumentType.BALANCE, edition_codes(doc), effective, FormVariant.FULL)
|
||
|
||
column = make_column()
|
||
lines = []
|
||
for ln in doc.lines:
|
||
fixed = ln is unsure
|
||
line = make_line(
|
||
ln.line_code,
|
||
row_index=ln.row_index,
|
||
status=StatementLineStatus(ln.line_status),
|
||
confidence=None if ln.line_code_confidence is None else str(ln.line_code_confidence),
|
||
modified=modified and fixed,
|
||
confirmed=confirmed and fixed,
|
||
)
|
||
lines.append((line, [make_value(line, column, v.amount, confidence=str(v.confidence)) for v in ln.values]))
|
||
form = build_stored_form(make_document(effective), make_meta(organization_name="ООО"), [column], lines, [])
|
||
assert ENGINE.recompute(form).edition == edition
|
||
|
||
|
||
def test_edition_codes_skip_ok_line_without_code() -> None:
|
||
"""T1.2: строка `ok` без кода не даёт пару `("None", …)`."""
|
||
doc = StatementDocumentV2.model_validate(_documents("result-statement-minimal.json")[0])
|
||
ok_line = next(ln for ln in doc.lines if ln.line_status == "ok")
|
||
count = len(edition_codes(doc))
|
||
doc = doc.model_copy(
|
||
update={"lines": [ln.model_copy(update={"line_code": None}) if ln is ok_line else ln for ln in doc.lines]}
|
||
)
|
||
pairs = edition_codes(doc)
|
||
assert len(pairs) == count - 1
|
||
assert all(code != "None" for code, _ in pairs)
|
||
|
||
|
||
def test_engine_scenario_recompute_after_edit() -> None:
|
||
column = make_column()
|
||
effective = datetime.date(2025, 9, 30)
|
||
l1210 = make_line("1210", row_index=0)
|
||
l1230 = make_line("1230", row_index=1)
|
||
l1250 = make_line("1250", row_index=2, confidence="25", modified=True) # бывшая unreadable с назначенным кодом
|
||
l1260 = make_line("1260", row_index=None, confidence=None, modified=None) # шаблонная строка
|
||
l1200 = make_line("1200", row_index=3)
|
||
detail = make_line(None, row_index=4, status=StatementLineStatus.detail, parent_row_index=3)
|
||
pairs = [
|
||
(l1210, [make_value(l1210, column, "100")]),
|
||
(l1230, [make_value(l1230, column, "428311", confidence="60", modified=True)]),
|
||
(l1250, [make_value(l1250, column, "50", confidence="67", modified=True)]),
|
||
(l1260, [make_value(l1260, column, "10", confidence=None, recognized=None, modified=True)]),
|
||
(l1200, [make_value(l1200, column, "428471")]),
|
||
(detail, [make_value(detail, column, "5", confidence="60")]),
|
||
]
|
||
form = build_stored_form(make_document(effective), make_meta(organization_name="ООО"), [column], pairs, [])
|
||
result = ENGINE.recompute(form)
|
||
|
||
assert result.edition is Edition.FSBU4
|
||
assert result.form_variant is FormVariant.FULL
|
||
key = str(column.id)
|
||
assert result.value_needs_review[(str(l1230.id), key)] is False
|
||
assert result.value_needs_review[(str(detail.id), key)] is False
|
||
assert result.line_needs_review[str(l1250.id)] is False
|
||
assert result.detail_parents[str(detail.id)] == str(l1200.id) # ближайшая выше строка ok
|
||
# 1200 = … + 1250 + 1260 сошлось только с учётом строки 1250 и шаблонной 1260
|
||
assert [f for f in result.checks.relations_failed if f.relation.startswith("1200")] == []
|
||
|
||
# без шаблонной строки 1260 сумма не сходится: соотношение нарушено с ключом колонки БД
|
||
without_template = build_stored_form(
|
||
make_document(effective),
|
||
make_meta(organization_name="ООО"),
|
||
[column],
|
||
[p for p in pairs if p[0] is not l1260],
|
||
[],
|
||
)
|
||
failed = [f for f in ENGINE.recompute(without_template).checks.relations_failed if f.relation.startswith("1200")]
|
||
assert [(f.column_key, f.expected, f.actual) for f in failed] == [(key, Decimal(428471), Decimal(428461))]
|
||
|
||
|
||
def test_engine_version_reads_file(tmp_path: Path) -> None:
|
||
assert ENGINE.version == statements_version()
|
||
assert statements_version(tmp_path / "VERSION") is None
|
||
|
||
|
||
def test_engine_rules_accessors() -> None:
|
||
engine = ArmStatementsEngine(default_rules())
|
||
assert engine.tolerance() == Decimal(1)
|
||
assert engine.editions() == (Edition.RAS_66N, Edition.FSBU4)
|
||
assert engine.code_order(Edition.FSBU4, DocumentType.DISCLOSURE, FormVariant.FULL) == ()
|
||
assert "1200" in engine.code_order(Edition.FSBU4, DocumentType.BALANCE, FormVariant.FULL)
|
||
assert engine.resolve_relative(PeriodKind.ABSOLUTE, datetime.date(2025, 9, 30)) is None
|
||
assert engine.resolve_relative(PeriodKind.NONE, datetime.date(2025, 9, 30)) is None
|
||
assert engine.resolve_relative(PeriodKind.PREV_YEAR_END, datetime.date(2025, 9, 30)) == datetime.date(2024, 12, 31)
|
||
|
||
|
||
def test_fake_engine_records_form() -> None:
|
||
fake = FakeEngine(empty_result(DocStatus.NEEDS_CHECK), document_status=DocStatus.NEEDS_CHECK, edition=None)
|
||
form = build_stored_form(make_document(), make_meta(), [], [], [])
|
||
assert fake.recompute(form).status is DocStatus.NEEDS_CHECK
|
||
assert fake.forms == [form]
|
||
assert fake.edition(DocumentType.BALANCE, [("1200", True)], None, None) is None
|
||
assert fake.edition_calls[0][1] == [("1200", True)]
|
||
assert fake.contract_document_status({}) is DocStatus.NEEDS_CHECK
|