Merge pull request 'NATS: subject'ы и JWT по окружению из OpenBao (OPENBAO__NATS__*) (#31)' (#37) from feat/31-nats-env-config into main
Reviewed-on: #37
This commit is contained in:
commit
143744b80d
@ -4,7 +4,7 @@
|
||||
|
||||
## Что это
|
||||
Stateless-сервис разбора XLS/XLSX/XLSB: ОСВ по счетам (§1 правил) и ББ/ОФР из Excel (§12). Вход — NATS
|
||||
`files.tabular_uploaded` + файл в S3 `uploads/`; выход — JSON по `../docs/contracts/recognition-result.schema.json`
|
||||
`files.tabular_uploaded` (по умолчанию; задаётся OPENBAO__NATS__NATS_*_SUBJECT) + файл в S3 `uploads/`; выход — JSON по `../docs/contracts/recognition-result.schema.json`
|
||||
в S3 `results/` + NATS `files.tabular_parsed`. **Нет доступа к БД.** Не знает про организации, периметры, план счетов —
|
||||
возвращает структуру файла; нормализация и резолв — в arm-core.
|
||||
|
||||
|
||||
@ -382,7 +382,7 @@ cd api && .venv/bin/python run_server.py
|
||||
curl -s localhost:8000/healthcheck
|
||||
```
|
||||
|
||||
В ответе — `uptime`, `version` и `nats`: `ok` — подключён, `down` — включён (`NATS_ENABLED=true`), но связи нет,
|
||||
В ответе — `uptime`, `version`, `subjects` (используемые subject'ы: `tabular_request`, `result`; JWT там нет) и `nats`: `ok` — подключён, `down` — включён (`NATS_ENABLED=true`), но связи нет,
|
||||
`off` — выключен (`NATS_ENABLED=false`, воркер не запущен).
|
||||
|
||||
Если NATS или MinIO недоступны при старте, приложение всё равно поднимается: ошибка логируется, соединение с NATS
|
||||
@ -468,6 +468,9 @@ ENV_FILE=.env.test .venv/bin/python -m src.parser.cli <файл> [--file-type XL
|
||||
| `OPENBAO__NATS__NATS_URL` | — (обязательна) | адрес NATS, `nats://host:4222` |
|
||||
| `OPENBAO__NATS__NATS_SEED` | пусто | nkeys seed; пусто вместе с JWT — подключение без аутентификации (dev) |
|
||||
| `OPENBAO__NATS__NATS_JWT` | пусто | JWT пользователя NATS |
|
||||
| `OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT` | `files.tabular_uploaded` | subject запроса `tabular`, на который подписан сервис |
|
||||
| `OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT` | `files.img_uploaded` | subject запроса `image` (слушает scan-gateway; здесь — единый набор ключей) |
|
||||
| `OPENBAO__NATS__NATS_RESULT_SUBJECT` | `files.tabular_parsed` | subject, в который сервис публикует ack и результат |
|
||||
| `DEBUG` | `false` | уровень логов DEBUG |
|
||||
| `ENV_FILE` | `.env` | путь к файлу окружения (от `api/`) |
|
||||
| `NATS_ENABLED` | `true` | `false` — не подключаться к NATS, воркер не стартует |
|
||||
|
||||
@ -27,6 +27,11 @@ OPENBAO__MINIO__MINIO_SECURE=false
|
||||
OPENBAO__NATS__NATS_URL=nats://localhost:4222
|
||||
OPENBAO__NATS__NATS_SEED=
|
||||
OPENBAO__NATS__NATS_JWT=
|
||||
# Subject'ы (необязательно; значения ниже — по умолчанию). Запрос tabular — подписка, результат — публикация.
|
||||
# Image-запрос слушает scan-gateway; здесь ключ нужен только для единого набора OPENBAO__NATS__*
|
||||
# OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT=files.tabular_uploaded
|
||||
# OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT=files.img_uploaded
|
||||
# OPENBAO__NATS__NATS_RESULT_SUBJECT=files.tabular_parsed
|
||||
|
||||
# Необязательные настройки (значения ниже — значения по умолчанию из src/core/config.py)
|
||||
# Префикс путей приложения за прокси и версия в OpenAPI
|
||||
|
||||
@ -2,7 +2,7 @@
|
||||
"$schema": "https://json-schema.org/draft/2020-12/schema",
|
||||
"$id": "arm-ka/recognition-request/v1",
|
||||
"title": "Запрос распознавания файла (arm-core → tabular-parser / scan-gateway), контракт v1.0",
|
||||
"description": "Тело сообщения на subject files.tabular_uploaded или files.img_uploaded; subject должен соответствовать полю pipeline (tabular → files.tabular_uploaded, image → files.img_uploaded). Заголовок Nats-Msg-Id = <job_id>:<attempt> (не поле тела). Один файл (stored_file) → один job.",
|
||||
"description": "Тело сообщения на subject запроса; subject — настройка окружения и должен соответствовать полю pipeline (tabular → NATS_TABULAR_REQUEST_SUBJECT, по умолчанию files.tabular_uploaded; image → NATS_IMAGE_REQUEST_SUBJECT, по умолчанию files.img_uploaded). Заголовок Nats-Msg-Id = <job_id>:<attempt> (не поле тела). Один файл (stored_file) → один job.",
|
||||
"$comment": "Версионирование: schema_version = MAJOR.MINOR. MINOR (1.1) — необязательные поля и новые значения enum; MAJOR (2.0) — удаление/переименование полей, смена типов или семантики. Потребитель принимает тот же MAJOR и любой MINOR. Новый MINOR раскатывается сначала на потребителей (ядро и сервисы обновляют копии схем раньше, чем производитель начнёт слать 1.1), потому что закрытые enum отвергают незнакомые значения.",
|
||||
"type": "object",
|
||||
"required": ["schema_version", "job_id", "file_id", "pipeline", "file_type", "original_name", "s3", "reply_deadline", "attempt"],
|
||||
|
||||
@ -2,7 +2,7 @@
|
||||
"$schema": "https://json-schema.org/draft/2020-12/schema",
|
||||
"$id": "arm-ka/recognition-result-message/v1",
|
||||
"title": "Сообщение результата распознавания (tabular-parser / scan-gateway → arm-core), контракт v1.0",
|
||||
"description": "Тело сообщения на subject files.tabular_parsed. Заголовок Nats-Msg-Id (не поле тела) на этом subject имеет две формы: ack:<job_id>:<attempt> — для status = processing; result:<job_id>:<attempt> — для done, needs_check и error. Форма запроса <job_id>:<attempt> описана в схеме recognition-request. status = processing — ack «взял в работу», публикуется до чтения файла из S3. group_schema в конверт не входит: источник — файл результата (recognition-result).",
|
||||
"description": "Тело сообщения на subject ответа (настройка окружения NATS_RESULT_SUBJECT, по умолчанию files.tabular_parsed). Заголовок Nats-Msg-Id (не поле тела) на этом subject имеет две формы: ack:<job_id>:<attempt> — для status = processing; result:<job_id>:<attempt> — для done, needs_check и error. Форма запроса <job_id>:<attempt> описана в схеме recognition-request. status = processing — ack «взял в работу», публикуется до чтения файла из S3. group_schema в конверт не входит: источник — файл результата (recognition-result).",
|
||||
"$comment": "Версионирование: schema_version = MAJOR.MINOR. MINOR (1.1) — необязательные поля и новые значения enum; MAJOR (2.0) — удаление/переименование полей, смена типов или семантики. Потребитель принимает тот же MAJOR и любой MINOR. Новый MINOR раскатывается сначала на потребителей (ядро и сервисы обновляют копии схем раньше, чем производитель начнёт слать 1.1), потому что закрытые enum отвергают незнакомые значения.",
|
||||
"type": "object",
|
||||
"required": ["schema_version", "job_id", "file_id", "status", "service"],
|
||||
|
||||
@ -2,7 +2,7 @@
|
||||
|
||||
Что здесь есть: настройки, подключение S3 (MinIO), NATS (`src/core/nats.py`: nkeys JWT, пустые seed/jwt — без
|
||||
аутентификации, фоновое подключение — старт не ждёт NATS), воркер `src/core/worker.py` (подписка на
|
||||
`files.tabular_uploaded` после подключения к NATS), middleware (gzip, proxy headers), `/healthcheck`
|
||||
subject запроса `tabular` из настроек после подключения к NATS), middleware (gzip, proxy headers), `/healthcheck`
|
||||
(`nats = ok | down | off`, R-17), обработчики исключений. Сервис без БД и фронта (DECISION-4).
|
||||
|
||||
При `NATS_ENABLED=false` ни NATS, ни воркер не запускаются. Остановка: воркер перестаёт принимать запросы и
|
||||
@ -24,7 +24,7 @@ from src.core.clients import S3Storage
|
||||
from src.core.config import get_settings
|
||||
from src.core.exception_handlers import register_exception_handlers
|
||||
from src.core.logger import setup_logger
|
||||
from src.core.nats import NatsConnection
|
||||
from src.core.nats import NatsConnection, used_subjects
|
||||
from src.core.worker import STOP_TIMEOUT, RecognitionWorker
|
||||
from src.parser import get_parser
|
||||
|
||||
@ -85,12 +85,14 @@ def _build_worker(storage: S3Storage) -> tuple[NatsConnection, RecognitionWorker
|
||||
service_version=VERSION,
|
||||
concurrency=settings.WORKER_CONCURRENCY,
|
||||
done_cache_size=settings.DONE_CACHE_SIZE,
|
||||
result_subject=settings.OPENBAO.NATS.NATS_RESULT_SUBJECT,
|
||||
)
|
||||
return nats, worker
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
logger.info("Subject'ы NATS. {subjects}", subjects=used_subjects(settings.OPENBAO.NATS))
|
||||
storage = await _connect_s3()
|
||||
nats: NatsConnection | None = None
|
||||
worker: RecognitionWorker | None = None
|
||||
@ -130,7 +132,12 @@ def _nats_state(application: FastAPI) -> str:
|
||||
|
||||
@app.get("/healthcheck", status_code=200)
|
||||
async def healthcheck(request: Request):
|
||||
return {"uptime": UP_TIME, "version": VERSION, "nats": _nats_state(request.app)}
|
||||
return {
|
||||
"uptime": UP_TIME,
|
||||
"version": VERSION,
|
||||
"nats": _nats_state(request.app),
|
||||
"subjects": used_subjects(settings.OPENBAO.NATS),
|
||||
}
|
||||
|
||||
|
||||
app.include_router(router)
|
||||
|
||||
@ -26,10 +26,12 @@ REQUEST_SUBJECTS: Mapping[str, str] = {
|
||||
}
|
||||
|
||||
|
||||
def request_subject(pipeline: str) -> str:
|
||||
"""Subject запроса по pipeline: `tabular` → `files.tabular_uploaded`, `image` → `files.img_uploaded`."""
|
||||
def request_subject(pipeline: str, subjects: Mapping[str, str] = REQUEST_SUBJECTS) -> str:
|
||||
"""Subject запроса по pipeline. `subjects` — карта `pipeline → subject` окружения
|
||||
(`NATSSettings.request_subjects()`); по умолчанию — имена контракта: `tabular` → `files.tabular_uploaded`,
|
||||
`image` → `files.img_uploaded`."""
|
||||
try:
|
||||
return REQUEST_SUBJECTS[pipeline]
|
||||
return subjects[pipeline]
|
||||
except KeyError:
|
||||
raise ValueError(f"неизвестный pipeline {pipeline!r}: допустимы {', '.join(get_args(Pipeline))}") from None
|
||||
|
||||
|
||||
@ -1,13 +1,21 @@
|
||||
import os
|
||||
import re
|
||||
from typing import Any, Literal
|
||||
|
||||
from pydantic import NonNegativeInt, PositiveInt, model_validator
|
||||
from pydantic import NonNegativeInt, PositiveInt, SecretStr, ValidationInfo, field_validator, model_validator
|
||||
from pydantic_settings import BaseSettings, PydanticBaseSettingsSource, SettingsConfigDict
|
||||
|
||||
from src.contract.subjects import IMAGE_REQUEST_SUBJECT, RESULT_SUBJECT, TABULAR_REQUEST_SUBJECT
|
||||
from src.core.settings_sources import OpenBaoAppSettingsSource
|
||||
|
||||
# Subject NATS без шаблонов: непустые токены через точку, без пробельных символов, `*` и `>` (docs nats-env-config)
|
||||
NATS_SUBJECT_RE = re.compile(r"[^\s.*>]+(?:\.[^\s.*>]+)*")
|
||||
NATS_SUBJECT_FIELDS = ("NATS_TABULAR_REQUEST_SUBJECT", "NATS_IMAGE_REQUEST_SUBJECT", "NATS_RESULT_SUBJECT")
|
||||
|
||||
|
||||
class MinioSettings(BaseSettings):
|
||||
model_config = SettingsConfigDict(hide_input_in_errors=True)
|
||||
|
||||
MINIO_URL: str
|
||||
MINIO_ADMIN_LOGIN: str
|
||||
MINIO_ADMIN_PASSWORD: str
|
||||
@ -16,12 +24,54 @@ class MinioSettings(BaseSettings):
|
||||
|
||||
|
||||
class NATSSettings(BaseSettings):
|
||||
"""Подключение и subject'ы NATS — секреты окружения `OPENBAO__NATS__*` (прод и тест — свои наборы).
|
||||
|
||||
Пустые `NATS_SEED` и `NATS_JWT` — подключение без nkeys (локальная разработка, тесты). Subject'ы по умолчанию —
|
||||
имена контракта v1.0; у ядра и сервисов одного окружения они обязаны совпадать (docs, change nats-env-config).
|
||||
"""
|
||||
|
||||
model_config = SettingsConfigDict(hide_input_in_errors=True)
|
||||
|
||||
NATS_URL: str
|
||||
NATS_SEED: str = "" # пусто — подключение к NATS без nkeys (локальная разработка, тесты)
|
||||
NATS_JWT: str = ""
|
||||
NATS_SEED: SecretStr = SecretStr("")
|
||||
NATS_JWT: SecretStr = SecretStr("")
|
||||
NATS_TABULAR_REQUEST_SUBJECT: str = TABULAR_REQUEST_SUBJECT
|
||||
NATS_IMAGE_REQUEST_SUBJECT: str = IMAGE_REQUEST_SUBJECT
|
||||
NATS_RESULT_SUBJECT: str = RESULT_SUBJECT
|
||||
|
||||
@field_validator(*NATS_SUBJECT_FIELDS)
|
||||
@classmethod
|
||||
def _check_subject(cls, value: str, info: ValidationInfo) -> str:
|
||||
"""Конкретный subject без шаблонов: иначе подписка молча слушала бы чужой трафик или ничего."""
|
||||
if not NATS_SUBJECT_RE.fullmatch(value):
|
||||
raise ValueError(
|
||||
f"{info.field_name}={value!r}: нужен subject NATS из непустых токенов через точку, "
|
||||
"без пробелов, '*' и '>'"
|
||||
)
|
||||
return value
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _check_distinct_subjects(self) -> "NATSSettings":
|
||||
"""Три subject'а попарно различны: совпадение запроса и ответа — сервис получает свои ответы, двух запросов —
|
||||
каждый запрос получают оба сервиса."""
|
||||
seen: dict[str, str] = {}
|
||||
for field in NATS_SUBJECT_FIELDS:
|
||||
subject = getattr(self, field)
|
||||
if subject in seen:
|
||||
raise ValueError(
|
||||
f"{seen[subject]} и {field} совпадают ({subject!r}): subject'ы NATS должны различаться"
|
||||
)
|
||||
seen[subject] = field
|
||||
return self
|
||||
|
||||
def request_subjects(self) -> dict[str, str]:
|
||||
"""Subject запроса по `pipeline` (`tabular`, `image`) — аргумент `request_subject` контракта."""
|
||||
return {"tabular": self.NATS_TABULAR_REQUEST_SUBJECT, "image": self.NATS_IMAGE_REQUEST_SUBJECT}
|
||||
|
||||
|
||||
class OpenBAOSettings(BaseSettings):
|
||||
model_config = SettingsConfigDict(hide_input_in_errors=True)
|
||||
|
||||
MINIO: MinioSettings
|
||||
NATS: NATSSettings
|
||||
|
||||
@ -37,6 +87,8 @@ class OpenBAOSettings(BaseSettings):
|
||||
|
||||
|
||||
class ServerSettings(BaseSettings):
|
||||
model_config = SettingsConfigDict(hide_input_in_errors=True)
|
||||
|
||||
HOST: str = "0.0.0.0"
|
||||
PORT: int = 8000
|
||||
|
||||
@ -84,6 +136,7 @@ class Settings(BaseSettings):
|
||||
)
|
||||
|
||||
model_config = SettingsConfigDict(
|
||||
hide_input_in_errors=True,
|
||||
extra="ignore",
|
||||
env_file=os.environ.get("ENV_FILE", ".env"),
|
||||
env_nested_delimiter="__",
|
||||
|
||||
@ -10,7 +10,8 @@
|
||||
завершился исключением, supervisor повторяет попытку с паузой 1 → 2 → 4 → … ≤ 30 с. После подключения один раз
|
||||
вызывается `on_connected(conn)` (подписки воркера); если он упал, созданные в этой попытке подписки снимаются и
|
||||
вызов повторяется с той же паузой. Подписки после разрыва nats-py восстанавливает сам.
|
||||
- `subscribe_requests(cb)` — подписка на `files.tabular_uploaded` в queue group `tabular-parser`: из нескольких
|
||||
- `subscribe_requests(cb)` — подписка на subject запроса `tabular` из настроек (по умолчанию
|
||||
`files.tabular_uploaded`) в queue group `tabular-parser`: из нескольких
|
||||
инстанций сообщение получает одна. После `subscribe` — `flush`: сервер обработал SUB, ошибки соединения проявляются
|
||||
здесь. Отказ в правах (`Permissions Violation`) nats-py 2.14 так не проявляет: соединение остаётся открытым,
|
||||
`subscribe`/`flush`/`publish` проходят, сервер лишь присылает `-ERR` в `error_cb` — он логируется отдельно уровнем
|
||||
@ -37,7 +38,7 @@ from loguru import logger
|
||||
from nats.aio.client import Client as NATS
|
||||
from nats.aio.msg import Msg
|
||||
|
||||
from src.contract.subjects import NATS_MSG_ID_HEADER, TABULAR_REQUEST_SUBJECT
|
||||
from src.contract.subjects import NATS_MSG_ID_HEADER
|
||||
from src.core.config import NATSSettings
|
||||
|
||||
REQUEST_QUEUE_GROUP = "tabular-parser"
|
||||
@ -88,8 +89,8 @@ def build_nats_auth(settings: NATSSettings) -> dict[str, Any]:
|
||||
Пустые `NATS_SEED` и `NATS_JWT` — без учётных данных (docker compose, тесты): `{}`. Заданы оба — колбэки
|
||||
nkeys/JWT, как в шаблоне. Задан только один или seed не читается — `NatsConfigError`.
|
||||
"""
|
||||
seed = settings.NATS_SEED
|
||||
jwt = settings.NATS_JWT
|
||||
seed = settings.NATS_SEED.get_secret_value()
|
||||
jwt = settings.NATS_JWT.get_secret_value()
|
||||
if not seed and not jwt:
|
||||
return {}
|
||||
if not seed or not jwt:
|
||||
@ -112,6 +113,11 @@ def build_nats_auth(settings: NATSSettings) -> dict[str, Any]:
|
||||
return {"user_jwt_cb": user_jwt_cb, "signature_cb": signature_cb}
|
||||
|
||||
|
||||
def used_subjects(settings: NATSSettings) -> dict[str, str]:
|
||||
"""Subject'ы сервиса для лога старта и `/healthcheck.subjects`: подписан на запрос `tabular`, публикует ответ."""
|
||||
return {"tabular_request": settings.NATS_TABULAR_REQUEST_SUBJECT, "result": settings.NATS_RESULT_SUBJECT}
|
||||
|
||||
|
||||
class NatsConnection:
|
||||
"""Соединение сервиса с NATS: фоновое подключение, подписка на запросы, публикация с повторами.
|
||||
|
||||
@ -276,9 +282,13 @@ class NatsConnection:
|
||||
|
||||
# --- подписка и публикация ------------------------------------------------------------------
|
||||
async def subscribe_requests(
|
||||
self, cb: MessageHandler, *, queue: str = REQUEST_QUEUE_GROUP, subject: str = TABULAR_REQUEST_SUBJECT
|
||||
self, cb: MessageHandler, *, queue: str = REQUEST_QUEUE_GROUP, subject: str | None = None
|
||||
) -> Any:
|
||||
"""Подписка на запросы (`files.tabular_uploaded`) в queue group; исключения `cb` логируются."""
|
||||
"""Подписка на запросы в queue group; исключения `cb` логируются.
|
||||
|
||||
Без `subject` — `NATS_TABULAR_REQUEST_SUBJECT` настроек (по умолчанию `files.tabular_uploaded`).
|
||||
"""
|
||||
subject = subject if subject is not None else self._settings.NATS_TABULAR_REQUEST_SUBJECT
|
||||
if not self.is_connected:
|
||||
raise NatsNotConnectedError(f"Нет соединения с NATS: подписка на {subject} невозможна")
|
||||
|
||||
|
||||
@ -1,4 +1,5 @@
|
||||
"""Воркер распознавания: запрос `files.tabular_uploaded` → ack → S3 → парсер → S3 → `files.tabular_parsed`.
|
||||
"""Воркер распознавания: запрос на subject запросов → ack → S3 → парсер → S3 → subject ответа
|
||||
(по умолчанию `files.tabular_uploaded` / `files.tabular_parsed`).
|
||||
|
||||
Design D-2, D-4, D-5, D-7; spec tabular-recognition-service «Подтверждение запроса до обработки», «Результат по
|
||||
контракту», «Ошибки транспорта превращаются в результат», «Парсер — заменяемый интерфейс».
|
||||
@ -141,11 +142,13 @@ class RecognitionWorker:
|
||||
concurrency: int = 1,
|
||||
done_cache_size: int = 1000,
|
||||
clock: Clock = time.monotonic,
|
||||
result_subject: str = RESULT_SUBJECT,
|
||||
) -> None:
|
||||
if concurrency < 1 or done_cache_size < 1:
|
||||
raise ValueError("concurrency и done_cache_size должны быть ≥ 1")
|
||||
self._storage = storage
|
||||
self._publisher = publisher
|
||||
self._result_subject = result_subject
|
||||
self._parser = parser
|
||||
self._service = ServiceInfo.model_validate({"name": service_name, "version": service_version})
|
||||
self._concurrency = concurrency
|
||||
@ -198,7 +201,8 @@ class RecognitionWorker:
|
||||
|
||||
# --- приём запроса ----------------------------------------------------------------------------------------------
|
||||
async def on_request(self, msg: InboundMessage) -> None:
|
||||
"""Обработчик подписки на `files.tabular_uploaded`; исключения наружу не выходят."""
|
||||
"""Обработчик подписки на subject запросов (по умолчанию `files.tabular_uploaded`);
|
||||
исключения наружу не выходят."""
|
||||
if not self._accepting:
|
||||
logger.warning("Воркер останавливается: запрос не принят (без ack, ядро переотправит)")
|
||||
return
|
||||
@ -464,7 +468,7 @@ class RecognitionWorker:
|
||||
"""Публикация конверта; исчерпанные повторы и прочие сбои — error-лог, не исключение (D-5)."""
|
||||
payload = json.dumps(envelope.to_wire(), ensure_ascii=False).encode("utf-8")
|
||||
try:
|
||||
await self._publisher.publish(RESULT_SUBJECT, payload, msg_id)
|
||||
await self._publisher.publish(self._result_subject, payload, msg_id)
|
||||
except NatsPublishError as exc:
|
||||
logger.error(
|
||||
"Конверт не опубликован после всех повторов; ядро закроет задание по reply_deadline. "
|
||||
|
||||
@ -3,8 +3,8 @@
|
||||
Приложение поднимается в процессе теста через lifespan (`src.app`) с `NATS_ENABLED=True` — тот же код, что в проде:
|
||||
`_connect_s3`, `NatsConnection` с подпиской в queue group `tabular-parser`, `RecognitionWorker` с парсером по умолчанию
|
||||
(`get_parser(settings)` → `OsvParser`); исходник — битый zip, поэтому результат — ошибка `UNREADABLE`.
|
||||
Тест играет роль ядра: кладёт исходник в MinIO `uploads/…`, публикует запрос в `files.tabular_uploaded` с
|
||||
`Nats-Msg-Id = <job_id>:<attempt>` и собирает конверты из `files.tabular_parsed`.
|
||||
Тест играет роль ядра: кладёт исходник в MinIO `uploads/…`, публикует запрос на subject запросов прогона с
|
||||
`Nats-Msg-Id = <job_id>:<attempt>` и собирает конверты с subject'а ответа прогона.
|
||||
|
||||
Spec tabular-recognition-service: «Ack раньше результата», «Повторный запрос во время обработки» (здесь — после
|
||||
обработки: битый файл отвергается мгновенно; повтор во время обработки покрыт unit-тестами воркера), ошибка разбора
|
||||
@ -12,8 +12,9 @@ Spec tabular-recognition-service: «Ack раньше результата», «
|
||||
«Реальная ОСВ через воркер» (маркер `fixtures`: реальный XLS из `../fixtures` → `done` и файл результата в MinIO),
|
||||
«Реальные формы через воркер» (книга НПЗ → `done`, документы BALANCE и INCOME, `result_schema_version 1.2`).
|
||||
|
||||
Изоляция: `job_id`/`file_id`/ключ S3 уникальны на прогон (uuid4), исходник удаляется после теста; на шине может
|
||||
работать другой сервис — сообщения `files.tabular_parsed` фильтруются по своему `job_id`. Ожидания — по событиям с
|
||||
Изоляция: subject'ы прогона уникальны — контейнер стенда сообщения теста не получает (OQ-83);
|
||||
`job_id`/`file_id`/ключ S3 уникальны на прогон (uuid4), исходник удаляется после теста; на шине может
|
||||
работать другой сервис — сообщения с subject'а ответа фильтруются по своему `job_id`. Ожидания — по событиям с
|
||||
таймаутом; единственная фиксированная пауза — явное окно тишины `QUIET_WINDOW` для отрицательных проверок.
|
||||
"""
|
||||
|
||||
@ -37,8 +38,6 @@ from result_contract import schema_errors
|
||||
|
||||
from src.contract import (
|
||||
NATS_MSG_ID_HEADER,
|
||||
RESULT_SUBJECT,
|
||||
TABULAR_REQUEST_SUBJECT,
|
||||
RecognitionResult,
|
||||
request_msg_id,
|
||||
)
|
||||
@ -71,6 +70,14 @@ SOURCE_BYTES = b"PK\x03\x04 round-trip: broken zip, the default parser answers U
|
||||
XLSX_CONTENT_TYPE = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
||||
XLS_CONTENT_TYPE = "application/vnd.ms-excel"
|
||||
|
||||
# Subject'ы прогона уникальны (OQ-83): контейнер tabular-parser стенда слушает умолчания и сообщения теста не получает
|
||||
RUN = uuid.uuid4().hex[:8]
|
||||
TEST_SUBJECTS = {
|
||||
"NATS_TABULAR_REQUEST_SUBJECT": f"test-{RUN}.files.tabular_uploaded",
|
||||
"NATS_IMAGE_REQUEST_SUBJECT": f"test-{RUN}.files.img_uploaded",
|
||||
"NATS_RESULT_SUBJECT": f"test-{RUN}.files.tabular_parsed",
|
||||
}
|
||||
|
||||
app_module = importlib.import_module("src.app") # модуль, а не объект `app`, который реэкспортирует пакет `src`
|
||||
|
||||
|
||||
@ -117,7 +124,7 @@ def _assert_ack(msg: Msg, request: dict[str, Any]) -> None:
|
||||
|
||||
|
||||
class Core:
|
||||
"""Тестовый клиент в роли ядра: публикует запросы и собирает `files.tabular_parsed` по своим `job_id`."""
|
||||
"""Тестовый клиент в роли ядра: публикует запросы и собирает сообщения с subject'а ответа по своим `job_id`."""
|
||||
|
||||
def __init__(self, client: NATS) -> None:
|
||||
self.client = client
|
||||
@ -136,7 +143,7 @@ class Core:
|
||||
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.publish(TEST_SUBJECTS["NATS_TABULAR_REQUEST_SUBJECT"], payload, headers=headers)
|
||||
await self.client.flush()
|
||||
|
||||
async def next_message(self, request: dict[str, Any]) -> Msg:
|
||||
@ -144,7 +151,7 @@ class Core:
|
||||
return await asyncio.wait_for(self._inboxes[request["job_id"]].get(), MESSAGE_TIMEOUT)
|
||||
except TimeoutError:
|
||||
pytest.fail(
|
||||
f"нет сообщения files.tabular_parsed для job_id={request['job_id']} за {MESSAGE_TIMEOUT} с "
|
||||
f"нет сообщения на subject ответа прогона для job_id={request['job_id']} за {MESSAGE_TIMEOUT} с "
|
||||
"(сервис не ответил; если на шине есть другой инстанс tabular-parser, запрос мог уйти ему)",
|
||||
pytrace=False,
|
||||
)
|
||||
@ -216,6 +223,8 @@ async def uploads() -> AsyncIterator[Uploads]:
|
||||
async def service(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[None]:
|
||||
"""Приложение в процессе теста: lifespan с `NATS_ENABLED=True` (в `.env.test` — false); выход — остановка."""
|
||||
monkeypatch.setattr(app_module.settings, "NATS_ENABLED", True)
|
||||
nats_settings = app_module.settings.OPENBAO.NATS
|
||||
monkeypatch.setattr(app_module.settings.OPENBAO, "NATS", nats_settings.model_copy(update=TEST_SUBJECTS))
|
||||
application = app_module.app
|
||||
tasks: list[asyncio.Task[None]] = []
|
||||
async with application.router.lifespan_context(application):
|
||||
@ -232,11 +241,11 @@ async def service(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[None]:
|
||||
|
||||
@pytest.fixture
|
||||
async def core(service: None) -> AsyncIterator[Core]:
|
||||
"""Клиент «ядра», подписанный на `files.tabular_parsed` до первой публикации запроса."""
|
||||
"""Клиент «ядра», подписанный на subject ответа прогона до первой публикации запроса."""
|
||||
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.subscribe(TEST_SUBJECTS["NATS_RESULT_SUBJECT"], cb=core.on_result)
|
||||
await client.flush()
|
||||
try:
|
||||
yield core
|
||||
|
||||
@ -16,6 +16,7 @@ from typing import Any
|
||||
import httpx
|
||||
import pytest
|
||||
from loguru import logger
|
||||
from pydantic import SecretStr
|
||||
|
||||
from src.contract import NATS_MSG_ID_HEADER, RESULT_SUBJECT
|
||||
from src.core.clients import S3Storage
|
||||
@ -124,6 +125,11 @@ def patched_app(monkeypatch: pytest.MonkeyPatch, events: list[str]) -> Iterator[
|
||||
yield app_module
|
||||
|
||||
|
||||
def _expected_subjects(module: Any) -> dict[str, str]:
|
||||
nats = module.settings.OPENBAO.NATS
|
||||
return {"tabular_request": nats.NATS_TABULAR_REQUEST_SUBJECT, "result": nats.NATS_RESULT_SUBJECT}
|
||||
|
||||
|
||||
async def test_nats_disabled_starts_nothing_and_healthcheck_reports_off(
|
||||
patched_app: Any, monkeypatch: pytest.MonkeyPatch, events: list[str]
|
||||
) -> None:
|
||||
@ -136,7 +142,8 @@ async def test_nats_disabled_starts_nothing_and_healthcheck_reports_off(
|
||||
|
||||
assert response.status_code == 200 # noqa: PLR2004 - HTTP 200
|
||||
body = response.json()
|
||||
assert set(body) == {"uptime", "version", "nats"}
|
||||
assert set(body) == {"uptime", "version", "nats", "subjects"}
|
||||
assert body["subjects"] == _expected_subjects(patched_app)
|
||||
assert body["nats"] == "off"
|
||||
assert body["version"] == patched_app.settings.VERSION
|
||||
assert FakeNatsConnection.instances == []
|
||||
@ -154,7 +161,7 @@ async def test_minimal_environment_starts_and_healthcheck_answers(
|
||||
response = await client.get("/healthcheck")
|
||||
|
||||
assert response.status_code == 200 # noqa: PLR2004 - HTTP 200
|
||||
assert set(response.json()) == {"uptime", "version", "nats"}
|
||||
assert set(response.json()) == {"uptime", "version", "nats", "subjects"}
|
||||
|
||||
|
||||
async def test_nats_enabled_subscribes_worker_and_reports_ok(
|
||||
@ -195,12 +202,79 @@ async def test_nats_enabled_but_not_connected_reports_down(patched_app: Any, mon
|
||||
FakeNatsConnection.connect_on_start = False
|
||||
|
||||
async with running_app(patched_app) as client:
|
||||
assert (await client.get("/healthcheck")).json()["nats"] == "down"
|
||||
response = await client.get("/healthcheck")
|
||||
assert response.json()["nats"] == "down"
|
||||
assert response.json()["subjects"] == _expected_subjects(patched_app)
|
||||
assert "jwt" not in response.text.lower()
|
||||
(connection,) = FakeNatsConnection.instances
|
||||
assert connection.handlers == [] # подписка — только после подключения
|
||||
assert patched_app.app.state.worker is not None
|
||||
|
||||
|
||||
# --- subject'ы из настроек, JWT вне ответа, лог старта, передача result_subject воркеру ------------------------------
|
||||
CUSTOM_SUBJECTS = {
|
||||
"NATS_TABULAR_REQUEST_SUBJECT": "test.files.tabular_uploaded",
|
||||
"NATS_RESULT_SUBJECT": "test.files.tabular_parsed",
|
||||
}
|
||||
SECRET_JWT = "eyJ0eXAiOiJKV1QiLCJhbGciOiJlZDI1NTE5LW5rZXkifQ.healthcheck-secret.sig"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def custom_nats_settings(patched_app: Any, monkeypatch: pytest.MonkeyPatch) -> Any:
|
||||
"""Подменяет `settings.OPENBAO.NATS` копией с нестандартными subject'ами и непустым JWT."""
|
||||
nats = patched_app.settings.OPENBAO.NATS.model_copy(update={**CUSTOM_SUBJECTS, "NATS_JWT": SecretStr(SECRET_JWT)})
|
||||
monkeypatch.setattr(patched_app.settings.OPENBAO, "NATS", nats)
|
||||
return nats
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("nats_enabled", "state"), [(False, "off"), (True, "down")])
|
||||
async def test_healthcheck_reports_configured_subjects_and_hides_jwt(
|
||||
patched_app: Any,
|
||||
custom_nats_settings: Any,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
nats_enabled: bool,
|
||||
state: str,
|
||||
) -> None:
|
||||
monkeypatch.setattr(patched_app.settings, "NATS_ENABLED", nats_enabled)
|
||||
FakeNatsConnection.connect_on_start = False # state=down: NATS включён, но связи нет
|
||||
|
||||
async with running_app(patched_app) as client:
|
||||
response = await client.get("/healthcheck")
|
||||
|
||||
body = response.json()
|
||||
assert body["nats"] == state
|
||||
assert body["subjects"] == {
|
||||
"tabular_request": "test.files.tabular_uploaded",
|
||||
"result": "test.files.tabular_parsed",
|
||||
}
|
||||
assert SECRET_JWT not in response.text
|
||||
assert SECRET_JWT.split(".")[1] not in response.text
|
||||
|
||||
|
||||
async def test_startup_logs_configured_subjects(
|
||||
patched_app: Any, custom_nats_settings: Any, monkeypatch: pytest.MonkeyPatch, log_records: list[Any]
|
||||
) -> None:
|
||||
monkeypatch.setattr(patched_app.settings, "NATS_ENABLED", False)
|
||||
|
||||
async with running_app(patched_app):
|
||||
pass
|
||||
|
||||
messages = [record["message"] for record in log_records if "Subject'ы NATS" in record["message"]]
|
||||
assert len(messages) == 1
|
||||
assert "test.files.tabular_uploaded" in messages[0]
|
||||
assert "test.files.tabular_parsed" in messages[0]
|
||||
assert SECRET_JWT not in messages[0]
|
||||
|
||||
|
||||
async def test_build_worker_passes_configured_result_subject(
|
||||
patched_app: Any, custom_nats_settings: Any, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Воркер публикует ack и результат в настроенный `NATS_RESULT_SUBJECT`, а не в константу контракта."""
|
||||
nats, worker = patched_app._build_worker(FakeStorage([]))
|
||||
assert nats.settings.NATS_RESULT_SUBJECT == "test.files.tabular_parsed"
|
||||
assert worker._result_subject == "test.files.tabular_parsed" # noqa: SLF001 - проверяем проводку настроек
|
||||
|
||||
|
||||
# --- проверка бакета при старте (_connect_s3) ------------------------------------------------------------------------
|
||||
@pytest.fixture
|
||||
def log_records() -> Iterator[list[Any]]:
|
||||
|
||||
@ -8,6 +8,42 @@ from pydantic import ValidationError
|
||||
from src.core.config import Settings
|
||||
|
||||
SettingsFactory = Callable[..., Settings]
|
||||
SettingsFromEnvFactory = Callable[..., Settings]
|
||||
|
||||
# Минимальный набор обязательных переменных (как MINIMAL_ENV в conftest): окружение процесса, без .env
|
||||
MINIMAL_ENV_VARS = {
|
||||
"DEBUG": "true",
|
||||
"OPENBAO__MINIO__MINIO_URL": "localhost:9000",
|
||||
"OPENBAO__MINIO__MINIO_ADMIN_LOGIN": "minioadmin",
|
||||
"OPENBAO__MINIO__MINIO_ADMIN_PASSWORD": "minioadmin",
|
||||
"OPENBAO__MINIO__MINIO_BUCKET": "arm-ka",
|
||||
"OPENBAO__NATS__NATS_URL": "nats://localhost:4222",
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def settings_from_env(minimal_settings: SettingsFactory, monkeypatch: pytest.MonkeyPatch) -> SettingsFromEnvFactory:
|
||||
"""Фабрика `settings_from_env(**env)`: Settings из переменных окружения поверх минимального набора.
|
||||
|
||||
Окружение процесса очищено фикстурой `minimal_settings` (она же убирает `OPENBAO__*`); значение `None` — удалить
|
||||
переменную. `.env` не читается (`_env_file=None`).
|
||||
"""
|
||||
|
||||
def build(**env: str | None) -> Settings:
|
||||
for name, value in MINIMAL_ENV_VARS.items():
|
||||
monkeypatch.setenv(name, value)
|
||||
for name, value in env.items():
|
||||
if value is None:
|
||||
monkeypatch.delenv(name, raising=False)
|
||||
else:
|
||||
monkeypatch.setenv(name, value)
|
||||
return Settings(_env_file=None) # type: ignore[call-arg]
|
||||
|
||||
return build
|
||||
|
||||
|
||||
JWT = "eyJ0eXAiOiJKV1QiLCJhbGciOiJlZDI1NTE5LW5rZXkifQ.secret-payload.sig"
|
||||
SEED = "SUACSSL3UAHUDXKFSNVUZRF5UHPMWZ6BFDTJ7M6USDXIEDNPPQYYYCU3VY"
|
||||
|
||||
|
||||
def test_minimal_environment_gives_defaults(minimal_settings: SettingsFactory) -> None:
|
||||
@ -16,8 +52,8 @@ def test_minimal_environment_gives_defaults(minimal_settings: SettingsFactory) -
|
||||
assert settings.DEBUG is True
|
||||
assert settings.OPENBAO.MINIO.MINIO_BUCKET == "arm-ka"
|
||||
assert settings.OPENBAO.NATS.NATS_URL == "nats://localhost:4222"
|
||||
assert settings.OPENBAO.NATS.NATS_SEED == ""
|
||||
assert settings.OPENBAO.NATS.NATS_JWT == ""
|
||||
assert settings.OPENBAO.NATS.NATS_SEED.get_secret_value() == ""
|
||||
assert settings.OPENBAO.NATS.NATS_JWT.get_secret_value() == ""
|
||||
assert settings.NATS_ENABLED is True
|
||||
assert settings.SERVICE_NAME == "tabular-parser"
|
||||
assert settings.WORKER_CONCURRENCY == 1
|
||||
@ -48,3 +84,93 @@ def test_service_name_other_than_contract_value_is_rejected(minimal_settings: Se
|
||||
"""SERVICE_NAME — `service.name` конвертов контракта: другое значение — ошибка конфигурации при старте."""
|
||||
with pytest.raises(ValidationError, match="SERVICE_NAME"):
|
||||
minimal_settings("SERVICE_NAME=scan-gateway\n")
|
||||
|
||||
|
||||
def test_nats_subjects_default_to_contract_names(settings_from_env: SettingsFromEnvFactory) -> None:
|
||||
nats = settings_from_env().OPENBAO.NATS
|
||||
assert nats.NATS_TABULAR_REQUEST_SUBJECT == "files.tabular_uploaded"
|
||||
assert nats.NATS_IMAGE_REQUEST_SUBJECT == "files.img_uploaded"
|
||||
assert nats.NATS_RESULT_SUBJECT == "files.tabular_parsed"
|
||||
assert nats.request_subjects() == {"tabular": "files.tabular_uploaded", "image": "files.img_uploaded"}
|
||||
|
||||
|
||||
def test_nats_subjects_from_openbao(settings_from_env: SettingsFromEnvFactory) -> None:
|
||||
nats = settings_from_env(
|
||||
OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT="test.files.tabular_uploaded",
|
||||
OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT="test.files.img_uploaded",
|
||||
OPENBAO__NATS__NATS_RESULT_SUBJECT="test.files.tabular_parsed",
|
||||
).OPENBAO.NATS
|
||||
assert nats.request_subjects() == {"tabular": "test.files.tabular_uploaded", "image": "test.files.img_uploaded"}
|
||||
assert nats.NATS_RESULT_SUBJECT == "test.files.tabular_parsed"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"value",
|
||||
["", "files.*", "files.>", ">", "files..tabular", ".files", "files.", "files tabular", "files.tabular_uploaded "],
|
||||
)
|
||||
@pytest.mark.parametrize("key", ["NATS_TABULAR_REQUEST_SUBJECT", "NATS_IMAGE_REQUEST_SUBJECT", "NATS_RESULT_SUBJECT"])
|
||||
def test_invalid_subject_rejected(settings_from_env: SettingsFromEnvFactory, key: str, value: str) -> None:
|
||||
with pytest.raises(ValidationError, match=key):
|
||||
settings_from_env(**{f"OPENBAO__NATS__{key}": value})
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("first", "second"),
|
||||
[
|
||||
("NATS_TABULAR_REQUEST_SUBJECT", "NATS_IMAGE_REQUEST_SUBJECT"),
|
||||
("NATS_TABULAR_REQUEST_SUBJECT", "NATS_RESULT_SUBJECT"),
|
||||
("NATS_IMAGE_REQUEST_SUBJECT", "NATS_RESULT_SUBJECT"),
|
||||
],
|
||||
)
|
||||
def test_equal_subjects_rejected(settings_from_env: SettingsFromEnvFactory, first: str, second: str) -> None:
|
||||
"""Любая пара из трёх subject'ов не может совпадать; сообщение называет оба ключа."""
|
||||
with pytest.raises(ValidationError, match=f"{first} и {second} совпадают"):
|
||||
settings_from_env(
|
||||
**{f"OPENBAO__NATS__{first}": "test.same_subject", f"OPENBAO__NATS__{second}": "test.same_subject"}
|
||||
)
|
||||
|
||||
|
||||
def test_nats_secrets_hidden_in_repr(settings_from_env: SettingsFromEnvFactory) -> None:
|
||||
settings = settings_from_env(OPENBAO__NATS__NATS_JWT=JWT, OPENBAO__NATS__NATS_SEED=SEED)
|
||||
assert settings.OPENBAO.NATS.NATS_JWT.get_secret_value() == JWT
|
||||
assert JWT not in repr(settings) and SEED not in repr(settings)
|
||||
assert JWT not in str(settings) and SEED not in str(settings)
|
||||
|
||||
|
||||
MINIO_KEYS = ("MINIO_URL", "MINIO_ADMIN_LOGIN", "MINIO_ADMIN_PASSWORD", "MINIO_BUCKET", "MINIO_SECURE")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"missing",
|
||||
[
|
||||
("OPENBAO__NATS__NATS_URL",), # нет поля в группе NATS: в ошибке — вход группы NATS (с JWT)
|
||||
tuple(f"OPENBAO__MINIO__{key}" for key in MINIO_KEYS), # нет группы MINIO: в ошибке — вход всего OPENBAO
|
||||
],
|
||||
)
|
||||
def test_validation_error_hides_nats_secrets(
|
||||
settings_from_env: SettingsFromEnvFactory, missing: tuple[str, ...]
|
||||
) -> None:
|
||||
"""Ошибка «нет поля» печатает вход всей группы — JWT и seed туда попадать не должны (Review Focus 1).
|
||||
|
||||
pydantic в тексте ошибки усекает значения (`'SUACSSL3UA...PQYYYCU3VY'`), поэтому полный секрет там не появляется
|
||||
даже без скрытия; проверяем края секретов — они и попадают в текст при утечке.
|
||||
"""
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
settings_from_env(OPENBAO__NATS__NATS_JWT=JWT, OPENBAO__NATS__NATS_SEED=SEED, **dict.fromkeys(missing))
|
||||
text = str(exc_info.value)
|
||||
for secret in (JWT, SEED):
|
||||
for fragment in (secret[:10], secret[-10:]):
|
||||
assert fragment not in text
|
||||
|
||||
|
||||
def test_invalid_subject_error_hides_nats_secrets(settings_from_env: SettingsFromEnvFactory) -> None:
|
||||
"""Невалидный subject при заданных JWT/seed: ошибка валидации печатает вход группы NATS — секретов там нет."""
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
settings_from_env(
|
||||
OPENBAO__NATS__NATS_JWT=JWT, OPENBAO__NATS__NATS_SEED=SEED, OPENBAO__NATS__NATS_RESULT_SUBJECT="files.*"
|
||||
)
|
||||
text = str(exc_info.value)
|
||||
assert "NATS_RESULT_SUBJECT" in text
|
||||
for secret in (JWT, SEED):
|
||||
for fragment in (secret[:10], secret[-10:]):
|
||||
assert fragment not in text
|
||||
|
||||
@ -650,6 +650,15 @@ def test_request_subject_rejects_unknown_pipeline() -> None:
|
||||
request_subject("ocr")
|
||||
|
||||
|
||||
def test_request_subject_uses_environment_map() -> None:
|
||||
"""Карта `subjects` окружения (NATSSettings.request_subjects()) подменяет имена контракта по умолчанию."""
|
||||
subjects = {"tabular": "test.files.tabular_uploaded", "image": "test.files.img_uploaded"}
|
||||
assert request_subject("tabular", subjects) == "test.files.tabular_uploaded"
|
||||
assert request_subject("image", subjects) == "test.files.img_uploaded"
|
||||
with pytest.raises(ValueError, match="неизвестный pipeline"):
|
||||
request_subject("ocr", subjects)
|
||||
|
||||
|
||||
def test_request_subject_matches_example_pipeline() -> None:
|
||||
for name in ("request-tabular.json", "request-image.json"):
|
||||
request = RecognitionRequest.model_validate(_example(name))
|
||||
|
||||
@ -26,6 +26,7 @@ from src.core.nats import (
|
||||
NatsPublishError,
|
||||
Publisher,
|
||||
build_nats_auth,
|
||||
used_subjects,
|
||||
)
|
||||
from src.core.worker import STOP_TIMEOUT
|
||||
|
||||
@ -347,6 +348,26 @@ async def test_subscribe_requests_uses_tabular_subject_and_queue_group():
|
||||
await conn.stop()
|
||||
|
||||
|
||||
async def test_subscribe_requests_default_subject_comes_from_settings():
|
||||
world = _FakeWorld()
|
||||
settings = NATSSettings(
|
||||
NATS_URL="nats://localhost:4222", NATS_TABULAR_REQUEST_SUBJECT="test.files.tabular_uploaded"
|
||||
)
|
||||
conn = NatsConnection(settings, client_factory=world, sleep=_RecordingSleep())
|
||||
await conn.start()
|
||||
assert await conn.wait_ready(1)
|
||||
|
||||
await conn.subscribe_requests(_noop_handler)
|
||||
|
||||
assert world.clients[-1].subscriptions[0]["subject"] == "test.files.tabular_uploaded"
|
||||
await conn.stop()
|
||||
|
||||
|
||||
def test_used_subjects_tabular_parser() -> None:
|
||||
nats = NATSSettings(NATS_URL="nats://x:4222", NATS_TABULAR_REQUEST_SUBJECT="test.files.tabular_uploaded")
|
||||
assert used_subjects(nats) == {"tabular_request": "test.files.tabular_uploaded", "result": "files.tabular_parsed"}
|
||||
|
||||
|
||||
async def test_subscribe_requests_flushes_after_subscribe():
|
||||
"""T8.3: подписка считается активной, когда сервер обработал SUB (flush после subscribe)."""
|
||||
world = _FakeWorld()
|
||||
|
||||
@ -174,7 +174,14 @@ def log_records() -> Iterator[list[Any]]:
|
||||
|
||||
|
||||
class Env:
|
||||
def __init__(self, parser: Any = None, *, concurrency: int = 1, done_cache_size: int = 1000) -> None:
|
||||
def __init__(
|
||||
self,
|
||||
parser: Any = None,
|
||||
*,
|
||||
concurrency: int = 1,
|
||||
done_cache_size: int = 1000,
|
||||
result_subject: str | None = None,
|
||||
) -> None:
|
||||
self.events: list[tuple[str, str]] = []
|
||||
self.storage = FakeStorage(self.events)
|
||||
self.storage.objects[(BUCKET, SOURCE_KEY)] = b"PK\x03\x04 xlsx bytes"
|
||||
@ -188,6 +195,7 @@ class Env:
|
||||
concurrency=concurrency,
|
||||
done_cache_size=done_cache_size,
|
||||
clock=FakeClock(),
|
||||
**({} if result_subject is None else {"result_subject": result_subject}),
|
||||
)
|
||||
|
||||
def bodies(self) -> list[dict[str, Any]]:
|
||||
@ -217,6 +225,20 @@ def _assert_message_valid(body: dict[str, Any]) -> None:
|
||||
RecognitionResultMessage.model_validate(body)
|
||||
|
||||
|
||||
async def test_ack_and_result_go_to_configured_result_subject() -> None:
|
||||
env = await _make_env(result_subject="test.files.tabular_parsed")
|
||||
try:
|
||||
await env.worker.on_request(_msg(_request()))
|
||||
await env.worker.join()
|
||||
finally:
|
||||
await env.worker.stop(timeout=1)
|
||||
|
||||
assert [(subject, msg_id) for subject, _body, msg_id in env.publisher.published] == [
|
||||
("test.files.tabular_parsed", f"ack:{JOB_ID}:1"),
|
||||
("test.files.tabular_parsed", f"result:{JOB_ID}:1"),
|
||||
]
|
||||
|
||||
|
||||
# --- ack раньше результата ------------------------------------------------------------------------------------------
|
||||
async def test_ack_is_published_before_s3_read_and_result(env: Env) -> None:
|
||||
await env.worker.on_request(_msg(_request()))
|
||||
|
||||
2
openspec/changes/nats-env-config/.openspec.yaml
Normal file
2
openspec/changes/nats-env-config/.openspec.yaml
Normal file
@ -0,0 +1,2 @@
|
||||
schema: spec-driven
|
||||
created: 2026-10-12
|
||||
20
openspec/changes/nats-env-config/design.md
Normal file
20
openspec/changes/nats-env-config/design.md
Normal file
@ -0,0 +1,20 @@
|
||||
## Context
|
||||
|
||||
Дизайн сквозной: `../docs/openspec/changes/nats-env-config/design.md` (решения 1–7, R-1…R-6). Здесь — только то, что
|
||||
специфично для `tabular-parser`.
|
||||
|
||||
## Decisions
|
||||
|
||||
1. Сервис подписан на `tabular_request` (queue group `tabular-parser`), публикует ack и результат в `result`; в `/healthcheck.subjects` — `tabular_request`, `result`.
|
||||
2. Проверяются все три subject'а, включая неиспользуемые сервисом (docs design, решение 1 и R-3).
|
||||
3. Выбор subject'а и подписки получают имена из `settings.OPENBAO.NATS`; константы `contract/subjects.py` остаются
|
||||
умолчаниями полей настроек. Форма передачи (аргумент, объект subject'ов) — в плане.
|
||||
|
||||
## Risks / Trade-offs
|
||||
|
||||
См. docs design: разъехавшиеся настройки — тишина на шине; JWT без прав на заданные subject'ы — ошибка прав NATS в логе.
|
||||
|
||||
## Decisions made without discussion
|
||||
|
||||
Решения без обсуждения ведутся в сквозном `../docs/openspec/changes/nats-env-config/design.md` (R-1…R-6); решения,
|
||||
принятые при исполнении в этом репозитории, дописываются сюда с маркером `R-n` сквозной нумерации.
|
||||
40
openspec/changes/nats-env-config/proposal.md
Normal file
40
openspec/changes/nats-env-config/proposal.md
Normal file
@ -0,0 +1,40 @@
|
||||
## Why
|
||||
|
||||
Прод и тест будут работать с разными наборами subject'ов NATS и разными JWT; платформа передаёт в под только
|
||||
`OPENBAO__*`. Сейчас subject'ы — константы `api/src/contract/subjects.py`. Решение владельца 12.10.2026: subject'ы и
|
||||
JWT задаются по окружению секретами группы `OPENBAO__NATS__`, как в `RSHB/armka_file_worker`. Сквозной change —
|
||||
`../docs/openspec/changes/nats-env-config` (RSHB/arm-ka-docs#80); здесь — его часть для `tabular-parser` (RSHB/arm-tabular-parser#31).
|
||||
|
||||
Срез: вне плана срезов `09-delivery-plan.md` — подготовка к контуру. Таблицы DBML v5.17 не затрагиваются
|
||||
(`recognition_job.request_subject` по-прежнему хранит фактический subject публикации).
|
||||
|
||||
## What Changes
|
||||
|
||||
- `NATSSettings`: `NATS_TABULAR_REQUEST_SUBJECT` (`files.tabular_uploaded`), `NATS_IMAGE_REQUEST_SUBJECT`
|
||||
(`files.img_uploaded`), `NATS_RESULT_SUBJECT` (`files.tabular_parsed`); проверка всех трёх при старте (непустые, без
|
||||
пробелов, `*`, `>` и пустых токенов, попарно различны).
|
||||
- `NATS_JWT`, `NATS_SEED` — `SecretStr`; `Settings(hide_input_in_errors=True)`.
|
||||
- Сервис подписан на `tabular_request` (queue group `tabular-parser`), публикует ack и результат в `result` — по настройкам; константы `contract/subjects.py` — только умолчания.
|
||||
- Лог при старте и `/healthcheck.subjects` (`tabular_request`, `result`), независимо от состояния NATS.
|
||||
- Тесты roundtrip (`tests/integration/test_roundtrip.py`) задают собственные subject'ы через настройки (OQ-83).
|
||||
- `.env.example`, README.
|
||||
|
||||
## Non-goals
|
||||
|
||||
- `schema_version` контракта и копии схем не меняются (кроме побайтной синхронизации `description` из docs, если она
|
||||
изменится).
|
||||
- Queue group'ы — константы; механизм nkeys/JWT не меняется.
|
||||
- Пароли MinIO/SMB в `SecretStr` не переводятся.
|
||||
- Настройка через `OPENBAO__APP__*` не вводится.
|
||||
|
||||
## Capabilities
|
||||
|
||||
### New Capabilities
|
||||
|
||||
### Modified Capabilities
|
||||
- `tabular-recognition-service`: subject'ы обмена и учётные данные NATS — настройки окружения; `/healthcheck.subjects`.
|
||||
|
||||
## Impact
|
||||
|
||||
Файлы: `api/src/core/config.py`, `api/src/contract/subjects.py`, `api/src/core/nats.py`, `api/src/core/worker.py`, `api/src/app.py` (`/healthcheck`, лог старта), тесты (`tests/unit/test_config*.py`, `tests/unit/test_nats.py`, `tests/unit/test_app.py`, `tests/integration/test_roundtrip.py`), `api/.env.example`, README. Выкат обратно совместим (умолчания — текущие имена); набор окружения задаётся одинаково всем трём
|
||||
сервисам.
|
||||
@ -0,0 +1,43 @@
|
||||
## ADDED Requirements
|
||||
|
||||
### Requirement: Subject'ы и учётные данные NATS из настроек окружения
|
||||
Сервис SHALL брать subject'ы обмена из `OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT`, `NATS_IMAGE_REQUEST_SUBJECT`,
|
||||
`NATS_RESULT_SUBJECT` (умолчания `files.tabular_uploaded`, `files.img_uploaded`, `files.tabular_parsed`): подписан на `tabular_request` (queue group `tabular-parser`), публикует ack и результат в `result`.
|
||||
При старте SHALL проверяться все три имени (непустое, без пробельных символов, без `*` и `>`, без пустых токенов,
|
||||
попарно различны); при нарушении приложение MUST NOT стартовать. Значения `NATS_JWT` и `NATS_SEED` MUST NOT попадать в
|
||||
логи, `repr` настроек и тексты ошибок валидации. Используемые subject'ы SHALL писаться в лог при старте и отдаваться в
|
||||
`/healthcheck` объектом `subjects` с ключами `tabular_request`, `result` при любом значении `nats` (`ok | down | off`).
|
||||
|
||||
#### Scenario: Умолчания
|
||||
- **WHEN** приложение стартует без ключей `OPENBAO__NATS__NATS_*_SUBJECT`
|
||||
- **THEN** `/healthcheck.subjects` содержит имена `files.*` по умолчанию, обмен идёт по ним
|
||||
|
||||
#### Scenario: Subject'ы окружения
|
||||
- **WHEN** заданы `OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT=test.files.tabular_uploaded`,
|
||||
`OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT=test.files.img_uploaded`,
|
||||
`OPENBAO__NATS__NATS_RESULT_SUBJECT=test.files.tabular_parsed`
|
||||
- **THEN** подписки и публикации идут только по этим subject'ам, `/healthcheck.subjects` их показывает
|
||||
|
||||
#### Scenario: Недопустимый subject
|
||||
- **WHEN** `OPENBAO__NATS__NATS_RESULT_SUBJECT` совпадает с subject'ом запроса или содержит `>`
|
||||
- **THEN** приложение не стартует, ошибка называет ключ
|
||||
|
||||
#### Scenario: JWT не раскрывается
|
||||
- **WHEN** заданы `NATS_JWT` и `NATS_SEED`, а валидация настроек падает на другом поле
|
||||
- **THEN** текст ошибки и лог не содержат значений JWT и seed; `repr` настроек показывает их скрытыми
|
||||
|
||||
#### Scenario: Тесты на своих subject'ах
|
||||
- **WHEN** тесты roundtrip (`tests/integration/test_roundtrip.py`) запускаются при работающих контейнерах стенда на том же NATS
|
||||
- **THEN** тесты используют уникальные subject'ы из настроек и не получают и не забирают сообщения стенда
|
||||
|
||||
## MODIFIED Requirements
|
||||
|
||||
### Requirement: Соединение с NATS по правилам шаблона, без JetStream
|
||||
При пустых `NATS_SEED`/`NATS_JWT` сервис SHALL подключаться без учётных данных, иначе — с nkeys/JWT из `OPENBAO__NATS__NATS_SEED`/`NATS_JWT` (значения — секреты). Сервис MUST NOT
|
||||
требовать stream'ов или consumer'ов JetStream; при разрыве соединения SHALL переподключаться автоматически и восстанавливать
|
||||
подписку; `/healthcheck` SHALL отражать состояние (`nats = ok | down`; `off` — NATS выключен флагом
|
||||
`NATS_ENABLED=false`).
|
||||
|
||||
#### Scenario: NATS перезапущен
|
||||
- **WHEN** контейнер NATS перезапущен во время работы сервиса
|
||||
- **THEN** в течение 20 с после старта NATS подписка восстановлена, `nats = ok`, новый запрос обрабатывается
|
||||
18
openspec/changes/nats-env-config/tasks.md
Normal file
18
openspec/changes/nats-env-config/tasks.md
Normal file
@ -0,0 +1,18 @@
|
||||
## 1. Тесты
|
||||
|
||||
- [x] 1.1 Unit: умолчания subject'ов, чтение `OPENBAO__NATS__NATS_*_SUBJECT`, отказ на `*`/`>`/пробеле/пустом токене и
|
||||
совпадающих именах; `SecretStr` для `NATS_JWT`/`NATS_SEED`, отсутствие значений в ошибке валидации и `repr`
|
||||
- [x] 1.2 Unit: `/healthcheck.subjects` (`tabular_request`, `result`) при `nats = ok | down | off`; лог старта с subject'ами
|
||||
- [x] 1.3 Тесты roundtrip (`tests/integration/test_roundtrip.py`) — уникальные subject'ы через настройки
|
||||
|
||||
## 2. Реализация
|
||||
|
||||
- [x] 2.1 `core/config.py`: поля и валидация `NATSSettings`, `SecretStr`, `hide_input_in_errors=True`
|
||||
- [x] 2.2 Колбэки nkeys — `get_secret_value()`
|
||||
- [x] 2.3 Подписка и публикация по настройкам; константы `contract/subjects.py` — умолчания
|
||||
- [x] 2.4 `/healthcheck.subjects`, лог при старте
|
||||
|
||||
## 3. Документы
|
||||
|
||||
- [x] 3.1 `api/.env.example`, README (переменные NATS, `/healthcheck`)
|
||||
- [x] 3.2 `make check-full` зелёный
|
||||
Loading…
x
Reference in New Issue
Block a user