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` (в БД не регистрируется), очистить распознанный документ), склеить и сохранить сырой ответ в S3 `raw/{job_id}.json` (в БД не регистрируется), очистить
горячую папку (только при `SCAN_CLEANUP=true`), нормализовать в `../docs/contracts/recognition-result.schema.json` горячую папку (только при `SCAN_CLEANUP=true`), нормализовать в `../docs/contracts/recognition-result.schema.json`
(`statementDocument` v1.2: по одному документу на таблицу `table_structure[i]`), опубликовать `files.tabular_parsed` (`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_BUCKET` | — (обязательна) | единственный бакет сервиса; запрос с другим `s3.bucket` -> `FILE_READ_ERROR` |
| `OPENBAO__MINIO__MINIO_SECURE` | `true` | https; для MinIO из compose — `false` | | `OPENBAO__MINIO__MINIO_SECURE` | `true` | https; для MinIO из compose — `false` |
| `OPENBAO__NATS__NATS_URL` | — (обязательна) | адрес NATS | | `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 | | `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`) | | `NATS_ENABLED` | `true` | `false` — воркер не запускается, подключения к NATS нет (`/healthcheck.nats = disabled`) |
| `PUBLISH_RETRIES` | `3` | повторы публикации ack/результата после первой попытки (всего до 4) | | `PUBLISH_RETRIES` | `3` | повторы публикации ack/результата после первой попытки (всего до 4) |
@ -362,8 +365,8 @@ MinIO и NATS); те же значения подхватываются из `EN
Схемы и примеры — `../docs/contracts/`; их байт-в-байт копии лежат в `api/contracts/` (идентичность проверяет unit-тест Схемы и примеры — `../docs/contracts/`; их байт-в-байт копии лежат в `api/contracts/` (идентичность проверяет unit-тест
`api/tests/unit/test_contract.py`, входит в `make check`), pydantic-модели — `api/src/contract/`. `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.img_uploaded` (по умолчанию; `OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT`), подписка в 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.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"`. `status = done | needs_check | error`, `Nats-Msg-Id = result:<job_id>:<attempt>`. `service.name = "scan-gateway"`.
- Ключи S3: исходник — из запроса (`s3.key`), сырой ответ (склейка файлов ИС) — `raw/{job_id}.json`, результат — - Ключи S3: исходник — из запроса (`s3.key`), сырой ответ (склейка файлов ИС) — `raw/{job_id}.json`, результат —
`results/{file_id}/{job_id}/result.json`. Бакет один. `results/{file_id}/{job_id}/result.json`. Бакет один.
@ -445,6 +448,7 @@ make golden-update # пересобрать эталоны нормализ
| `version` | `VERSION` | | `version` | `VERSION` |
| `nats` | `ok` — соединение установлено; `down` — подключение идёт, переподключение, сбой; `disabled` — `NATS_ENABLED=false` | | `nats` | `ok` — соединение установлено; `down` — подключение идёт, переподключение, сбой; `disabled` — `NATS_ENABLED=false` |
| `hotfolder` | `ok` — `HotFolder.healthy()` (корень существует и доступен); `down` — нет, проверка не уложилась в 2 с или упала | | `hotfolder` | `ok` — `HotFolder.healthy()` (корень существует и доступен); `down` — нет, проверка не уложилась в 2 с или упала |
| `subjects` | `{image_request, result}` — subject'ы NATS из настроек; отдаётся при любом состоянии NATS; JWT/seed в ответе нет |
Подключение к NATS идёт в фоне: HTTP отвечает сразу (`nats = down`), подписка появляется после подключения. Подключение к NATS идёт в фоне: HTTP отвечает сразу (`nats = down`), подписка появляется после подключения.
Результат проверки горячей папки кэшируется на 5 с. Результат проверки горячей папки кэшируется на 5 с.

View File

@ -26,8 +26,14 @@ OPENBAO__MINIO__MINIO_SECURE=false
OPENBAO__NATS__NATS_URL=nats://localhost:4222 OPENBAO__NATS__NATS_URL=nats://localhost:4222
OPENBAO__NATS__NATS_SEED= OPENBAO__NATS__NATS_SEED=
OPENBAO__NATS__NATS_JWT= 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 NATS_ENABLED=true
# PUBLISH_RETRIES — число повторов публикации ack/результата в NATS после первой попытки (при 3 — до 4 попыток всего); # PUBLISH_RETRIES — число повторов публикации ack/результата в NATS после первой попытки (при 3 — до 4 попыток всего);
# DONE_CACHE_SIZE — размер LRU обработанных job_id (защита от повторной доставки запроса) # DONE_CACHE_SIZE — размер LRU обработанных job_id (защита от повторной доставки запроса)

