ae2922eb30
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W
110 lines
5.7 KiB
Python
110 lines
5.7 KiB
Python
"""`OcrNormalizer`: набор json ИС → `NormalizeOutcome` (статус, ошибка набора, предупреждения файла, метрики, документы
|
||
контракта 2.0). Реальные ответы ИС — `test_golden.py` / `test_real_answers.py` (задача 6)."""
|
||
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
import src.normalizer.ocr as ocr_module
|
||
import src.normalizer.ocr.answer_set as answer_set_module
|
||
from arm_statements import default_rules
|
||
from src.contract import ErrorCode, FileWarning, ResultStatus, Severity, StatementDocument
|
||
from src.normalizer import OcrNormalizer
|
||
from src.normalizer.ocr.answer_set import NO_TABLES_MESSAGE
|
||
from tests.unit.normalizer.ocr_builders import BAD, BALANCE, HEAD, INCOME, OK, merged, ocr_table, response
|
||
from tests.worker_helpers import SUCCESS_ANSWER
|
||
|
||
VERSION = "abc1234"
|
||
|
||
|
||
def normalizer() -> OcrNormalizer:
|
||
return OcrNormalizer(default_rules(), version=VERSION)
|
||
|
||
|
||
def two_forms() -> dict[str, Any]:
|
||
return response(ocr_table("Бухгалтерский баланс", BALANCE), ocr_table("Отчет о финансовых результатах", INCOME))
|
||
|
||
|
||
def test_done_with_documents_metrics_and_version() -> None:
|
||
outcome = normalizer().normalize(merged(("j.json", response(ocr_table("Бухгалтерский баланс", BALANCE)))))
|
||
assert outcome.status is ResultStatus.DONE and outcome.error is None and outcome.warnings == []
|
||
assert [d.document_type for d in outcome.documents] == ["BALANCE"]
|
||
assert all(isinstance(d, StatementDocument) for d in outcome.documents)
|
||
assert outcome.metrics is not None
|
||
assert outcome.metrics.rows == sum(len(d.lines) for d in outcome.documents) > 0
|
||
assert outcome.statements_version == VERSION
|
||
|
||
|
||
def test_one_document_needs_check_makes_set_needs_check() -> None:
|
||
"""Синтетический ОФР без строки 2300 не сходится по §7: документ needs_check, набор — тоже."""
|
||
outcome = normalizer().normalize(merged(("j.json", two_forms())))
|
||
assert [d.document_type for d in outcome.documents] == ["BALANCE", "INCOME"]
|
||
assert [d.status for d in outcome.documents] == ["done", "needs_check"]
|
||
assert outcome.status is ResultStatus.NEEDS_CHECK and outcome.warnings == []
|
||
|
||
|
||
def test_lost_json_keeps_documents_and_gives_needs_check() -> None:
|
||
outcome = normalizer().normalize(
|
||
merged(("j.json", response(ocr_table("Бухгалтерский баланс", BALANCE))), ("j_ofr.json", response(error=BAD)))
|
||
)
|
||
assert outcome.status is ResultStatus.NEEDS_CHECK and len(outcome.documents) == 1
|
||
[warning] = outcome.warnings
|
||
assert isinstance(warning, FileWarning)
|
||
assert (warning.code, warning.severity, warning.locator) == ("JSON_FAILED", Severity.ERROR, "file:j_ofr.json")
|
||
assert warning.message == "INVALID_DOCUMENT: Не удалось определить тип документа."
|
||
|
||
|
||
def test_all_json_failed_takes_first_code() -> None:
|
||
outcome = normalizer().normalize(merged(("j.json", response(error=BAD)), ("j_ofr.json", "не объект")))
|
||
assert outcome.status is ResultStatus.ERROR and outcome.error is not None
|
||
assert (outcome.error.code, outcome.error.message) == (ErrorCode.INVALID_DOCUMENT, BAD["message"])
|
||
assert outcome.documents == [] and outcome.metrics is None and outcome.statements_version is None
|
||
assert [w.locator for w in outcome.warnings] == ["file:j.json", "file:j_ofr.json"]
|
||
|
||
|
||
def test_no_tables_is_recognition_failed() -> None:
|
||
outcome = normalizer().normalize(merged(("j.json", {"error": OK, "data_ocr": {**HEAD, "table_structure": []}})))
|
||
assert outcome.status is ResultStatus.ERROR and outcome.error is not None
|
||
assert (outcome.error.code, outcome.error.message) == (ErrorCode.RECOGNITION_FAILED, NO_TABLES_MESSAGE)
|
||
assert [(w.code, w.message) for w in outcome.warnings] == [("JSON_FAILED", "в ответе ИС нет таблиц")]
|
||
|
||
|
||
def test_table_exception_becomes_table_parse_error(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
real = answer_set_module.normalize_candidate
|
||
|
||
def flaky(candidate: Any, rules: Any = None) -> Any:
|
||
if candidate.locator.endswith("table:1"):
|
||
raise RuntimeError("сломалось")
|
||
return real(candidate, rules)
|
||
|
||
monkeypatch.setattr(answer_set_module, "normalize_candidate", flaky)
|
||
outcome = normalizer().normalize(merged(("j.json", two_forms())))
|
||
assert outcome.status is ResultStatus.NEEDS_CHECK and len(outcome.documents) == 1
|
||
[warning] = outcome.warnings
|
||
assert (warning.code, warning.locator) == ("TABLE_PARSE_ERROR", "file:j.json;table:1")
|
||
assert warning.message == "BACKEND_PARSING_ERROR: RuntimeError: сломалось"
|
||
|
||
|
||
@pytest.mark.parametrize("raw", [None, {"job_id": "j"}, {"job_id": "j", "files": []}])
|
||
def test_not_merged_structure_is_invalid_format(raw: object) -> None:
|
||
outcome = normalizer().normalize(raw)
|
||
assert outcome.status is ResultStatus.ERROR and outcome.error is not None
|
||
assert outcome.error.code is ErrorCode.INVALID_FORMAT and outcome.warnings == []
|
||
|
||
|
||
def test_success_answer_helper_from_worker_tests() -> None:
|
||
outcome = normalizer().normalize(merged(("j.json", SUCCESS_ANSWER)))
|
||
assert outcome.status is not ResultStatus.ERROR and len(outcome.documents) == 1
|
||
|
||
|
||
def test_default_rules_and_version_from_package() -> None:
|
||
created = OcrNormalizer()
|
||
assert created.statements_version == ocr_module.statements_version()
|
||
|
||
|
||
def test_requires_statements_version(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.setattr(ocr_module, "statements_version", lambda path=None: None)
|
||
with pytest.raises(ValueError, match="без VERSION"):
|
||
OcrNormalizer(default_rules())
|
||
assert OcrNormalizer(default_rules(), version="x").statements_version == "x"
|