c780dfd858
- OQ-138 (решение владельца 06.10.2026): единицы шапки верны; жёсткое расхождение до rounding_tolerance_percent = 0,001 % от наибольшего по модулю числа соотношения — округление итогов: soft_relations_failed + предупреждение, статус не меняется. НПЗ, ОФР стр. 1 Унагранде Компани и Групп теперь done. - N-1/N-2: лист с 1–2 кодами — продолжение только сразу после листа формы, коды из диапазона формы открытого документа; присоединение видно в предупреждениях. N-3: негативные тесты BookParser и StatementParser. - M-9: пятизначная подстрока относится к ближайшей четырёхзначной строке выше с теми же тремя цифрами (12310 под 1230). - N-4: на продолжении без шапки колонка «Пояснения» узнаётся по коротким значениям и не даёт line_name. - T7.1: ответ с ошибкой листа ОСВ называет неразобранные листы форм; T7.2: metrics.rows для форм — в спеке. make check-full: 2217 passed, интеграция 10 passed; golden 138/138 (снимки — arm-ka-fixtures fix/20-golden). Refs #20 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EAKRDNzXtRrwmWz9z6PZu3
356 lines
19 KiB
Python
356 lines
19 KiB
Python
"""Интеграционный round-trip сервиса против compose-NATS (без JetStream) и compose-MinIO (`.env.test`).
|
||
|
||
Приложение поднимается в процессе теста через lifespan (`src.app`) с `NATS_ENABLED=True` — тот же код, что в проде:
|
||
`_connect_s3`, `NatsConnection` с подпиской в queue group `tabular-parser`, `RecognitionWorker` с парсером по умолчанию
|
||
(`get_parser(settings)` → `OsvParser`); исходник — битый zip, поэтому результат — ошибка `UNREADABLE`.
|
||
Тест играет роль ядра: кладёт исходник в MinIO `uploads/…`, публикует запрос в `recognition.tabular.request` с
|
||
`Nats-Msg-Id = <job_id>:<attempt>` и собирает конверты из `recognition.result`.
|
||
|
||
Spec tabular-recognition-service: «Ack раньше результата», «Повторный запрос во время обработки» (здесь — после
|
||
обработки: битый файл отвергается мгновенно; повтор во время обработки покрыт unit-тестами воркера), ошибка разбора
|
||
конвертом (бывш. «Запрос с заглушкой парсера»), «Нет объекта в S3», «Несовместимая версия»,
|
||
«Реальная ОСВ через воркер» (маркер `fixtures`: реальный XLS из `../fixtures` → `done` и файл результата в MinIO),
|
||
«Реальные формы через воркер» (книга НПЗ → `done`, документы BALANCE и INCOME, `result_schema_version 1.2`).
|
||
|
||
Изоляция: `job_id`/`file_id`/ключ S3 уникальны на прогон (uuid4), исходник удаляется после теста; на шине может
|
||
работать другой сервис — сообщения `recognition.result` фильтруются по своему `job_id`. Ожидания — по событиям с
|
||
таймаутом; единственная фиксированная пауза — явное окно тишины `QUIET_WINDOW` для отрицательных проверок.
|
||
"""
|
||
|
||
import asyncio
|
||
import contextlib
|
||
import copy
|
||
import importlib
|
||
import io
|
||
import json
|
||
import uuid
|
||
from collections.abc import AsyncIterator
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pytest
|
||
from jsonschema import Draft202012Validator, FormatChecker
|
||
from miniopy_async import Minio
|
||
from nats.aio.client import Client as NATS
|
||
from nats.aio.msg import Msg
|
||
from result_contract import schema_errors
|
||
|
||
from src.contract import (
|
||
NATS_MSG_ID_HEADER,
|
||
RESULT_SUBJECT,
|
||
TABULAR_REQUEST_SUBJECT,
|
||
RecognitionResult,
|
||
request_msg_id,
|
||
)
|
||
from src.core.config import get_settings
|
||
|
||
pytestmark = pytest.mark.integration
|
||
|
||
REAL_OSV_66 = Path("osv") / "rb" / "2024-FY" / "ОСВ_66 за 2024.xls"
|
||
REAL_NPZ = (
|
||
Path("statements-xlsx")
|
||
/ "npz"
|
||
/ "2025-Q3"
|
||
/ "Бухгалтерская отчетность за Январь - Сентябрь 2025 г. (НОВОРОССИЙСКИЙ ПРОКАТНЫЙ ЗАВОД).xlsx"
|
||
)
|
||
|
||
CONTRACTS_DIR = Path(__file__).resolve().parents[2] / "contracts"
|
||
EXAMPLE_REQUEST = json.loads((CONTRACTS_DIR / "examples" / "request-tabular.json").read_text(encoding="utf-8"))
|
||
MESSAGE_VALIDATOR = Draft202012Validator(
|
||
json.loads((CONTRACTS_DIR / "recognition-result-message.schema.json").read_text(encoding="utf-8")),
|
||
format_checker=FormatChecker(),
|
||
)
|
||
|
||
SERVICE_NAME = "tabular-parser"
|
||
READY_TIMEOUT = 5.0 # подключение к NATS и подписка приложения
|
||
MESSAGE_TIMEOUT = 10.0 # одно сообщение сервиса (с запасом на повторы публикации)
|
||
# Окно тишины для отрицательных проверок («второго результата нет», «ack нет»): битый файл отвергается за миллисекунды,
|
||
# поэтому лишнее сообщение, если бы оно было, пришло бы гораздо раньше конца окна
|
||
QUIET_WINDOW = 1.5
|
||
SOURCE_BYTES = b"PK\x03\x04 round-trip: broken zip, the default parser answers UNREADABLE"
|
||
XLSX_CONTENT_TYPE = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
||
XLS_CONTENT_TYPE = "application/vnd.ms-excel"
|
||
|
||
app_module = importlib.import_module("src.app") # модуль, а не объект `app`, который реэкспортирует пакет `src`
|
||
|
||
|
||
# --- вспомогательное --------------------------------------------------------------------------------------------------
|
||
def _new_request(*, key_name: str | None = None, schema_version: str | None = None) -> dict[str, Any]:
|
||
"""Запрос из `contracts/examples/request-tabular.json` с уникальными `job_id`/`file_id`/`s3.key`."""
|
||
request = copy.deepcopy(EXAMPLE_REQUEST)
|
||
request["job_id"] = str(uuid.uuid4())
|
||
request["file_id"] = str(uuid.uuid4())
|
||
name = key_name or request["original_name"]
|
||
request["s3"]["key"] = f"uploads/{request['file_id']}/{request['job_id']}/{name}"
|
||
if schema_version is not None:
|
||
request["schema_version"] = schema_version
|
||
return request
|
||
|
||
|
||
def _describe(msg: Msg) -> str:
|
||
return f"Nats-Msg-Id={(msg.headers or {}).get(NATS_MSG_ID_HEADER)!r}, body={msg.data.decode('utf-8', 'replace')}"
|
||
|
||
|
||
def _envelope(msg: Msg, request: dict[str, Any], msg_id: str) -> dict[str, Any]:
|
||
"""Общие проверки конверта: заголовок, схема recognition-result-message, `job_id`/`file_id`, `service.name`."""
|
||
assert msg.headers is not None and msg.headers.get(NATS_MSG_ID_HEADER) == msg_id, _describe(msg)
|
||
body: dict[str, Any] = json.loads(msg.data)
|
||
errors = [f"{list(e.absolute_path)}: {e.message}" for e in MESSAGE_VALIDATOR.iter_errors(body)]
|
||
assert errors == [], f"конверт не проходит recognition-result-message.schema.json: {errors}; {_describe(msg)}"
|
||
assert body["job_id"] == request["job_id"]
|
||
assert body["file_id"] == request["file_id"]
|
||
assert body["service"]["name"] == SERVICE_NAME
|
||
return body
|
||
|
||
|
||
def _assert_error(msg: Msg, request: dict[str, Any], code: str) -> dict[str, Any]:
|
||
body = _envelope(msg, request, f"result:{request['job_id']}:{request['attempt']}")
|
||
assert body["status"] == "error", _describe(msg)
|
||
assert body["error"]["code"] == code, _describe(msg)
|
||
assert body["metrics"]["duration_ms"] >= 0
|
||
return body
|
||
|
||
|
||
def _assert_ack(msg: Msg, request: dict[str, Any]) -> None:
|
||
body = _envelope(msg, request, f"ack:{request['job_id']}:{request['attempt']}")
|
||
assert body["status"] == "processing", _describe(msg)
|
||
|
||
|
||
class Core:
|
||
"""Тестовый клиент в роли ядра: публикует запросы и собирает `recognition.result` по своим `job_id`."""
|
||
|
||
def __init__(self, client: NATS) -> None:
|
||
self.client = client
|
||
self._inboxes: dict[str, asyncio.Queue[Msg]] = {}
|
||
|
||
async def on_result(self, msg: Msg) -> None:
|
||
try:
|
||
job_id = json.loads(msg.data).get("job_id")
|
||
except (ValueError, AttributeError):
|
||
return # чужое сообщение на общей шине
|
||
inbox = self._inboxes.get(job_id) if isinstance(job_id, str) else None
|
||
if inbox is not None:
|
||
inbox.put_nowait(msg)
|
||
|
||
async def publish(self, request: dict[str, Any]) -> None:
|
||
self._inboxes.setdefault(request["job_id"], asyncio.Queue())
|
||
payload = json.dumps(request, ensure_ascii=False).encode("utf-8")
|
||
headers = {NATS_MSG_ID_HEADER: request_msg_id(request["job_id"], request["attempt"])}
|
||
await self.client.publish(TABULAR_REQUEST_SUBJECT, payload, headers=headers)
|
||
await self.client.flush()
|
||
|
||
async def next_message(self, request: dict[str, Any]) -> Msg:
|
||
try:
|
||
return await asyncio.wait_for(self._inboxes[request["job_id"]].get(), MESSAGE_TIMEOUT)
|
||
except TimeoutError:
|
||
pytest.fail(
|
||
f"нет сообщения recognition.result для job_id={request['job_id']} за {MESSAGE_TIMEOUT} с "
|
||
"(сервис не ответил; если на шине есть другой инстанс tabular-parser, запрос мог уйти ему)",
|
||
pytrace=False,
|
||
)
|
||
|
||
async def assert_quiet(self, request: dict[str, Any]) -> None:
|
||
"""Отрицательная проверка: в окне `QUIET_WINDOW` по этому `job_id` больше ничего не приходит."""
|
||
try:
|
||
extra = await asyncio.wait_for(self._inboxes[request["job_id"]].get(), QUIET_WINDOW)
|
||
except TimeoutError:
|
||
return
|
||
pytest.fail(f"лишнее сообщение по job_id={request['job_id']}: {_describe(extra)}", pytrace=False)
|
||
|
||
|
||
# --- фикстуры ---------------------------------------------------------------------------------------------------------
|
||
class Uploads:
|
||
"""Исходники в MinIO `uploads/…` с уникальными ключами; удаляются после теста."""
|
||
|
||
def __init__(self, client: Minio) -> None:
|
||
self.client = client
|
||
self.keys: list[tuple[str, str]] = []
|
||
self.buckets: set[str] = set()
|
||
|
||
async def put(
|
||
self, request: dict[str, Any], data: bytes = SOURCE_BYTES, content_type: str = XLSX_CONTENT_TYPE
|
||
) -> None:
|
||
bucket, key = request["s3"]["bucket"], request["s3"]["key"]
|
||
await self._ensure_bucket(bucket)
|
||
await self.client.put_object(bucket, key, io.BytesIO(data), len(data), content_type=content_type)
|
||
self.keys.append((bucket, key))
|
||
|
||
async def _ensure_bucket(self, bucket: str) -> None:
|
||
"""`NoSuchBucket` (OQ-80): на чистом MinIO бакета нет — тест создаёт его сам, compose не трогаем."""
|
||
if bucket not in self.buckets:
|
||
if not await self.client.bucket_exists(bucket):
|
||
await self.client.make_bucket(bucket)
|
||
self.buckets.add(bucket)
|
||
|
||
async def read(self, bucket: str, key: str) -> bytes:
|
||
response = await self.client.get_object(bucket, key)
|
||
try:
|
||
return await response.read()
|
||
finally:
|
||
response.release()
|
||
|
||
|
||
@pytest.fixture
|
||
async def uploads() -> AsyncIterator[Uploads]:
|
||
minio = get_settings().OPENBAO.MINIO
|
||
client = Minio(
|
||
minio.MINIO_URL,
|
||
access_key=minio.MINIO_ADMIN_LOGIN,
|
||
secret_key=minio.MINIO_ADMIN_PASSWORD,
|
||
secure=minio.MINIO_SECURE,
|
||
)
|
||
store = Uploads(client)
|
||
try:
|
||
yield store
|
||
finally:
|
||
try:
|
||
for bucket, key in store.keys:
|
||
# сбой удаления одного объекта не должен оставлять в MinIO остальные
|
||
with contextlib.suppress(Exception):
|
||
await client.remove_object(bucket, key)
|
||
finally:
|
||
await client.close_session()
|
||
|
||
|
||
@pytest.fixture
|
||
async def service(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[None]:
|
||
"""Приложение в процессе теста: lifespan с `NATS_ENABLED=True` (в `.env.test` — false); выход — остановка."""
|
||
monkeypatch.setattr(app_module.settings, "NATS_ENABLED", True)
|
||
application = app_module.app
|
||
tasks: list[asyncio.Task[None]] = []
|
||
async with application.router.lifespan_context(application):
|
||
if not await application.state.nats.wait_ready(READY_TIMEOUT):
|
||
pytest.fail(
|
||
f"NATS недоступен ({app_module.settings.OPENBAO.NATS.NATS_URL}, .env.test; make up)", pytrace=False
|
||
)
|
||
tasks = list(application.state.worker._tasks)
|
||
yield
|
||
# `_tasks` обнуляется в `stop()` безусловно — проверяем сами задачи, снятые до выхода из lifespan
|
||
assert tasks, "воркер не запустил рабочие корутины"
|
||
assert all(t.done() for t in tasks), "рабочие корутины воркера не остановлены"
|
||
|
||
|
||
@pytest.fixture
|
||
async def core(service: None) -> AsyncIterator[Core]:
|
||
"""Клиент «ядра», подписанный на `recognition.result` до первой публикации запроса."""
|
||
client = NATS()
|
||
await client.connect(get_settings().OPENBAO.NATS.NATS_URL, connect_timeout=2, max_reconnect_attempts=1)
|
||
core = Core(client)
|
||
await client.subscribe(RESULT_SUBJECT, cb=core.on_result)
|
||
await client.flush()
|
||
try:
|
||
yield core
|
||
finally:
|
||
await client.close()
|
||
|
||
|
||
# --- сценарии ---------------------------------------------------------------------------------------------------------
|
||
async def test_request_gets_ack_then_parser_error_and_duplicate_gets_only_ack(core: Core, uploads: Uploads) -> None:
|
||
"""«Ack раньше результата», ошибка разбора (`UNREADABLE`) конвертом, повтор запроса — только ещё один ack."""
|
||
request = _new_request()
|
||
await uploads.put(request)
|
||
|
||
await core.publish(request)
|
||
_assert_ack(await core.next_message(request), request)
|
||
_assert_error(await core.next_message(request), request, "UNREADABLE")
|
||
|
||
await core.publish(request) # та же доставка ещё раз: тот же `Nats-Msg-Id = <job_id>:<attempt>`
|
||
_assert_ack(await core.next_message(request), request)
|
||
await core.assert_quiet(request) # второго результата нет
|
||
|
||
|
||
async def test_missing_source_object_gives_file_read_error_and_service_keeps_working(
|
||
core: Core, uploads: Uploads
|
||
) -> None:
|
||
"""«Нет объекта в S3»: ack, затем `FILE_READ_ERROR`; следующий запрос обрабатывается как обычно."""
|
||
missing = _new_request(key_name="missing.xlsx") # объект не кладётся
|
||
|
||
await core.publish(missing)
|
||
_assert_ack(await core.next_message(missing), missing)
|
||
body = _assert_error(await core.next_message(missing), missing, "FILE_READ_ERROR")
|
||
assert missing["s3"]["key"] in body["error"]["message"]
|
||
|
||
following = _new_request()
|
||
await uploads.put(following)
|
||
await core.publish(following)
|
||
_assert_ack(await core.next_message(following), following)
|
||
_assert_error(await core.next_message(following), following, "UNREADABLE")
|
||
|
||
|
||
async def test_unsupported_schema_version_gives_single_schema_mismatch_without_ack(
|
||
core: Core, uploads: Uploads
|
||
) -> None:
|
||
"""«Несовместимая версия»: `schema_version = 2.0` → один конверт `SCHEMA_MISMATCH`, ack не публикуется."""
|
||
request = _new_request(schema_version="2.0")
|
||
await uploads.put(request)
|
||
|
||
await core.publish(request)
|
||
# Ack публикуется до результата тем же соединением: если бы он был, он пришёл бы первым
|
||
body = _assert_error(await core.next_message(request), request, "SCHEMA_MISMATCH")
|
||
assert "2.0" in body["error"]["message"]
|
||
await core.assert_quiet(request)
|
||
|
||
|
||
@pytest.mark.fixtures
|
||
async def test_real_osv_through_worker_gives_done_and_valid_result_file(
|
||
core: Core, uploads: Uploads, fixtures_dir: Path
|
||
) -> None:
|
||
"""«Реальная ОСВ через воркер»: реальный XLS в MinIO → ack → `done`, `result_schema_version 1.1`, файл по схеме."""
|
||
request = _new_request(key_name="osv_66_2024.xls")
|
||
request["file_type"] = "XLS"
|
||
await uploads.put(request, (fixtures_dir / REAL_OSV_66).read_bytes(), XLS_CONTENT_TYPE)
|
||
|
||
await core.publish(request)
|
||
_assert_ack(await core.next_message(request), request)
|
||
body = _envelope(await core.next_message(request), request, f"result:{request['job_id']}:{request['attempt']}")
|
||
|
||
assert body["status"] == "done", body
|
||
assert body["result_schema_version"] == "1.2"
|
||
assert body["metrics"]["duration_ms"] >= 0
|
||
result_ref = body["result_s3"]
|
||
assert result_ref["bucket"] == request["s3"]["bucket"]
|
||
assert result_ref["key"] == f"results/{request['file_id']}/{request['job_id']}/result.json"
|
||
uploads.keys.append((result_ref["bucket"], result_ref["key"])) # результат тоже убирается после теста
|
||
|
||
stored: dict[str, Any] = json.loads(await uploads.read(result_ref["bucket"], result_ref["key"]))
|
||
assert schema_errors(stored) == []
|
||
assert RecognitionResult.model_validate(stored).to_wire() == stored
|
||
assert stored["status"] == "done"
|
||
assert stored["job_id"] == request["job_id"]
|
||
assert [document["document_type"] for document in stored["documents"]] == ["OSV"]
|
||
assert stored["documents"][0]["header"]["account_code_raw"] == "66"
|
||
await core.assert_quiet(request)
|
||
|
||
|
||
@pytest.mark.fixtures
|
||
async def test_real_statements_through_worker_give_done_with_two_documents(
|
||
core: Core, uploads: Uploads, fixtures_dir: Path
|
||
) -> None:
|
||
"""«Реальные формы через воркер»: книга НПЗ → ack → `done`; расхождения итогов — округление (OQ-138, мягкие)."""
|
||
request = _new_request(key_name="npz_2025_q3.xlsx")
|
||
request["file_type"] = "XLSX"
|
||
await uploads.put(request, (fixtures_dir / REAL_NPZ).read_bytes(), XLSX_CONTENT_TYPE)
|
||
|
||
await core.publish(request)
|
||
_assert_ack(await core.next_message(request), request)
|
||
body = _envelope(await core.next_message(request), request, f"result:{request['job_id']}:{request['attempt']}")
|
||
|
||
assert body["status"] == "done", body
|
||
assert body["result_schema_version"] == "1.2"
|
||
result_ref = body["result_s3"]
|
||
uploads.keys.append((result_ref["bucket"], result_ref["key"]))
|
||
|
||
stored: dict[str, Any] = json.loads(await uploads.read(result_ref["bucket"], result_ref["key"]))
|
||
assert schema_errors(stored) == []
|
||
assert RecognitionResult.model_validate(stored).to_wire() == stored
|
||
assert stored["status"] == "done"
|
||
balance, income = stored["documents"]
|
||
assert [balance["document_type"], income["document_type"]] == ["BALANCE", "INCOME"]
|
||
assert balance["checks"]["relations_failed"] == [] and income["checks"]["relations_failed"] == []
|
||
# расхождения в пределах округления (OQ-138) видны аналитику как мягкие
|
||
assert "1600 = 1700" in {item["relation"] for item in balance["checks"]["soft_relations_failed"]}
|
||
assert "2100 = 2110 - 2120" in {item["relation"] for item in income["checks"]["soft_relations_failed"]}
|
||
# лист с единственной строкой 2400 присоединён (O-9)
|
||
assert "2400" in {line["line_code"] for line in income["lines"]}
|
||
await core.assert_quiet(request)
|