View File

@ -2,7 +2,7 @@
"$schema": "https://json-schema.org/draft/2020-12/schema", "$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "arm-ka/recognition-request/v1", "$id": "arm-ka/recognition-request/v1",
"title": "Запрос распознавания файла (arm-core → tabular-parser / scan-gateway), контракт v1.0", "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 отвергают незнакомые значения.", "$comment": "Версионирование: schema_version = MAJOR.MINOR. MINOR (1.1) — необязательные поля и новые значения enum; MAJOR (2.0) — удаление/переименование полей, смена типов или семантики. Потребитель принимает тот же MAJOR и любой MINOR. Новый MINOR раскатывается сначала на потребителей (ядро и сервисы обновляют копии схем раньше, чем производитель начнёт слать 1.1), потому что закрытые enum отвергают незнакомые значения.",
"type": "object", "type": "object",
"required": ["schema_version", "job_id", "file_id", "pipeline", "file_type", "original_name", "s3", "reply_deadline", "attempt"], "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", "$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "arm-ka/recognition-result-message/v1", "$id": "arm-ka/recognition-result-message/v1",
"title": "Сообщение результата распознавания (tabular-parser / scan-gateway → arm-core), контракт v1.0", "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 отвергают незнакомые значения.", "$comment": "Версионирование: schema_version = MAJOR.MINOR. MINOR (1.1) — необязательные поля и новые значения enum; MAJOR (2.0) — удаление/переименование полей, смена типов или семантики. Потребитель принимает тот же MAJOR и любой MINOR. Новый MINOR раскатывается сначала на потребителей (ядро и сервисы обновляют копии схем раньше, чем производитель начнёт слать 1.1), потому что закрытые enum отвергают незнакомые значения.",
"type": "object", "type": "object",
"required": ["schema_version", "job_id", "file_id", "status", "service"], "required": ["schema_version", "job_id", "file_id", "status", "service"],

View File

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

View File

@ -22,10 +22,12 @@ REQUEST_SUBJECTS: Mapping[str, str] = {
} }
def request_subject(pipeline: str) -> str: def request_subject(pipeline: str, subjects: Mapping[str, str] = REQUEST_SUBJECTS) -> str:
"""Subject запроса по pipeline: `tabular` → `files.tabular_uploaded`, `image` → `files.img_uploaded`.""" """Subject запроса по pipeline. `subjects` — карта `pipeline → subject` окружения
(`NATSSettings.request_subjects()`); по умолчанию — имена контракта: `tabular` → `files.tabular_uploaded`,
`image` → `files.img_uploaded`."""
try: try:
return REQUEST_SUBJECTS[pipeline] return subjects[pipeline]
except KeyError: except KeyError:
raise ValueError(f"неизвестный pipeline {pipeline!r}: допустимы {', '.join(get_args(Pipeline))}") from None raise ValueError(f"неизвестный pipeline {pipeline!r}: допустимы {', '.join(get_args(Pipeline))}") from None

View File

