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