NATS: subject'ы и JWT по окружению из OpenBao (OPENBAO__NATS__*) (#32) #33

Merged
aep merged 6 commits from feat/32-nats-env-config into main 2026-10-13 12:54:57 +03:00
23 changed files with 589 additions and 83 deletions

View File

@ -9,7 +9,7 @@
распознанный документ), склеить и сохранить сырой ответ в S3 `raw/{job_id}.json` (в БД не регистрируется), очистить
горячую папку (только при `SCAN_CLEANUP=true`), нормализовать в `../docs/contracts/recognition-result.schema.json`
(`statementDocument` v1.2: по одному документу на таблицу `table_structure[i]`), опубликовать `files.tabular_parsed`
(`done` / `needs_check` / `error`). Таймаут → ошибка `TIMEOUT` по `min(reply_deadline, старт + SCAN_TIMEOUT_SECONDS=1200)`,
(по умолчанию; задаётся `OPENBAO__NATS__NATS_*_SUBJECT`) (`done` / `needs_check` / `error`). Таймаут → ошибка `TIMEOUT` по `min(reply_deadline, старт + SCAN_TIMEOUT_SECONDS=1200)`,
старт — запись файла во «Входящие». **Нет доступа к БД.**
## Правила

View File

@ -298,7 +298,10 @@ pydantic-валидацией; битый файл (код не из 4–5 ци
| `OPENBAO__MINIO__MINIO_BUCKET` | — (обязательна) | единственный бакет сервиса; запрос с другим `s3.bucket` -> `FILE_READ_ERROR` |
| `OPENBAO__MINIO__MINIO_SECURE` | `true` | https; для MinIO из compose — `false` |
| `OPENBAO__NATS__NATS_URL` | — (обязательна) | адрес NATS |
| `OPENBAO__NATS__NATS_SEED`, `OPENBAO__NATS__NATS_JWT` | пусто | nkeys/JWT платформы; оба пусты — без учётных данных; задан один — ошибка конфигурации |
| `OPENBAO__NATS__NATS_SEED`, `OPENBAO__NATS__NATS_JWT` | пусто | nkeys/JWT платформы (`SecretStr`, в логи и `/healthcheck` не попадают); оба пусты — без учётных данных; задан один — ошибка конфигурации |
| `OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT` | `files.img_uploaded` | subject запроса, на который подписан сервис; у ядра окружения то же значение |
| `OPENBAO__NATS__NATS_RESULT_SUBJECT` | `files.tabular_parsed` | subject, в который сервис публикует ack и итог |
| `OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT` | `files.tabular_uploaded` | запрос `tabular` (сервис его не слушает; значение участвует в проверке: три имени непусты, без пробелов, `*`, `>` и попарно различны, иначе сервис не стартует) |
| `OPENBAO__SMB__SMB_HOST`, `OPENBAO__SMB__SMB_PATH`, `OPENBAO__SMB__SMB_LOGIN`, `OPENBAO__SMB__SMB_PASSWORD` | пусто | шара горячих папок; обязательны только при `SCAN_HOTFOLDER_MODE=smb`; `SMB_PATH` — `share/dir` или полный UNC |
| `NATS_ENABLED` | `true` | `false` — воркер не запускается, подключения к NATS нет (`/healthcheck.nats = disabled`) |
| `PUBLISH_RETRIES` | `3` | повторы публикации ack/результата после первой попытки (всего до 4) |
@ -362,8 +365,8 @@ MinIO и NATS); те же значения подхватываются из `EN
Схемы и примеры — `../docs/contracts/`; их байт-в-байт копии лежат в `api/contracts/` (идентичность проверяет unit-тест
`api/tests/unit/test_contract.py`, входит в `make check`), pydantic-модели — `api/src/contract/`.
- Запрос: subject `files.img_uploaded`, подписка в queue group `scan-gateway`, `Nats-Msg-Id = <job_id>:<attempt>`.
- Ответы — subject `files.tabular_parsed`: ack `status=processing`, `Nats-Msg-Id = ack:<job_id>:<attempt>`; итог
- Запрос: subject `files.img_uploaded` (по умолчанию; `OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT`), подписка в queue group `scan-gateway`, `Nats-Msg-Id = <job_id>:<attempt>`.
- Ответы — subject `files.tabular_parsed` (по умолчанию; `OPENBAO__NATS__NATS_RESULT_SUBJECT`): ack `status=processing`, `Nats-Msg-Id = ack:<job_id>:<attempt>`; итог
`status = done | needs_check | error`, `Nats-Msg-Id = result:<job_id>:<attempt>`. `service.name = "scan-gateway"`.
- Ключи S3: исходник — из запроса (`s3.key`), сырой ответ (склейка файлов ИС) — `raw/{job_id}.json`, результат —
`results/{file_id}/{job_id}/result.json`. Бакет один.
@ -445,6 +448,7 @@ make golden-update # пересобрать эталоны нормализ
| `version` | `VERSION` |
| `nats` | `ok` — соединение установлено; `down` — подключение идёт, переподключение, сбой; `disabled` — `NATS_ENABLED=false` |
| `hotfolder` | `ok` — `HotFolder.healthy()` (корень существует и доступен); `down` — нет, проверка не уложилась в 2 с или упала |
| `subjects` | `{image_request, result}` — subject'ы NATS из настроек; отдаётся при любом состоянии NATS; JWT/seed в ответе нет |
Подключение к NATS идёт в фоне: HTTP отвечает сразу (`nats = down`), подписка появляется после подключения.
Результат проверки горячей папки кэшируется на 5 с.

View File

@ -26,8 +26,14 @@ OPENBAO__MINIO__MINIO_SECURE=false
OPENBAO__NATS__NATS_URL=nats://localhost:4222
OPENBAO__NATS__NATS_SEED=
OPENBAO__NATS__NATS_JWT=
# Subject'ы по окружению (прод и тест в одном NATS — разные наборы). Умолчания — имена контракта; у ядра и сервисов
# одного окружения значения обязаны совпадать. Сервис использует запрос `image` и ответ; три имени попарно различны,
# без пробелов, '*' и '>'. Фактические имена видны в логе старта и в /healthcheck.subjects.
# 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
# Включает подписку на files.img_uploaded. false — воркер не запускается (сервис поднимется без NATS).
# Включает подписку на subject запроса image (по умолчанию files.img_uploaded). false — воркер не запускается (сервис поднимется без NATS).
NATS_ENABLED=true
# PUBLISH_RETRIES — число повторов публикации ack/результата в NATS после первой попытки (при 3 — до 4 попыток всего);
# DONE_CACHE_SIZE — размер LRU обработанных job_id (защита от повторной доставки запроса)

View File

@ -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"],

View File

@ -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"],

View File

@ -1,11 +1,12 @@
"""Точка входа FastAPI сервиса scan-gateway (адаптер ИС Сканирование). Без БД и без фронта.
- `create_app(settings)` строит горячую папку (`create_hotfolder`, без обращения к ФС/шаре) и `HealthProvider`;
`GET /healthcheck` отдаёт `uptime`, `version`, `nats` (`ok`/`down`/`disabled`), `hotfolder` (`ok`/`down`).
`GET /healthcheck` отдаёт `uptime`, `version`, `nats` (`ok`/`down`/`disabled`), `hotfolder` (`ok`/`down`),
`subjects` (`image_request`, `result` — subject'ы NATS из настроек).
- lifespan: регистрация SMB-сессии (только `SCAN_HOTFOLDER_MODE=smb`), S3 (`S3Storage`), NATS (`NatsGateway`), воркер
(`Worker` с `OcrNormalizer`). Справочники нормализатора загружаются из `SCAN_RULES_DIR` в `create_app`: битый
справочник — `ValueError`, сервис не стартует. Подключение к NATS блокирует до доступности сервера, поэтому идёт
фоновой задачей: HTTP отвечает сразу (`nats = down`), подписка `files.img_uploaded` появляется после
фоновой задачей: HTTP отвечает сразу (`nats = down`), подписка на subject запроса `image` появляется после
подключения. Сбой этой задачи (например, `NatsConfigError`) — лог error с трассировкой и `nats = down`,
HTTP-сервер жив.
Остановка: отмена подключения, остановка воркера (задания в работе отменяются), закрытие NATS и S3.
@ -30,7 +31,7 @@ from src.core.config import Settings, get_settings
from src.core.exception_handlers import register_exception_handlers
from src.core.health import HealthProvider
from src.core.logger import setup_logger
from src.core.nats import NatsGateway
from src.core.nats import NatsGateway, used_subjects
from src.core.worker import Worker
from src.gateway import create_hotfolder
from src.normalizer import OcrNormalizer
@ -66,10 +67,12 @@ def create_app(settings: Settings | None = None) -> FastAPI:
up_time = datetime.now(UTC)
hotfolder = create_hotfolder(settings)
normalizer = OcrNormalizer(load_rules(settings.SCAN_RULES_DIR)) # битый справочник — сервис не стартует (D-7)
health = HealthProvider(nats_enabled=settings.NATS_ENABLED, hotfolder=hotfolder)
subjects = used_subjects(settings.OPENBAO.NATS)
health = HealthProvider(nats_enabled=settings.NATS_ENABLED, hotfolder=hotfolder, subjects=subjects)
@asynccontextmanager
async def lifespan(app: FastAPI):
logger.info("Subject'ы NATS. {subjects}", subjects=subjects)
smb_session = False
if settings.SCAN_HOTFOLDER_MODE == "smb":
smb = settings.OPENBAO.SMB

View File

@ -22,10 +22,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

View File

@ -11,16 +11,32 @@
"""
import os
import re
from functools import lru_cache
from typing import Any, Literal
from pydantic import NonNegativeFloat, PositiveFloat, PositiveInt, model_validator
from pydantic import (
NonNegativeFloat,
PositiveFloat,
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
@ -29,15 +45,57 @@ 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/JWT (локальная разработка, тесты)
NATS_JWT: str = ""
NATS_SEED: SecretStr = SecretStr("") # пусто — подключение к NATS без nkeys/JWT (локальная разработка, тесты)
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 SMBSettings(BaseSettings):
"""SMB-шара горячих папок ИС Сканирование. Нужна только при `SCAN_HOTFOLDER_MODE=smb` (там все поля обязательны,
проверяет `Settings`); в режиме local весь блок можно не задавать."""
model_config = SettingsConfigDict(hide_input_in_errors=True)
SMB_HOST: str = ""
SMB_PATH: str = ""
SMB_LOGIN: str = ""
@ -45,6 +103,8 @@ class SMBSettings(BaseSettings):
class OpenBAOSettings(BaseSettings):
model_config = SettingsConfigDict(hide_input_in_errors=True)
MINIO: MinioSettings
NATS: NATSSettings
SMB: SMBSettings = SMBSettings() # весь блок необязателен: в режиме local не требуется
@ -61,6 +121,8 @@ class OpenBAOSettings(BaseSettings):
class ServerSettings(BaseSettings):
model_config = SettingsConfigDict(hide_input_in_errors=True)
HOST: str = "0.0.0.0"
PORT: int = 8000
@ -99,6 +161,7 @@ class Settings(BaseSettings):
env_file_encoding="utf-8",
env_nested_delimiter="__",
nested_model_default_partial_update=False,
hide_input_in_errors=True,
)
@classmethod

View File

@ -1,4 +1,4 @@
"""Состояние сервиса для `GET /healthcheck`: поля `nats` и `hotfolder` (D-8; spec image-recognition-gateway).
"""Состояние сервиса для `GET /healthcheck`: поля `nats`, `hotfolder`, `subjects` (D-8; spec image-recognition-gateway).
- `nats`: `disabled` при `NATS_ENABLED=false`; `ok`, если соединение установлено (`is_connected`), иначе `down` — в том
числе пока идёт первое подключение, во время переподключения и после сбоя фоновой задачи подключения.
@ -11,8 +11,8 @@
import asyncio
import time
from collections.abc import Callable
from typing import Literal, Protocol
from collections.abc import Callable, Mapping
from typing import Any, Literal, Protocol
from loguru import logger
@ -37,11 +37,12 @@ class ConnectionState(Protocol):
class HealthProvider:
"""Поля `nats`/`hotfolder` для `/healthcheck`. `nats` присваивается, когда соединение создано (lifespan)."""
def __init__(
def __init__( # noqa: PLR0913 - именованные параметры: подмены для тестов и тайминги
self,
*,
nats_enabled: bool,
hotfolder: HotFolder | None,
subjects: Mapping[str, str],
check_timeout: float = HOTFOLDER_CHECK_TIMEOUT,
cache_ttl: float = HOTFOLDER_CACHE_TTL,
clock: Callable[[], float] = time.monotonic,
@ -49,6 +50,7 @@ class HealthProvider:
self.nats: ConnectionState | None = None
self._nats_enabled = nats_enabled
self._hotfolder = hotfolder
self._subjects = dict(subjects)
self._check_timeout = check_timeout
self._cache_ttl = cache_ttl
self._clock = clock
@ -86,8 +88,13 @@ class HealthProvider:
self._cached, self._cached_at = status, self._clock()
return status
async def snapshot(self) -> dict[str, str]:
return {"nats": self.nats_status(), "hotfolder": await self.hotfolder_status()}
async def snapshot(self) -> dict[str, Any]:
"""`subjects` отдаётся при любом состоянии NATS (имена subject'ов — не секрет)."""
return {
"nats": self.nats_status(),
"hotfolder": await self.hotfolder_status(),
"subjects": dict(self._subjects),
}
def _consume_exception(task: "asyncio.Task[bool]") -> None:

View File

@ -8,9 +8,10 @@
поэтому `connect()` возвращается только при успехе (вызывающий оборачивает его в задачу или `wait_for`).
После переподключения nats-py сам отправляет SUB для всех подписок с прежней queue group — подписку повторно
оформлять не нужно.
- `subscribe_requests(cb)` — подписка на `files.img_uploaded` в queue group `scan-gateway`: из нескольких
инстансов сообщение получает ровно один. `cb` получает исходное сообщение nats-py (`msg.data`, `msg.headers`);
исключения `cb` логируются, подписка живёт.
- `subscribe_requests(cb)` — подписка на subject запроса `image` из настроек
(`OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT`, по умолчанию `files.img_uploaded`) в queue group `scan-gateway`:
из нескольких инстансов сообщение получает ровно один. `cb` получает исходное сообщение nats-py
(`msg.data`, `msg.headers`); исключения `cb` логируются, подписка живёт.
- `publish(subject, payload, msg_id)` — публикация с заголовком `Nats-Msg-Id` и `flush` (подтверждение сервером) с
повторами на временных ошибках (первая попытка + до `PUBLISH_RETRIES` повторов); исчерпаны — `PublishFailed`.
Воркер логирует error и идёт дальше: ядро закроет задание по `reply_deadline` (D-7).
@ -31,9 +32,7 @@ from nats.aio.client import Client as NATS
from nats.aio.msg import Msg
from src.contract import (
IMAGE_REQUEST_SUBJECT,
NATS_MSG_ID_HEADER,
RESULT_SUBJECT,
MessageStatus,
RecognitionResultMessage,
ack_msg_id,
@ -92,8 +91,8 @@ class PublishFailed(NatsError):
def build_nats_auth(settings: NATSSettings) -> dict[str, Any]:
"""Параметры аутентификации для `Client.connect`: `{}` без учётных данных или колбэки nkeys/JWT шаблона."""
seed = settings.NATS_SEED.strip()
jwt = settings.NATS_JWT.strip()
seed = settings.NATS_SEED.get_secret_value().strip()
jwt = settings.NATS_JWT.get_secret_value().strip()
if not seed and not jwt:
return {}
if not seed or not jwt:
@ -116,6 +115,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`: подписан на запрос `image`, публикует ответ."""
return {"image_request": settings.NATS_IMAGE_REQUEST_SUBJECT, "result": settings.NATS_RESULT_SUBJECT}
def message_msg_id(message: RecognitionResultMessage, attempt: int) -> str:
"""`Nats-Msg-Id` конверта: `ack:<job_id>:<attempt>` для `processing`, иначе `result:<job_id>:<attempt>`."""
if message.status is MessageStatus.PROCESSING:
@ -209,15 +213,17 @@ class NatsGateway:
# --- подписка ---------------------------------------------------------------------------------------------
async def subscribe_requests(self, cb: MessageHandler, queue: str = REQUEST_QUEUE_GROUP) -> Any:
"""Подписка на `files.img_uploaded` в queue group `queue`; возвращает подписку nats-py после
"""Подписка на subject запроса `image` из настроек (по умолчанию `files.img_uploaded`) в queue group
`queue`; возвращает подписку nats-py после
подтверждения сервером (`flush`).
`cb` получает исходное `Msg` (`data`, `headers`); его исключения логируются, подписка не падает.
`cb` должен возвращаться быстро: nats-py вызывает колбэки одной подписки последовательно, долгая обработка
(ожидание ИС) задерживает следующие запросы — её выносят в отдельную задачу.
"""
subject = self._settings.OPENBAO.NATS.NATS_IMAGE_REQUEST_SUBJECT
if self._client is None:
raise NatsNotConnectedError(f"Нет соединения с NATS: подписка на {IMAGE_REQUEST_SUBJECT} невозможна")
raise NatsNotConnectedError(f"Нет соединения с NATS: подписка на {subject} невозможна")
async def safe_cb(msg: Msg) -> None:
try:
@ -230,11 +236,9 @@ class NatsGateway:
error=repr(exc),
)
subscription = await self._client.subscribe(IMAGE_REQUEST_SUBJECT, queue=queue, cb=safe_cb)
subscription = await self._client.subscribe(subject, queue=queue, cb=safe_cb)
await self._client.flush(timeout=self._flush_timeout) # SUB принят сервером: подписка действует по возврату
logger.info(
"Подписка NATS активна. subject={subject}, queue={queue}", subject=IMAGE_REQUEST_SUBJECT, queue=queue
)
logger.info("Подписка NATS активна. subject={subject}, queue={queue}", subject=subject, queue=queue)
return subscription
# --- публикация -------------------------------------------------------------------------------------------
@ -292,10 +296,11 @@ class NatsGateway:
) from exc
async def publish_result(self, message: RecognitionResultMessage, attempt: int) -> None:
"""Публикует конверт в `files.tabular_parsed`: тело — `message.to_wire()` (JSON, UTF-8), `Nats-Msg-Id` —
"""Публикует конверт в subject ответа из настроек (по умолчанию `files.tabular_parsed`):
тело — `message.to_wire()` (JSON, UTF-8), `Nats-Msg-Id` —
`ack:…` для `processing`, `result:…` для итога; `attempt` — из запроса. Ошибки — как у `publish`."""
payload = json.dumps(message.to_wire(), ensure_ascii=False).encode("utf-8")
await self.publish(RESULT_SUBJECT, payload, message_msg_id(message, attempt))
await self.publish(self._settings.OPENBAO.NATS.NATS_RESULT_SUBJECT, payload, message_msg_id(message, attempt))
# --- колбэки nats-py --------------------------------------------------------------------------------------
async def _on_error(self, exc: Exception) -> None:

View File

@ -107,7 +107,7 @@ class ObjectStorage(Protocol):
class ResultPublisher(Protocol):
"""Часть `NatsGateway`: публикация конверта в `files.tabular_parsed` с `Nats-Msg-Id` и повторами."""
"""Часть `NatsGateway`: публикация конверта в subject ответа с `Nats-Msg-Id` и повторами."""
async def publish_result(self, message: RecognitionResultMessage, attempt: int) -> None: ...
@ -181,7 +181,7 @@ class _WaitResult:
class Worker:
"""Обработка запросов `files.img_uploaded` (см. докстринг модуля).
"""Обработка запросов на subject запроса `image` (см. докстринг модуля).
Зависимости внедряются: `storage` (S3), `publisher` (NATS), `hotfolder`, `normalizer`; `clock` — часы UTC для
дедлайна (R-20), `sleep` — пауза между опросами (подмены для тестов).

View File

@ -3,7 +3,9 @@
Недоступная инфраструктура — падение с понятным сообщением, не skip. Общий контейнер NATS не перезапускается:
сценарий переподключения покрыт unit-тестом колбэков (tests/unit/test_nats.py). Чужой трафик на тех же subject'ах
отсекается уникальным `job_id` теста; ключи MinIO — под уникальным префиксом `itest/<uuid>/` и удаляются.
отсекается уникальным `job_id` теста; кроме того, тесты работают на своих subject'ах прогона (`TEST_SUBJECTS`, суффикс
`RUN`), поэтому контейнеры стенда на умолчательных `files.*` их не видят; ключи MinIO — под уникальным префиксом
`itest/<uuid>/` и удаляются.
"""
import asyncio
@ -18,22 +20,23 @@ import pytest
from nats.aio.client import Client as NATS
from nats.aio.msg import Msg
from src.contract import (
IMAGE_REQUEST_SUBJECT,
NATS_MSG_ID_HEADER,
RESULT_SUBJECT,
ServiceInfo,
build_ack_message,
)
from src.contract import NATS_MSG_ID_HEADER, ServiceInfo, build_ack_message
from src.core.clients import S3ObjectNotFoundError, S3Storage
from src.core.config import Settings, get_settings
from src.core.nats import REQUEST_QUEUE_GROUP, NatsGateway
from src.core.nats import REQUEST_QUEUE_GROUP, NatsGateway, used_subjects
pytestmark = pytest.mark.integration
CONNECT_WAIT = 5.0 # секунд на подключение к NATS (nats-py ждёт сервер бесконечно — ограничиваем снаружи)
DELIVERY_WAIT = 5.0
# Уникальные subject'ы прогона (OPENBAO__NATS__NATS_*_SUBJECT): общий NATS стенда слушают другие экземпляры сервисов
RUN = uuid.uuid4().hex[:8]
TEST_SUBJECTS = {
"image_request": f"itest.{RUN}.files.img_uploaded",
"result": f"itest.{RUN}.files.tabular_parsed",
}
def _require_port(host: str, port: int, what: str) -> None:
try:
@ -44,8 +47,12 @@ def _require_port(host: str, port: int, what: str) -> None:
@pytest.fixture
def settings() -> Settings:
def settings(monkeypatch: pytest.MonkeyPatch) -> Settings:
monkeypatch.setenv("OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT", TEST_SUBJECTS["image_request"])
monkeypatch.setenv("OPENBAO__NATS__NATS_RESULT_SUBJECT", TEST_SUBJECTS["result"])
get_settings.cache_clear()
settings = get_settings()
assert used_subjects(settings.OPENBAO.NATS) == TEST_SUBJECTS
nats_url = urlparse(settings.OPENBAO.NATS.NATS_URL)
_require_port(nats_url.hostname or "127.0.0.1", nats_url.port or 4222, "NATS")
host, _, port = settings.OPENBAO.MINIO.MINIO_URL.partition(":")
@ -102,16 +109,16 @@ async def test_subscribe_and_publish_with_msg_id_without_jetstream(settings: Set
received: list[Msg] = []
async def on_request(msg: Msg) -> None:
if json.loads(msg.data).get("job_id") == job_id: # чужие сообщения на общем subject пропускаем
if json.loads(msg.data).get("job_id") == job_id: # страховка: на своих subject'ах чужого трафика нет
received.append(msg)
await subscriber.subscribe_requests(on_request)
await publisher.publish(IMAGE_REQUEST_SUBJECT, json.dumps({"job_id": job_id}).encode(), f"{job_id}:1")
await publisher.publish(TEST_SUBJECTS["image_request"], json.dumps({"job_id": job_id}).encode(), f"{job_id}:1")
await _wait_until(lambda: len(received) >= 1)
assert len(received) == 1
[msg] = received
assert msg.subject == IMAGE_REQUEST_SUBJECT
assert msg.subject == TEST_SUBJECTS["image_request"]
assert msg.headers is not None
assert msg.headers[NATS_MSG_ID_HEADER] == f"{job_id}:1"
@ -138,7 +145,7 @@ async def test_queue_group_delivers_each_message_to_exactly_one_member(settings:
total = 40
sent = {f"{run_id}-{n}:1" for n in range(total)}
for msg_id in sorted(sent):
await publisher.publish(IMAGE_REQUEST_SUBJECT, json.dumps({"run_id": run_id}).encode(), msg_id)
await publisher.publish(TEST_SUBJECTS["image_request"], json.dumps({"run_id": run_id}).encode(), msg_id)
await _wait_until(lambda: len(deliveries) >= total)
await asyncio.sleep(0.3) # лишние (дублирующие) доставки успели бы прийти
@ -161,7 +168,7 @@ async def test_publish_result_reaches_result_subject_with_ack_msg_id(settings: S
received.append(msg)
try:
await listener.subscribe(RESULT_SUBJECT, cb=on_result)
await listener.subscribe(TEST_SUBJECTS["result"], cb=on_result)
await listener.flush()
message = build_ack_message(job_id, file_id, ServiceInfo(name="scan-gateway", version="1.0.0"))
await gateway.publish_result(message, attempt=1)

View File

@ -22,14 +22,12 @@
ответа, в порядке возрастания имени, с телами байт в байт; входящий файл совпадает с объектом S3. Готовность набора
— по settle (`SETTLE_SECONDS`, 0.2 с).
Общая инфраструктура: NATS и MinIO используют и другие сервисы workspace. Чужие конверты на `files.tabular_parsed`
отсекаются по `job_id` теста; ключи MinIO уникальны (`uploads/itest-<uuid>/…`, `raw/<job_id>.json`, `results/…`) и
удаляются.
Подписка шлюза идёт в queue group `scan-gateway`: если в этом NATS живёт другой экземпляр scan-gateway, он может
забрать запрос теста (тест тогда упадёт по таймауту ожидания ack). Перед запуском занятость группы можно проверить
мониторингом NATS: `curl -s 'http://127.0.0.1:8222/subsz?subs=1'` — подписчиков `files.img_uploaded` быть
не должно. Автоматической защиты в тесте нет намеренно (она была бы гонкой). Недоступная инфраструктура — падение
теста с понятным сообщением, не skip; нет репозитория fixtures — skip сценариев на реальных данных (маркер `fixtures`).
Общая инфраструктура: NATS и MinIO используют и другие сервисы workspace. Тест слушает ответы на своём
subject'е прогона; конверты отсекаются по `job_id` теста; ключи MinIO уникальны (`uploads/itest-<uuid>/…`,
`raw/<job_id>.json`, `results/…`) и удаляются.
Подписка шлюза идёт в queue group `scan-gateway`. Тест работает на своих subject'ах прогона (`TEST_SUBJECTS`), поэтому
экземпляры стенда на умолчательных `files.*` его запросы не получают. Недоступная инфраструктура — падение теста с
понятным сообщением, не skip; нет репозитория fixtures — skip сценариев на реальных данных (маркер `fixtures`).
"""
import asyncio
@ -56,9 +54,10 @@ from nats.aio.client import Client as NATS
from nats.aio.msg import Msg
from src.app import create_app
from src.contract import CONTRACT_VERSION, IMAGE_REQUEST_SUBJECT, NATS_MSG_ID_HEADER, RESULT_SUBJECT, request_msg_id
from src.contract import CONTRACT_VERSION, NATS_MSG_ID_HEADER, request_msg_id
from src.core.clients import JSON_CONTENT_TYPE, S3Storage
from src.core.config import Settings, get_settings
from src.core.nats import used_subjects
from src.scan_emulator import EmulatorSettings, run
pytestmark = [
@ -76,6 +75,13 @@ CONTRACTS_DIR = API_DIR / "contracts"
REAL_FIXTURES = API_DIR.parent.parent / "fixtures" / "ocr"
GOLDEN_JOB_ID = "3f1c2b7a-0000-4000-8000-000000000001" # условный job_id эталонов (`tests/golden_update.py`)
# Уникальные subject'ы прогона (OPENBAO__NATS__NATS_*_SUBJECT): шлюз теста и слушатель работают на них, контейнеры
# стенда на умолчательных `files.*` запросы теста не получают и не отвечают на них
RUN = uuid.uuid4().hex[:8]
TEST_SUBJECTS = {
"image_request": f"itest.{RUN}.files.img_uploaded",
"result": f"itest.{RUN}.files.tabular_parsed",
}
CONNECT_WAIT = 10.0 # секунд на подключение к NATS (тест и фоновая задача приложения)
SCENARIO_WAIT = 20.0 # секунд на ожидание конвертов одного сценария
QUIET_PERIOD = 1.0 # секунд тишины после повторного запроса: второго результата быть не должно
@ -208,7 +214,7 @@ class Roundtrip:
"""Публикация как в ядре: `files.img_uploaded`, `Nats-Msg-Id = <job_id>:<attempt>`."""
payload = json.dumps(body, ensure_ascii=False).encode("utf-8")
headers = {NATS_MSG_ID_HEADER: request_msg_id(body["job_id"], body["attempt"])}
await self.publisher.publish(IMAGE_REQUEST_SUBJECT, payload, headers=headers)
await self.publisher.publish(TEST_SUBJECTS["image_request"], payload, headers=headers)
await self.publisher.flush()
# --- конверты -------------------------------------------------------------------------------------------------
@ -357,7 +363,11 @@ def _settings(monkeypatch: pytest.MonkeyPatch, root: Path, *, cleanup: bool) ->
monkeypatch.setenv("SCAN_TIMEOUT_SECONDS", str(TIMEOUT_SECONDS))
monkeypatch.setenv("SCAN_CLEANUP", "true" if cleanup else "false")
monkeypatch.setenv("NATS_ENABLED", "true")
monkeypatch.setenv("OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT", TEST_SUBJECTS["image_request"])
monkeypatch.setenv("OPENBAO__NATS__NATS_RESULT_SUBJECT", TEST_SUBJECTS["result"])
get_settings.cache_clear()
settings = get_settings()
assert used_subjects(settings.OPENBAO.NATS) == TEST_SUBJECTS
parsed = urlparse(settings.OPENBAO.NATS.NATS_URL)
_require_port(parsed.hostname or "127.0.0.1", parsed.port or 4222, "NATS")
host, _, port = settings.OPENBAO.MINIO.MINIO_URL.partition(":")
@ -366,7 +376,7 @@ def _settings(monkeypatch: pytest.MonkeyPatch, root: Path, *, cleanup: bool) ->
async def _start_gateway(stack: AsyncExitStack, app: FastAPI) -> None:
"""Lifespan приложения; возврат — когда шлюз подписан на `files.img_uploaded`."""
"""Lifespan приложения; возврат — когда шлюз подписан на subject запроса."""
await stack.enter_async_context(app.router.lifespan_context(app))
[connect_task] = [t for t in asyncio.all_tasks() if t.get_name() == "nats-connect"]
# задача завершается после SUB с flush: дальше подписка шлюза действует на сервере
@ -432,7 +442,7 @@ async def _roundtrip(
rt.received.append(msg)
# core NATS без JetStream не хранит сообщения: слушатель подписан до первой публикации запроса
await client.subscribe(RESULT_SUBJECT, cb=on_result)
await client.subscribe(TEST_SUBJECTS["result"], cb=on_result)
await client.flush()
await _start_gateway(stack, rt.app)
_start_emulator(stack, settings, root, tmp_path / "fixtures")

View File

@ -52,11 +52,69 @@ async def test_healthcheck_without_lifespan_has_all_fields(local_env: Path) -> N
assert response.status_code == 200
body = response.json()
assert set(body) == {"uptime", "version", "nats", "hotfolder"}
assert set(body) == {"uptime", "version", "nats", "hotfolder", "subjects"}
assert body["version"] == get_settings().VERSION
assert body["subjects"] == {"image_request": "files.img_uploaded", "result": "files.tabular_parsed"}
datetime.fromisoformat(body["uptime"])
async def test_healthcheck_subjects_from_env_without_jwt(local_env: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT", "test.files.img_uploaded")
monkeypatch.setenv("OPENBAO__NATS__NATS_RESULT_SUBJECT", "test.files.tabular_parsed")
monkeypatch.setenv("OPENBAO__NATS__NATS_JWT", "secret-jwt-value")
get_settings.cache_clear()
app = create_app(get_settings())
async with _client(app) as client:
response = await client.get("/healthcheck")
assert response.json()["subjects"] == {
"image_request": "test.files.img_uploaded",
"result": "test.files.tabular_parsed",
}
assert "secret-jwt-value" not in response.text
async def test_healthcheck_nats_down_hides_jwt(local_env: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""NATS включён, но не подключён (`down`): значения JWT в теле `/healthcheck` нет."""
monkeypatch.setenv("NATS_ENABLED", "true")
monkeypatch.setenv("OPENBAO__NATS__NATS_JWT", "secret-jwt-value")
never = asyncio.Event()
async def fake_connect(self: NatsGateway) -> bool:
await never.wait()
return True
async def fake_close(self: NatsGateway) -> None:
return None
monkeypatch.setattr(NatsGateway, "connect", fake_connect)
monkeypatch.setattr(NatsGateway, "close", fake_close)
get_settings.cache_clear()
app = create_app(get_settings())
async with app.router.lifespan_context(app), _client(app) as client:
response = await client.get("/healthcheck")
assert response.json()["nats"] == "down"
assert "secret-jwt-value" not in response.text
async def test_startup_logs_nats_subjects(
local_env: Path, monkeypatch: pytest.MonkeyPatch, log_lines: list[str]
) -> None:
monkeypatch.setenv("OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT", "test.files.img_uploaded")
monkeypatch.setenv("OPENBAO__NATS__NATS_RESULT_SUBJECT", "test.files.tabular_parsed")
get_settings.cache_clear()
app = create_app(get_settings())
async with app.router.lifespan_context(app):
pass
line = next(line for line in log_lines if "Subject'ы NATS" in line)
assert "test.files.img_uploaded" in line and "test.files.tabular_parsed" in line
async def test_start_in_local_mode(local_env: Path) -> None:
"""Spec «Старт в режиме локальных папок»: 200 с uptime, version, nats, hotfolder = ok; OPENBAO__SMB__* не нужны."""
app = create_app(get_settings())

View File

@ -1,3 +1,6 @@
import os
from collections.abc import Callable
import pytest
from pydantic import ValidationError
@ -35,8 +38,8 @@ def test_defaults_local_mode_without_smb_and_nats_credentials(monkeypatch, minim
assert s.SCAN_TIMEOUT_SECONDS == 1200
assert s.SCAN_MAX_INFLIGHT == 4
assert s.OPENBAO.SMB.SMB_HOST == ""
assert s.OPENBAO.NATS.NATS_SEED == ""
assert s.OPENBAO.NATS.NATS_JWT == ""
assert s.OPENBAO.NATS.NATS_SEED.get_secret_value() == ""
assert s.OPENBAO.NATS.NATS_JWT.get_secret_value() == ""
assert not hasattr(s.OPENBAO, "DATABASE")
@ -171,7 +174,7 @@ def test_env_test_file_loads_for_tests():
assert s.OPENBAO.MINIO.MINIO_URL == "127.0.0.1:9000"
assert s.OPENBAO.MINIO.MINIO_SECURE is False
assert s.OPENBAO.NATS.NATS_URL == "nats://127.0.0.1:4222"
assert s.OPENBAO.NATS.NATS_SEED == ""
assert s.OPENBAO.NATS.NATS_SEED.get_secret_value() == ""
def test_env_example_loads(monkeypatch):
@ -185,4 +188,127 @@ def test_env_example_loads(monkeypatch):
assert not hasattr(s, "SCAN_TMP_DIR")
assert s.SCAN_SETTLE_SECONDS == 2
assert s.OPENBAO.MINIO.MINIO_BUCKET == "arm-ka"
assert s.OPENBAO.NATS.NATS_JWT == ""
assert s.OPENBAO.NATS.NATS_JWT.get_secret_value() == ""
# --- настройки NATS: subject'ы и секреты окружения OPENBAO__NATS__* (change nats-env-config) ----------------------
REQUIRED_OPENBAO_ENV = {
"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(monkeypatch, tmp_path) -> Callable[..., Settings]:
"""Фабрика `Settings(_env_file=None)`: минимальный набор обязательных `OPENBAO__*` плюс `env` поверх; значение
`None` удаляет переменную. Файл окружения не читается: иначе удалённая переменная осталась бы в нём."""
def build(**env: str | None) -> Settings:
_clean_env(monkeypatch)
for key in [key for key in os.environ if key.startswith("OPENBAO__")]:
monkeypatch.delenv(key)
values = {**REQUIRED_OPENBAO_ENV, "SCAN_HOTFOLDER_PATH": str(tmp_path), **env}
for key, value in values.items():
if value is None:
monkeypatch.delenv(key, raising=False)
else:
monkeypatch.setenv(key, value)
return Settings(_env_file=None) # type: ignore[call-arg]
return build
JWT = "eyJ0eXAiOiJKV1QiLCJhbGciOiJlZDI1NTE5LW5rZXkifQ.secret-payload.sig"
SEED = "SUACSSL3UAHUDXKFSNVUZRF5UHPMWZ6BFDTJ7M6USDXIEDNPPQYYYCU3VY"
def test_nats_subjects_default_to_contract_names(settings_from_env) -> 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) -> 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"
SUBJECT_KEYS = ("NATS_TABULAR_REQUEST_SUBJECT", "NATS_IMAGE_REQUEST_SUBJECT", "NATS_RESULT_SUBJECT")
@pytest.mark.parametrize("key", SUBJECT_KEYS)
@pytest.mark.parametrize(
"value",
["", "files.*", "files.>", ">", "files..tabular", ".files", "files.", "files tabular", "files.tabular_uploaded "],
)
def test_invalid_subject_rejected(settings_from_env, 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, first: str, second: str) -> None:
with pytest.raises(ValidationError, match=f"{first} и {second} совпадают"):
settings_from_env(**{f"OPENBAO__NATS__{first}": "test.same", f"OPENBAO__NATS__{second}": "test.same"})
def test_invalid_subject_error_hides_nats_secrets(settings_from_env) -> None:
"""Невалидный subject при заданных JWT и seed: секреты не попадают в текст ошибки валидации."""
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)
for secret in (JWT, SEED):
assert secret not in text
assert secret[:10] not in text
assert secret[-10:] not in text
def test_nats_secrets_hidden_in_repr(settings_from_env) -> 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, missing: tuple[str, ...]) -> None:
"""Ошибка «нет поля» печатает вход всей группы — JWT и seed туда попадать не должны (Review Focus 1).
pydantic урезает длинные значения в тексте ошибки: видны начало и конец секрета, а не он целиком. Поэтому проверяем
и голову, и хвост каждого секрета (без скрытия входа в тексте видны оба куска)."""
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):
assert secret not in text
assert secret[:10] not in text
assert secret[-10:] not in text