@ -11,16 +11,32 @@
""" """
import os import os
import re
from functools import lru_cache from functools import lru_cache
from typing import Any, Literal 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 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 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): class MinioSettings(BaseSettings):
model_config = SettingsConfigDict(hide_input_in_errors=True)
MINIO_URL: str MINIO_URL: str
MINIO_ADMIN_LOGIN: str MINIO_ADMIN_LOGIN: str
MINIO_ADMIN_PASSWORD: str MINIO_ADMIN_PASSWORD: str
@ -29,15 +45,57 @@ class MinioSettings(BaseSettings):
class NATSSettings(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_URL: str
NATS_SEED: str = "" # пусто — подключение к NATS без nkeys/JWT (локальная разработка, тесты) NATS_SEED: SecretStr = SecretStr("") # пусто — подключение к NATS без nkeys/JWT (локальная разработка, тесты)
NATS_JWT: str = "" 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): class SMBSettings(BaseSettings):
"""SMB-шара горячих папок ИС Сканирование. Нужна только при `SCAN_HOTFOLDER_MODE=smb` (там все поля обязательны, """SMB-шара горячих папок ИС Сканирование. Нужна только при `SCAN_HOTFOLDER_MODE=smb` (там все поля обязательны,
проверяет `Settings`); в режиме local весь блок можно не задавать.""" проверяет `Settings`); в режиме local весь блок можно не задавать."""
model_config = SettingsConfigDict(hide_input_in_errors=True)
SMB_HOST: str = "" SMB_HOST: str = ""
SMB_PATH: str = "" SMB_PATH: str = ""
SMB_LOGIN: str = "" SMB_LOGIN: str = ""
@ -45,6 +103,8 @@ class SMBSettings(BaseSettings):
class OpenBAOSettings(BaseSettings): class OpenBAOSettings(BaseSettings):
model_config = SettingsConfigDict(hide_input_in_errors=True)
MINIO: MinioSettings MINIO: MinioSettings
NATS: NATSSettings NATS: NATSSettings
SMB: SMBSettings = SMBSettings() # весь блок необязателен: в режиме local не требуется SMB: SMBSettings = SMBSettings() # весь блок необязателен: в режиме local не требуется
@ -61,6 +121,8 @@ class OpenBAOSettings(BaseSettings):
class ServerSettings(BaseSettings): class ServerSettings(BaseSettings):
model_config = SettingsConfigDict(hide_input_in_errors=True)
HOST: str = "0.0.0.0" HOST: str = "0.0.0.0"
PORT: int = 8000 PORT: int = 8000
@ -99,6 +161,7 @@ class Settings(BaseSettings):
env_file_encoding="utf-8", env_file_encoding="utf-8",
env_nested_delimiter="__", env_nested_delimiter="__",
nested_model_default_partial_update=False, nested_model_default_partial_update=False,
hide_input_in_errors=True,
) )
@classmethod @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` — в том - `nats`: `disabled` при `NATS_ENABLED=false`; `ok`, если соединение установлено (`is_connected`), иначе `down` — в том
числе пока идёт первое подключение, во время переподключения и после сбоя фоновой задачи подключения. числе пока идёт первое подключение, во время переподключения и после сбоя фоновой задачи подключения.
@ -11,8 +11,8 @@
import asyncio import asyncio
import time import time
from collections.abc import Callable from collections.abc import Callable, Mapping
from typing import Literal, Protocol from typing import Any, Literal, Protocol
from loguru import logger from loguru import logger
@ -37,11 +37,12 @@ class ConnectionState(Protocol):
class HealthProvider: class HealthProvider:
"""Поля `nats`/`hotfolder` для `/healthcheck`. `nats` присваивается, когда соединение создано (lifespan).""" """Поля `nats`/`hotfolder` для `/healthcheck`. `nats` присваивается, когда соединение создано (lifespan)."""
def __init__( def __init__( # noqa: PLR0913 - именованные параметры: подмены для тестов и тайминги
self, self,
*, *,
nats_enabled: bool, nats_enabled: bool,
hotfolder: HotFolder | None, hotfolder: HotFolder | None,
subjects: Mapping[str, str],
check_timeout: float = HOTFOLDER_CHECK_TIMEOUT, check_timeout: float = HOTFOLDER_CHECK_TIMEOUT,
cache_ttl: float = HOTFOLDER_CACHE_TTL, cache_ttl: float = HOTFOLDER_CACHE_TTL,
clock: Callable[[], float] = time.monotonic, clock: Callable[[], float] = time.monotonic,
@ -49,6 +50,7 @@ class HealthProvider:
self.nats: ConnectionState | None = None self.nats: ConnectionState | None = None
self._nats_enabled = nats_enabled self._nats_enabled = nats_enabled
self._hotfolder = hotfolder self._hotfolder = hotfolder
self._subjects = dict(subjects)
self._check_timeout = check_timeout self._check_timeout = check_timeout
self._cache_ttl = cache_ttl self._cache_ttl = cache_ttl
self._clock = clock self._clock = clock
@ -86,8 +88,13 @@ class HealthProvider:
self._cached, self._cached_at = status, self._clock() self._cached, self._cached_at = status, self._clock()
return status return status
async def snapshot(self) -> dict[str, str]: async def snapshot(self) -> dict[str, Any]:
return {"nats": self.nats_status(), "hotfolder": await self.hotfolder_status()} """`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: def _consume_exception(task: "asyncio.Task[bool]") -> None:

View File

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

View File

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

View File

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

View File

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

View File

