From 157929596155faa5c79a8491c1f55a1a7ff906e4 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Mon, 12 Oct 2026 21:40:35 +0400 Subject: [PATCH 1/6] =?UTF-8?q?openspec:=20change=20nats-env-config=20?= =?UTF-8?q?=E2=80=94=20subject'=D1=8B=20=D0=B8=20JWT=20NATS=20=D0=BF=D0=BE?= =?UTF-8?q?=20=D0=BE=D0=BA=D1=80=D1=83=D0=B6=D0=B5=D0=BD=D0=B8=D1=8E=20?= =?UTF-8?q?=D0=B8=D0=B7=20OpenBao?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- .../changes/nats-env-config/.openspec.yaml | 2 + openspec/changes/nats-env-config/design.md | 20 +++++++++ openspec/changes/nats-env-config/proposal.md | 40 ++++++++++++++++++ .../specs/image-recognition-gateway/spec.md | 42 +++++++++++++++++++ openspec/changes/nats-env-config/tasks.md | 18 ++++++++ 5 files changed, 122 insertions(+) create mode 100644 openspec/changes/nats-env-config/.openspec.yaml create mode 100644 openspec/changes/nats-env-config/design.md create mode 100644 openspec/changes/nats-env-config/proposal.md create mode 100644 openspec/changes/nats-env-config/specs/image-recognition-gateway/spec.md create mode 100644 openspec/changes/nats-env-config/tasks.md diff --git a/openspec/changes/nats-env-config/.openspec.yaml b/openspec/changes/nats-env-config/.openspec.yaml new file mode 100644 index 0000000..b9ebfc7 --- /dev/null +++ b/openspec/changes/nats-env-config/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-10-12 diff --git a/openspec/changes/nats-env-config/design.md b/openspec/changes/nats-env-config/design.md new file mode 100644 index 0000000..1a98b82 --- /dev/null +++ b/openspec/changes/nats-env-config/design.md @@ -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` сквозной нумерации. diff --git a/openspec/changes/nats-env-config/proposal.md b/openspec/changes/nats-env-config/proposal.md new file mode 100644 index 0000000..26c9c8a --- /dev/null +++ b/openspec/changes/nats-env-config/proposal.md @@ -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. Выкат обратно совместим (умолчания — текущие имена); набор окружения задаётся одинаково всем трём +сервисам. diff --git a/openspec/changes/nats-env-config/specs/image-recognition-gateway/spec.md b/openspec/changes/nats-env-config/specs/image-recognition-gateway/spec.md new file mode 100644 index 0000000..861be2a --- /dev/null +++ b/openspec/changes/nats-env-config/specs/image-recognition-gateway/spec.md @@ -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` diff --git a/openspec/changes/nats-env-config/tasks.md b/openspec/changes/nats-env-config/tasks.md new file mode 100644 index 0000000..743c337 --- /dev/null +++ b/openspec/changes/nats-env-config/tasks.md @@ -0,0 +1,18 @@ +## 1. Тесты + +- [ ] 1.1 Unit: умолчания subject'ов, чтение `OPENBAO__NATS__NATS_*_SUBJECT`, отказ на `*`/`>`/пробеле/пустом токене и + совпадающих именах; `SecretStr` для `NATS_JWT`/`NATS_SEED`, отсутствие значений в ошибке валидации и `repr` +- [ ] 1.2 Unit: `/healthcheck.subjects` (`image_request`, `result`) при `nats = ok | down | off`; лог старта с subject'ами +- [ ] 1.3 Тесты интеграционные (`tests/integration/`) — уникальные subject'ы через настройки + +## 2. Реализация + +- [ ] 2.1 `core/config.py`: поля и валидация `NATSSettings`, `SecretStr`, `hide_input_in_errors=True` +- [ ] 2.2 Колбэки nkeys — `get_secret_value()` +- [ ] 2.3 Подписка и публикация по настройкам; константы `contract/subjects.py` — умолчания +- [ ] 2.4 `/healthcheck.subjects`, лог при старте + +## 3. Документы + +- [ ] 3.1 `api/.env.example`, README (переменные NATS, `/healthcheck`) +- [ ] 3.2 `make check-full` зелёный From f1f7d784a5aceded1f129ccd218055b70a559183 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Tue, 13 Oct 2026 12:06:11 +0400 Subject: [PATCH 2/6] =?UTF-8?q?NATS:=20subject'=D1=8B=20=D0=B8=20JWT=20?= =?UTF-8?q?=D0=B8=D0=B7=20OPENBAO=5F=5FNATS=5F=5F*,=20SecretStr,=20=D1=81?= =?UTF-8?q?=D0=BA=D1=80=D1=8B=D1=82=D0=B8=D0=B5=20=D0=B2=D1=85=D0=BE=D0=B4?= =?UTF-8?q?=D0=B0=20=D0=B2=20=D0=BE=D1=88=D0=B8=D0=B1=D0=BA=D0=B0=D1=85=20?= =?UTF-8?q?(#32)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- api/contracts/recognition-request.schema.json | 2 +- .../recognition-result-message.schema.json | 2 +- api/src/contract/subjects.py | 8 +- api/src/core/config.py | 69 ++++++++++- api/src/core/nats.py | 4 +- api/tests/unit/test_config.py | 109 +++++++++++++++++- api/tests/unit/test_contract.py | 9 ++ 7 files changed, 189 insertions(+), 14 deletions(-) diff --git a/api/contracts/recognition-request.schema.json b/api/contracts/recognition-request.schema.json index 9d59f2c..afd5077 100644 --- a/api/contracts/recognition-request.schema.json +++ b/api/contracts/recognition-request.schema.json @@ -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 = : (не поле тела). Один файл (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 = : (не поле тела). Один файл (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"], diff --git a/api/contracts/recognition-result-message.schema.json b/api/contracts/recognition-result-message.schema.json index 7c1cfb5..4d9f401 100644 --- a/api/contracts/recognition-result-message.schema.json +++ b/api/contracts/recognition-result-message.schema.json @@ -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:: — для status = processing; result:: — для done, needs_check и error. Форма запроса : описана в схеме recognition-request. status = processing — ack «взял в работу», публикуется до чтения файла из S3. group_schema в конверт не входит: источник — файл результата (recognition-result).", + "description": "Тело сообщения на subject ответа (настройка окружения NATS_RESULT_SUBJECT, по умолчанию files.tabular_parsed). Заголовок Nats-Msg-Id (не поле тела) на этом subject имеет две формы: ack:: — для status = processing; result:: — для done, needs_check и error. Форма запроса : описана в схеме 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"], diff --git a/api/src/contract/subjects.py b/api/src/contract/subjects.py index 243b11d..19a4bb7 100644 --- a/api/src/contract/subjects.py +++ b/api/src/contract/subjects.py @@ -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 diff --git a/api/src/core/config.py b/api/src/core/config.py index db9eb68..6b3de7d 100644 --- a/api/src/core/config.py +++ b/api/src/core/config.py @@ -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 diff --git a/api/src/core/nats.py b/api/src/core/nats.py index 2144880..37bc6e9 100644 --- a/api/src/core/nats.py +++ b/api/src/core/nats.py @@ -92,8 +92,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: diff --git a/api/tests/unit/test_config.py b/api/tests/unit/test_config.py index 04d29e5..3b84166 100644 --- a/api/tests/unit/test_config.py +++ b/api/tests/unit/test_config.py @@ -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,102 @@ 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" + + +@pytest.mark.parametrize( + "value", + ["", "files.*", "files.>", ">", "files..tabular", ".files", "files.", "files tabular", "files.tabular_uploaded "], +) +def test_invalid_subject_rejected(settings_from_env, value: str) -> None: + with pytest.raises(ValidationError, match="NATS_RESULT_SUBJECT"): + settings_from_env(OPENBAO__NATS__NATS_RESULT_SUBJECT=value) + + +def test_equal_subjects_rejected(settings_from_env) -> None: + with pytest.raises(ValidationError, match="NATS_TABULAR_REQUEST_SUBJECT и NATS_RESULT_SUBJECT совпадают"): + settings_from_env(OPENBAO__NATS__NATS_RESULT_SUBJECT="files.tabular_uploaded") + + +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 diff --git a/api/tests/unit/test_contract.py b/api/tests/unit/test_contract.py index c2691f9..d1de807 100644 --- a/api/tests/unit/test_contract.py +++ b/api/tests/unit/test_contract.py @@ -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)) From 192500e9f9de0c41c3ca7d0b24d9a786d60b36e0 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Tue, 13 Oct 2026 12:19:11 +0400 Subject: [PATCH 3/6] =?UTF-8?q?NATS:=20=D0=BF=D0=BE=D0=B4=D0=BF=D0=B8?= =?UTF-8?q?=D1=81=D0=BA=D0=B0,=20=D0=BF=D1=83=D0=B1=D0=BB=D0=B8=D0=BA?= =?UTF-8?q?=D0=B0=D1=86=D0=B8=D1=8F=20=D0=B8=20/healthcheck.subjects=20?= =?UTF-8?q?=D0=BF=D0=BE=20=D0=BD=D0=B0=D1=81=D1=82=D1=80=D0=BE=D0=B9=D0=BA?= =?UTF-8?q?=D0=B0=D0=BC;=20=D0=B8=D0=BD=D1=82=D0=B5=D0=B3=D1=80=D0=B0?= =?UTF-8?q?=D1=86=D0=B8=D0=BE=D0=BD=D0=BD=D1=8B=D0=B5=20=D1=82=D0=B5=D1=81?= =?UTF-8?q?=D1=82=D1=8B=20=D0=BD=D0=B0=20=D1=81=D0=B2=D0=BE=D0=B8=D1=85=20?= =?UTF-8?q?subject'=D0=B0=D1=85=20(#32)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- README.md | 10 ++-- api/.env.example | 8 ++- api/src/app.py | 11 ++-- api/src/core/health.py | 19 ++++--- api/src/core/nats.py | 28 ++++++---- .../test_nats_minio_integration.py | 37 ++++++++------ api/tests/integration/test_roundtrip.py | 34 ++++++++----- api/tests/unit/test_app.py | 20 +++++++- api/tests/unit/test_health.py | 38 ++++++++++---- api/tests/unit/test_nats.py | 51 ++++++++++++++++++- 10 files changed, 192 insertions(+), 64 deletions(-) diff --git a/README.md b/README.md index c9ebe66..9e9ff3d 100644 --- a/README.md +++ b/README.md @@ -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 = :`. -- Ответы — subject `files.tabular_parsed`: ack `status=processing`, `Nats-Msg-Id = ack::`; итог +- Запрос: subject `files.img_uploaded` (по умолчанию; `OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT`), подписка в queue group `scan-gateway`, `Nats-Msg-Id = :`. +- Ответы — subject `files.tabular_parsed` (по умолчанию; `OPENBAO__NATS__NATS_RESULT_SUBJECT`): ack `status=processing`, `Nats-Msg-Id = ack::`; итог `status = done | needs_check | error`, `Nats-Msg-Id = result::`. `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 с. diff --git a/api/.env.example b/api/.env.example index 2808c3d..cab74f2 100644 --- a/api/.env.example +++ b/api/.env.example @@ -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 (защита от повторной доставки запроса) diff --git a/api/src/app.py b/api/src/app.py index 59f13c2..f8fbb7e 100644 --- a/api/src/app.py +++ b/api/src/app.py @@ -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 diff --git a/api/src/core/health.py b/api/src/core/health.py index 20f4c06..10e2667 100644 --- a/api/src/core/health.py +++ b/api/src/core/health.py @@ -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: diff --git a/api/src/core/nats.py b/api/src/core/nats.py index 37bc6e9..163329c 100644 --- a/api/src/core/nats.py +++ b/api/src/core/nats.py @@ -8,7 +8,9 @@ поэтому `connect()` возвращается только при успехе (вызывающий оборачивает его в задачу или `wait_for`). После переподключения nats-py сам отправляет SUB для всех подписок с прежней queue group — подписку повторно оформлять не нужно. -- `subscribe_requests(cb)` — подписка на `files.img_uploaded` в queue group `scan-gateway`: из нескольких +- `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` (подтверждение сервером) с @@ -31,9 +33,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, @@ -116,6 +116,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::` для `processing`, иначе `result::`.""" if message.status is MessageStatus.PROCESSING: @@ -209,15 +214,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 +237,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 +297,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: diff --git a/api/tests/integration/test_nats_minio_integration.py b/api/tests/integration/test_nats_minio_integration.py index d1f0810..674abc8 100644 --- a/api/tests/integration/test_nats_minio_integration.py +++ b/api/tests/integration/test_nats_minio_integration.py @@ -3,7 +3,9 @@ Недоступная инфраструктура — падение с понятным сообщением, не skip. Общий контейнер NATS не перезапускается: сценарий переподключения покрыт unit-тестом колбэков (tests/unit/test_nats.py). Чужой трафик на тех же subject'ах -отсекается уникальным `job_id` теста; ключи MinIO — под уникальным префиксом `itest//` и удаляются. +отсекается уникальным `job_id` теста; кроме того, тесты работают на своих subject'ах прогона (`TEST_SUBJECTS`, суффикс +`RUN`), поэтому контейнеры стенда на умолчательных `files.*` их не видят; ключи MinIO — под уникальным префиксом +`itest//` и удаляются. """ 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) diff --git a/api/tests/integration/test_roundtrip.py b/api/tests/integration/test_roundtrip.py index 20f9363..b6e7084 100644 --- a/api/tests/integration/test_roundtrip.py +++ b/api/tests/integration/test_roundtrip.py @@ -22,14 +22,12 @@ ответа, в порядке возрастания имени, с телами байт в байт; входящий файл совпадает с объектом S3. Готовность набора — по settle (`SETTLE_SECONDS`, 0.2 с). -Общая инфраструктура: NATS и MinIO используют и другие сервисы workspace. Чужие конверты на `files.tabular_parsed` -отсекаются по `job_id` теста; ключи MinIO уникальны (`uploads/itest-/…`, `raw/.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-/…`, +`raw/.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 = :`.""" 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") diff --git a/api/tests/unit/test_app.py b/api/tests/unit/test_app.py index 9cbfc1f..f30f7ab 100644 --- a/api/tests/unit/test_app.py +++ b/api/tests/unit/test_app.py @@ -52,11 +52,29 @@ 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_start_in_local_mode(local_env: Path) -> None: """Spec «Старт в режиме локальных папок»: 200 с uptime, version, nats, hotfolder = ok; OPENBAO__SMB__* не нужны.""" app = create_app(get_settings()) diff --git a/api/tests/unit/test_health.py b/api/tests/unit/test_health.py index 2517573..29eecd0 100644 --- a/api/tests/unit/test_health.py +++ b/api/tests/unit/test_health.py @@ -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 diff --git a/api/tests/unit/test_nats.py b/api/tests/unit/test_nats.py index ca359cf..993075a 100644 --- a/api/tests/unit/test_nats.py +++ b/api/tests/unit/test_nats.py @@ -21,13 +21,14 @@ 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, + used_subjects, ) @@ -429,3 +430,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" From cd5409be6e8a9e8c338a3fbe0bb27274b055b349 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Tue, 13 Oct 2026 12:56:27 +0400 Subject: [PATCH 4/6] =?UTF-8?q?=D0=A2=D0=B5=D1=81=D1=82=D1=8B=20NATS-?= =?UTF-8?q?=D0=BD=D0=B0=D1=81=D1=82=D1=80=D0=BE=D0=B5=D0=BA:=20=D0=B2?= =?UTF-8?q?=D1=81=D0=B5=20=D0=BA=D0=BB=D1=8E=D1=87=D0=B8=20=D0=B8=20=D0=BF?= =?UTF-8?q?=D0=B0=D1=80=D1=8B=20subject,=20=D1=83=D1=82=D0=B5=D1=87=D0=BA?= =?UTF-8?q?=D0=B0=20=D1=81=D0=B5=D0=BA=D1=80=D0=B5=D1=82=D0=BE=D0=B2,=20st?= =?UTF-8?q?rip=20seed,=20JWT=20=D0=B2=20/healthcheck,=20=D0=BB=D0=BE=D0=B3?= =?UTF-8?q?=20=D1=81=D1=82=D0=B0=D1=80=D1=82=D0=B0=20(#32)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Фикс-волна minor-находок ревью (T6.1, T6.3, T2.2-аналог); tasks.md отмечен, кроме make check-full. Co-Authored-By: Claude Opus 5.5 --- api/tests/unit/test_app.py | 40 +++++++++++++++++++++++ api/tests/unit/test_config.py | 37 +++++++++++++++++---- api/tests/unit/test_nats.py | 17 ++++++++++ openspec/changes/nats-env-config/tasks.md | 16 ++++----- 4 files changed, 96 insertions(+), 14 deletions(-) diff --git a/api/tests/unit/test_app.py b/api/tests/unit/test_app.py index f30f7ab..7f436eb 100644 --- a/api/tests/unit/test_app.py +++ b/api/tests/unit/test_app.py @@ -75,6 +75,46 @@ async def test_healthcheck_subjects_from_env_without_jwt(local_env: Path, monkey 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()) diff --git a/api/tests/unit/test_config.py b/api/tests/unit/test_config.py index 3b84166..577ef7f 100644 --- a/api/tests/unit/test_config.py +++ b/api/tests/unit/test_config.py @@ -244,18 +244,43 @@ def test_nats_subjects_from_openbao(settings_from_env) -> None: 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, value: str) -> None: - with pytest.raises(ValidationError, match="NATS_RESULT_SUBJECT"): - settings_from_env(OPENBAO__NATS__NATS_RESULT_SUBJECT=value) +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}) -def test_equal_subjects_rejected(settings_from_env) -> None: - with pytest.raises(ValidationError, match="NATS_TABULAR_REQUEST_SUBJECT и NATS_RESULT_SUBJECT совпадают"): - settings_from_env(OPENBAO__NATS__NATS_RESULT_SUBJECT="files.tabular_uploaded") +@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: diff --git a/api/tests/unit/test_nats.py b/api/tests/unit/test_nats.py index 993075a..bd1aa2d 100644 --- a/api/tests/unit/test_nats.py +++ b/api/tests/unit/test_nats.py @@ -28,6 +28,7 @@ from src.core.nats import ( NatsGateway, NatsNotConnectedError, PublishFailed, + build_nats_auth, used_subjects, ) @@ -152,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", [ diff --git a/openspec/changes/nats-env-config/tasks.md b/openspec/changes/nats-env-config/tasks.md index 743c337..ef689d0 100644 --- a/openspec/changes/nats-env-config/tasks.md +++ b/openspec/changes/nats-env-config/tasks.md @@ -1,18 +1,18 @@ ## 1. Тесты -- [ ] 1.1 Unit: умолчания subject'ов, чтение `OPENBAO__NATS__NATS_*_SUBJECT`, отказ на `*`/`>`/пробеле/пустом токене и +- [x] 1.1 Unit: умолчания subject'ов, чтение `OPENBAO__NATS__NATS_*_SUBJECT`, отказ на `*`/`>`/пробеле/пустом токене и совпадающих именах; `SecretStr` для `NATS_JWT`/`NATS_SEED`, отсутствие значений в ошибке валидации и `repr` -- [ ] 1.2 Unit: `/healthcheck.subjects` (`image_request`, `result`) при `nats = ok | down | off`; лог старта с subject'ами -- [ ] 1.3 Тесты интеграционные (`tests/integration/`) — уникальные subject'ы через настройки +- [x] 1.2 Unit: `/healthcheck.subjects` (`image_request`, `result`) при `nats = ok | down | off`; лог старта с subject'ами +- [x] 1.3 Тесты интеграционные (`tests/integration/`) — уникальные subject'ы через настройки ## 2. Реализация -- [ ] 2.1 `core/config.py`: поля и валидация `NATSSettings`, `SecretStr`, `hide_input_in_errors=True` -- [ ] 2.2 Колбэки nkeys — `get_secret_value()` -- [ ] 2.3 Подписка и публикация по настройкам; константы `contract/subjects.py` — умолчания -- [ ] 2.4 `/healthcheck.subjects`, лог при старте +- [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. Документы -- [ ] 3.1 `api/.env.example`, README (переменные NATS, `/healthcheck`) +- [x] 3.1 `api/.env.example`, README (переменные NATS, `/healthcheck`) - [ ] 3.2 `make check-full` зелёный From 0ef4a8a6bc4ffc26718e89a1a38e1716a7788a29 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Tue, 13 Oct 2026 13:18:15 +0400 Subject: [PATCH 5/6] =?UTF-8?q?openspec:=20nats-env-config=20=E2=80=94=20c?= =?UTF-8?q?heck-full=20=D0=B7=D0=B5=D0=BB=D1=91=D0=BD=D1=8B=D0=B9,=20?= =?UTF-8?q?=D0=BF=D1=83=D0=BD=D0=BA=D1=82=203.2=20=D0=BE=D1=82=D0=BC=D0=B5?= =?UTF-8?q?=D1=87=D0=B5=D0=BD=20(#32)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- openspec/changes/nats-env-config/tasks.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/openspec/changes/nats-env-config/tasks.md b/openspec/changes/nats-env-config/tasks.md index ef689d0..f5e340e 100644 --- a/openspec/changes/nats-env-config/tasks.md +++ b/openspec/changes/nats-env-config/tasks.md @@ -15,4 +15,4 @@ ## 3. Документы - [x] 3.1 `api/.env.example`, README (переменные NATS, `/healthcheck`) -- [ ] 3.2 `make check-full` зелёный +- [x] 3.2 `make check-full` зелёный From c10f927730b0c5acc0029d2105f578ca23f6df68 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Tue, 13 Oct 2026 13:26:44 +0400 Subject: [PATCH 6/6] =?UTF-8?q?=D0=9F=D1=80=D0=B0=D0=B2=D0=BA=D0=B8=20?= =?UTF-8?q?=D1=84=D0=B8=D0=BD=D0=B0=D0=BB=D1=8C=D0=BD=D0=BE=D0=B3=D0=BE=20?= =?UTF-8?q?=D1=80=D0=B5=D0=B2=D1=8C=D1=8E:=20=D1=83=D0=BC=D0=BE=D0=BB?= =?UTF-8?q?=D1=87=D0=B0=D0=BD=D0=B8=D1=8F=20subject'=D0=BE=D0=B2=20=D0=BD?= =?UTF-8?q?=D0=B5=20=D0=BD=D0=B0=D0=B7=D0=B2=D0=B0=D0=BD=D1=8B=20=D1=84?= =?UTF-8?q?=D0=B0=D0=BA=D1=82=D0=B8=D1=87=D0=B5=D1=81=D0=BA=D0=B8=D0=BC?= =?UTF-8?q?=D0=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit M-3: докстринги worker.py не называют умолчание фактическим subject'ом. T7.1: выровнены переносы докстринга core/nats.py. M-5: CLAUDE.md помечает files.tabular_parsed как умолчание. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 2 +- api/src/core/nats.py | 7 +++---- api/src/core/worker.py | 4 ++-- 3 files changed, 6 insertions(+), 7 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index a47c1e3..ac3f99d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -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)`, старт — запись файла во «Входящие». **Нет доступа к БД.** ## Правила diff --git a/api/src/core/nats.py b/api/src/core/nats.py index 163329c..6f41dc1 100644 --- a/api/src/core/nats.py +++ b/api/src/core/nats.py @@ -9,10 +9,9 @@ После переподключения nats-py сам отправляет SUB для всех подписок с прежней queue group — подписку повторно оформлять не нужно. - `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` логируются, подписка живёт. + (`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). diff --git a/api/src/core/worker.py b/api/src/core/worker.py index a02f996..8252f38 100644 --- a/api/src/core/worker.py +++ b/api/src/core/worker.py @@ -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` — пауза между опросами (подмены для тестов).