View File

@ -640,6 +640,15 @@ def test_request_subject_rejects_unknown_pipeline() -> None:
request_subject("ocr")
def test_request_subject_uses_environment_map() -> None:
"""Карта `pipeline → subject` окружения (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))

View File

@ -11,6 +11,8 @@ from src.gateway.hotfolder import AnswerFile, LocalHotFolder
pytestmark = pytest.mark.usefixtures("log_lines")
SUBJECTS = {"image_request": "files.img_uploaded", "result": "files.tabular_parsed"}
class Connection:
def __init__(self, connected: bool) -> None:
@ -50,17 +52,17 @@ class Clock:
async def test_local_root_exists_is_ok(tmp_path: Path) -> None:
health = HealthProvider(nats_enabled=False, hotfolder=LocalHotFolder(tmp_path))
assert await health.snapshot() == {"nats": "disabled", "hotfolder": "ok"}
health = HealthProvider(nats_enabled=False, hotfolder=LocalHotFolder(tmp_path), subjects=SUBJECTS)
assert await health.snapshot() == {"nats": "disabled", "hotfolder": "ok", "subjects": SUBJECTS}
async def test_missing_root_is_down(tmp_path: Path) -> None:
health = HealthProvider(nats_enabled=False, hotfolder=LocalHotFolder(tmp_path / "missing"))
health = HealthProvider(nats_enabled=False, hotfolder=LocalHotFolder(tmp_path / "missing"), subjects=SUBJECTS)
assert await health.hotfolder_status() == "down"
async def test_no_hotfolder_is_down() -> None:
assert await HealthProvider(nats_enabled=False, hotfolder=None).hotfolder_status() == "down"
assert await HealthProvider(nats_enabled=False, hotfolder=None, subjects=SUBJECTS).hotfolder_status() == "down"
@pytest.mark.parametrize(
@ -74,7 +76,7 @@ async def test_no_hotfolder_is_down() -> None:
],
)
def test_nats_status(enabled: bool, connection: Connection | None, expected: str) -> None:
health = HealthProvider(nats_enabled=enabled, hotfolder=None)
health = HealthProvider(nats_enabled=enabled, hotfolder=None, subjects=SUBJECTS)
health.nats = connection
assert health.nats_status() == expected
@ -82,7 +84,7 @@ def test_nats_status(enabled: bool, connection: Connection | None, expected: str
async def test_hotfolder_result_is_cached_for_ttl() -> None:
hotfolder = CountingHotFolder(result=True)
clock = Clock()
health = HealthProvider(nats_enabled=False, hotfolder=hotfolder, cache_ttl=5, clock=clock)
health = HealthProvider(nats_enabled=False, hotfolder=hotfolder, cache_ttl=5, clock=clock, subjects=SUBJECTS)
assert await health.hotfolder_status() == "ok"
hotfolder.result = False
clock.now += 4.9
@ -97,7 +99,9 @@ async def test_hanging_check_is_down_within_timeout_and_not_repeated() -> None:
hotfolder = CountingHotFolder()
hotfolder.release = threading.Event()
clock = Clock()
health = HealthProvider(nats_enabled=False, hotfolder=hotfolder, check_timeout=0.05, cache_ttl=5, clock=clock)
health = HealthProvider(
nats_enabled=False, hotfolder=hotfolder, check_timeout=0.05, cache_ttl=5, clock=clock, subjects=SUBJECTS
)
try:
assert await health.hotfolder_status() == "down"
clock.now += 10 # кэш истёк, но прежняя проверка ещё висит — новый поток не запускается
@ -110,11 +114,25 @@ async def test_hanging_check_is_down_within_timeout_and_not_repeated() -> None:
async def test_check_raising_is_down() -> None:
hotfolder = CountingHotFolder()
hotfolder.error = OSError("smb")
health = HealthProvider(nats_enabled=False, hotfolder=hotfolder)
health = HealthProvider(nats_enabled=False, hotfolder=hotfolder, subjects=SUBJECTS)
assert await health.hotfolder_status() == "down"
async def test_snapshot_with_connected_nats(tmp_path: Path) -> None:
health = HealthProvider(nats_enabled=True, hotfolder=LocalHotFolder(tmp_path))
health = HealthProvider(nats_enabled=True, hotfolder=LocalHotFolder(tmp_path), subjects=SUBJECTS)
health.nats = Connection(True)
assert await health.snapshot() == {"nats": "ok", "hotfolder": "ok"}
assert await health.snapshot() == {"nats": "ok", "hotfolder": "ok", "subjects": SUBJECTS}
@pytest.mark.parametrize(
("enabled", "connection", "expected"),
[(True, Connection(True), "ok"), (True, Connection(False), "down"), (False, None, "disabled")],
)
async def test_snapshot_contains_subjects(enabled: bool, connection: Connection | None, expected: str) -> None:
health = HealthProvider(nats_enabled=enabled, hotfolder=None, subjects=SUBJECTS)
health.nats = connection
snapshot = await health.snapshot()
assert snapshot["nats"] == expected
assert snapshot["subjects"] == SUBJECTS

View File

@ -21,13 +21,15 @@ from src.contract import (
build_ack_message,
build_error_message,
)
from src.core.config import Settings, get_settings
from src.core.config import NATSSettings, Settings, get_settings
from src.core.nats import (
REQUEST_QUEUE_GROUP,
NatsConfigError,
NatsGateway,
NatsNotConnectedError,
PublishFailed,
build_nats_auth,
used_subjects,
)
@ -151,6 +153,22 @@ async def test_connect_with_nkeys_and_jwt_when_both_set(monkeypatch: pytest.Monk
assert fake.jetstream_calls == 0
def test_build_nats_auth_strips_whitespace_around_seed_and_jwt(monkeypatch: pytest.MonkeyPatch):
"""Пробелы и перевод строки по краям (типично для значений из OpenBao) срезаются: seed читается, JWT чистый."""
keypair = nkeys.from_seed(nkeys.encode_seed(b"\x02" * 32, nkeys.PREFIX_BYTE_USER))
settings = make_settings(
monkeypatch,
OPENBAO__NATS__NATS_SEED=f" {keypair.seed.decode()}\n",
OPENBAO__NATS__NATS_JWT="\n dummy.jwt.token \n",
)
kwargs = build_nats_auth(settings.OPENBAO.NATS)
assert kwargs["user_jwt_cb"]() == b"dummy.jwt.token"
signature = base64.b64decode(kwargs["signature_cb"]("nonce-1"))
assert keypair.verify(b"nonce-1", signature) is True
@pytest.mark.parametrize(
"env",
[
@ -429,3 +447,51 @@ async def test_publish_result_final_uses_result_msg_id_and_utf8(gateway: NatsGat
assert headers == {"Nats-Msg-Id": f"result:{job_id}:1"}
assert "ИС не ответила".encode() in payload # UTF-8 без \u-экранирования
assert json.loads(payload) == message.to_wire()
# --- subject'ы из настроек (#32) ---------------------------------------------------------------------------------
CUSTOM_ENV = {
"OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT": "test.files.img_uploaded",
"OPENBAO__NATS__NATS_RESULT_SUBJECT": "test.files.tabular_parsed",
}
def test_used_subjects_scan_gateway() -> None:
nats = NATSSettings(NATS_URL="nats://x:4222", NATS_IMAGE_REQUEST_SUBJECT="test.files.img_uploaded")
assert used_subjects(nats) == {"image_request": "test.files.img_uploaded", "result": "files.tabular_parsed"}
async def test_subscribe_requests_uses_subject_from_settings(monkeypatch: pytest.MonkeyPatch, fake: FakeNatsClient):
gw = NatsGateway(make_settings(monkeypatch, **CUSTOM_ENV), client_factory=lambda: fake)
async def on_request(msg: Any) -> None:
return None
await gw.connect()
await gw.subscribe_requests(on_request)
[subscription] = fake.subscriptions
assert subscription["subject"] == "test.files.img_uploaded"
assert subscription["queue"] == "scan-gateway"
async def test_subscribe_requests_error_names_configured_subject(monkeypatch: pytest.MonkeyPatch):
gw = NatsGateway(make_settings(monkeypatch, **CUSTOM_ENV))
async def on_request(msg: Any) -> None:
return None
with pytest.raises(NatsNotConnectedError, match=r"test\.files\.img_uploaded"):
await gw.subscribe_requests(on_request)
async def test_publish_result_uses_subject_from_settings(monkeypatch: pytest.MonkeyPatch, fake: FakeNatsClient):
gw = NatsGateway(make_settings(monkeypatch, **CUSTOM_ENV), client_factory=lambda: fake)
message = build_ack_message(uuid.uuid4(), uuid.uuid4(), SERVICE)
await gw.connect()
await gw.publish_result(message, attempt=1)
[(subject, _, _)] = fake.published
assert subject == "test.files.tabular_parsed"

View File

@ -0,0 +1,2 @@
schema: spec-driven
created: 2026-10-12

View File

@ -0,0 +1,20 @@
## Context
Дизайн сквозной: `../docs/openspec/changes/nats-env-config/design.md` (решения 1–7, R-1…R-6). Здесь — только то, что
специфично для `scan-gateway`.
## Decisions
1. Сервис подписан на `image_request` (queue group `scan-gateway`), публикует ack и результат в `result`; в `/healthcheck.subjects` — `image_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` сквозной нумерации.

View 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); здесь — его часть для `scan-gateway` (RSHB/arm-scan-gateway#32).
Срез: вне плана срезов `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)`.
- Сервис подписан на `image_request` (queue group `scan-gateway`), публикует ack и результат в `result` — по настройкам; константы `contract/subjects.py` — только умолчания.
- Лог при старте и `/healthcheck.subjects` (`image_request`, `result`), независимо от состояния NATS.
- Тесты интеграционные (`tests/integration/`) задают собственные 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
- `image-recognition-gateway`: subject'ы обмена и учётные данные NATS — настройки окружения; `/healthcheck.subjects`.
## Impact
Файлы: `api/src/core/config.py`, `api/src/contract/subjects.py`, `api/src/core/nats.py`, `api/src/gateway/`, `api/src/core/health.py`, `api/src/app.py` (лог старта), тесты (`tests/unit/test_config*.py`, `tests/unit/test_nats.py`, `tests/unit/test_app.py`, `tests/integration/`), `api/.env.example`, README. Выкат обратно совместим (умолчания — текущие имена); набор окружения задаётся одинаково всем трём
сервисам.

View File

@ -0,0 +1,42 @@
## 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`): подписан на `image_request` (queue group `scan-gateway`), публикует ack и результат в `result`.
При старте SHALL проверяться все три имени (непустое, без пробельных символов, без `*` и `>`, без пустых токенов,
попарно различны); при нарушении приложение MUST NOT стартовать. Значения `NATS_JWT` и `NATS_SEED` MUST NOT попадать в
логи, `repr` настроек и тексты ошибок валидации. Используемые subject'ы SHALL писаться в лог при старте и отдаваться в
`/healthcheck` объектом `subjects` с ключами `image_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** тесты интеграционные (`tests/integration/`) запускаются при работающих контейнерах стенда на том же 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.nats`
SHALL отражать состояние.
#### Scenario: NATS перезапущен
- **WHEN** контейнер NATS перезапущен во время работы сервиса
- **THEN** в течение 20 с после старта NATS подписка восстановлена, `nats = ok`

View 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` (`image_request`, `result`) при `nats = ok | down | off`; лог старта с subject'ами
- [x] 1.3 Тесты интеграционные (`tests/integration/`) — уникальные 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` зелёный