@ -1,3 +1,6 @@
import os
from collections.abc import Callable
import pytest import pytest
from pydantic import ValidationError 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_TIMEOUT_SECONDS == 1200
assert s.SCAN_MAX_INFLIGHT == 4 assert s.SCAN_MAX_INFLIGHT == 4
assert s.OPENBAO.SMB.SMB_HOST == "" assert s.OPENBAO.SMB.SMB_HOST == ""
assert s.OPENBAO.NATS.NATS_SEED == "" assert s.OPENBAO.NATS.NATS_SEED.get_secret_value() == ""
assert s.OPENBAO.NATS.NATS_JWT == "" assert s.OPENBAO.NATS.NATS_JWT.get_secret_value() == ""
assert not hasattr(s.OPENBAO, "DATABASE") 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_URL == "127.0.0.1:9000"
assert s.OPENBAO.MINIO.MINIO_SECURE is False 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_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): def test_env_example_loads(monkeypatch):
@ -185,4 +188,127 @@ def test_env_example_loads(monkeypatch):
assert not hasattr(s, "SCAN_TMP_DIR") assert not hasattr(s, "SCAN_TMP_DIR")
assert s.SCAN_SETTLE_SECONDS == 2 assert s.SCAN_SETTLE_SECONDS == 2
assert s.OPENBAO.MINIO.MINIO_BUCKET == "arm-ka" 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") 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: def test_request_subject_matches_example_pipeline() -> None:
for name in ("request-tabular.json", "request-image.json"): for name in ("request-tabular.json", "request-image.json"):
request = RecognitionRequest.model_validate(_example(name)) 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") pytestmark = pytest.mark.usefixtures("log_lines")
SUBJECTS = {"image_request": "files.img_uploaded", "result": "files.tabular_parsed"}
class Connection: class Connection:
def __init__(self, connected: bool) -> None: def __init__(self, connected: bool) -> None:
@ -50,17 +52,17 @@ class Clock:
async def test_local_root_exists_is_ok(tmp_path: Path) -> None: async def test_local_root_exists_is_ok(tmp_path: Path) -> None:
health = HealthProvider(nats_enabled=False, hotfolder=LocalHotFolder(tmp_path)) health = HealthProvider(nats_enabled=False, hotfolder=LocalHotFolder(tmp_path), subjects=SUBJECTS)
assert await health.snapshot() == {"nats": "disabled", "hotfolder": "ok"} assert await health.snapshot() == {"nats": "disabled", "hotfolder": "ok", "subjects": SUBJECTS}
async def test_missing_root_is_down(tmp_path: Path) -> None: 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" assert await health.hotfolder_status() == "down"
async def test_no_hotfolder_is_down() -> None: 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( @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: 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 health.nats = connection
assert health.nats_status() == expected 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: async def test_hotfolder_result_is_cached_for_ttl() -> None:
hotfolder = CountingHotFolder(result=True) hotfolder = CountingHotFolder(result=True)
clock = Clock() 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" assert await health.hotfolder_status() == "ok"
hotfolder.result = False hotfolder.result = False
clock.now += 4.9 clock.now += 4.9
@ -97,7 +99,9 @@ async def test_hanging_check_is_down_within_timeout_and_not_repeated() -> None:
hotfolder = CountingHotFolder() hotfolder = CountingHotFolder()
hotfolder.release = threading.Event() hotfolder.release = threading.Event()
clock = Clock() 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: try:
assert await health.hotfolder_status() == "down" assert await health.hotfolder_status() == "down"
clock.now += 10 # кэш истёк, но прежняя проверка ещё висит — новый поток не запускается 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: async def test_check_raising_is_down() -> None:
hotfolder = CountingHotFolder() hotfolder = CountingHotFolder()
hotfolder.error = OSError("smb") 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" assert await health.hotfolder_status() == "down"
async def test_snapshot_with_connected_nats(tmp_path: Path) -> None: 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) 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_ack_message,
build_error_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 ( from src.core.nats import (
REQUEST_QUEUE_GROUP, REQUEST_QUEUE_GROUP,
NatsConfigError, NatsConfigError,
NatsGateway, NatsGateway,
NatsNotConnectedError, NatsNotConnectedError,
PublishFailed, 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 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( @pytest.mark.parametrize(
"env", "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 headers == {"Nats-Msg-Id": f"result:{job_id}:1"}
assert "ИС не ответила".encode() in payload # UTF-8 без \u-экранирования assert "ИС не ответила".encode() in payload # UTF-8 без \u-экранирования
assert json.loads(payload) == message.to_wire() 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` зелёный