Merge branch 'main' into add-report-shahmatka

This commit is contained in:
PotapovaA 2026-10-15 13:18:22 +03:00
commit 4390ce8f27
38 changed files with 1477 additions and 129 deletions

View File

@ -67,7 +67,7 @@
(`docker build`; убран из `.gitignore` шаблона). CI в репозитории нет: перед PR — `make check-full`. Без Nexus:
`make venv` ставит зависимости без `raisa_*`, npm-пакеты `raisa-*` — `optionalDependencies`.
- **NATS без JetStream** (`NATS_ENABLED=true`): `core/nats.py` — подключение в фоне (supervisor повторяет попытки),
подписка queue group `arm-core` на `files.tabular_parsed`. Обмен — `services/recognition_exchange.py`: `publish_pending`
подписка queue group `arm-core` на `files.tabular_parsed` (по умолчанию; задаётся OPENBAO__NATS__NATS_*_SUBJECT). Обмен — `services/recognition_exchange.py`: `publish_pending`
(outbox), `expire_deadlines` (вахтёр сроков), `handle_message` (inbox: ack — конверт `processing`, результат);
публикатор и вахтёр — фоновые задачи lifespan. Запрос без ack повторно публикуется (`ACK_TIMEOUT_SECONDS`=30 ×
`OUTBOX_MAX_REPUBLISH`=3, счётчик — `recognition_job.metrics.republish_count`) → `dlq`; дедуп — `request_msg_id` /
@ -114,7 +114,7 @@
(`lock_visible_version`/`lock_editable_version`: версия `FOR UPDATE` → видимость 404 → статус «в работе» → своя блокировка),
`consolidation_locks` (взять/перехватить/снять; без истечения, OQ-119), `consolidation_status` (`transition` + ручной
`POST …/status`), `consolidation_documents` (`fits`, `bind_confirmed`, `unbind_entity`, хуки `on_document_confirmed(changed_by, comment)` /
`on_documents_superseded` / `on_documents_mapped`; `bound_versions` / `unbind_documents` — снятие привязок автоподтверждённых документов при удалении файла, OQ-157), `consolidated_osv` (единая сборка `_build`: страница, сводка, кандидаты,
`on_documents_superseded` / `on_documents_mapped`; `bound_versions` / `unbind_documents` — снятие привязок автоподтверждённых документов при удалении файла, OQ-157; `for_consolidation` — ВРЕМЕННЫЙ режим «период ОСВ из версии» `CONSOLIDATION_PERIOD_FROM_VERSION`, `domain/version_period.py`, arm-core#103: привязка, свод, чек-лист и `osv_accounts` читают период документа только через него), `consolidated_osv` (единая сборка `_build`: страница, сводка, кандидаты,
экспорт), `counterparty_mapping` (ручное сопоставление, алиас, автоприменение), `osv_export` (openpyxl, лист «Пробелы»).
Создание и `PUT …/members` проверяют периметр одной функцией `_validate_perimeter` (#42); ячейки чек-листа самоописывающие
(`document_type`, `account`); `members[].osv_accounts` — `repositories/consolidations.osv_accounts` («что загружено к периоду

View File

@ -129,7 +129,7 @@ curl -s localhost:8000/healthcheck
```
```json
{"uptime":"2026-10-03T18:37:15.121394+00:00","version":"1.0.0","nats":"ok"}
{"uptime":"2026-10-03T18:37:15.121394+00:00","version":"1.0.0","nats":"ok","subjects":{"tabular_request":"files.tabular_uploaded","image_request":"files.img_uploaded","result":"files.tabular_parsed"}}
```
`nats`: `ok` — подключён, `down` — включён (`NATS_ENABLED=true`), но связи нет, `off` — выключен.
@ -712,6 +712,13 @@ Q2–Q4 не подходят (R-14a). Привязываются только `
и удаление ЮЛ из периметра. Версии `CANCELLED` не трогаются.
- **Удаление файла с документом в версии:** файл с документом, подтверждённым пользователем, удалить нельзя; файл с автоподтверждённым документом — можно, пока все версии с его привязками «в работе» (привязки снимаются), иначе 409 `FILE_HAS_CONFIRMED_DOCUMENTS` (OQ-110, уточнено OQ-157).
- `period_check` документа остаётся `UNKNOWN` — подходящесть — функция (документ, версия), а не свойство документа (OQ-120).
- **Временный режим «период ОСВ из версии»** (`CONSOLIDATION_PERIOD_FROM_VERSION=true`, в контуре —
`OPENBAO__APP__CONSOLIDATION_PERIOD_FROM_VERSION`; по умолчанию выключен; arm-core#103, OQ-161): ОСВ, загруженная в
консолидацию (`stored_file.consolidation_id`), период шапки которой версии **не подходит**, считается нарастающей за год
и квартал консолидации — для привязки, свода, чек-листа и колонки «ОСВ» периметра (`src/domain/version_period.py`).
ОСВ, подходящие по шапке, не трогаются (квартальные склеиваются как раньше). Документ не меняется (карточка показывает
период из шапки); у подменённых ОСВ склейки нет — на счёт ЮЛ один слот, побеждает подтверждённая последней; ББ/ОФР и
файлы с рабочего стола режим не затрагивает. Только для тестов на произвольных ОСВ.
### Чек-лист
@ -915,6 +922,9 @@ email без учёта регистра проверяет только сер
| `OPENBAO__NATS__NATS_URL` | — (обязательно) | адрес NATS (`nats://localhost:4222`) |
| `OPENBAO__NATS__NATS_SEED` | пусто | nkey seed; пусто — подключение без учётных данных (compose, тесты). Задаётся вместе с `NATS_JWT` или не задаётся вовсе |
| `OPENBAO__NATS__NATS_JWT` | пусто | JWT NATS; см. `NATS_SEED` |
| `OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT` | `files.tabular_uploaded` | subject запроса распознавания таблиц (публикация) |
| `OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT` | `files.img_uploaded` | subject запроса распознавания изображений (публикация) |
| `OPENBAO__NATS__NATS_RESULT_SUBJECT` | `files.tabular_parsed` | subject ответов (подписка, queue group `arm-core`). Три имени различны, без пробелов, `*`, `>` и пустых токенов; нарушение — отказ старта. Действующие значения — в логе старта и в `/healthcheck.subjects` |
| `OPENBAO__SMB__SMB_HOST` | пусто | SMB из шаблона, необязателен (ядру в Срезе 0 не нужен). Пусто — сессия SMB не регистрируется; задан — регистрация при старте (в отдельном потоке) и сброс кэша соединений при остановке |
| `OPENBAO__SMB__SMB_PATH` | пусто | путь SMB (в Срезе 0 не используется) |
| `OPENBAO__SMB__SMB_LOGIN` | пусто | логин SMB |
@ -927,6 +937,7 @@ email без учёта регистра проверяет только сер
| `UPLOAD_MAX_BYTES` | `52428800` (50 МиБ) | предельный размер файла первичной отчётности при загрузке; больше — `FILE_TOO_LARGE` в `rejected` |
| `MAPPING_POLL_SECONDS` | `5` | период опроса цикла маппинга результатов распознавания |
| `MAPPING_MAX_ATTEMPTS` | `3` | число попыток маппинга файла до `ERROR` |
| `CONSOLIDATION_PERIOD_FROM_VERSION` | `false` | **временно**, для тестов на произвольных ОСВ: ОСВ, загруженная в консолидацию, берёт год и квартал консолидации (раздел «Привязка документов», arm-core#103); в контуре — `OPENBAO__APP__CONSOLIDATION_PERIOD_FROM_VERSION` |
| `AUTH_DEV_ISSUER` | `false` | dev-эмитент токенов: работает **только** при `AUTH_DEV_ISSUER=true` **и** `DEBUG=true` |
| `AUTH_DEV_SECRET` | пусто | секрет HS256 dev-эмитента (не короче 32 байт); только для разработки |
| `AUTH_JWKS_URL` или `AUTH_PUBLIC_KEY` | пусто | проверка токенов РАИСА: JWKS по URL либо публичный ключ (PEM) |
@ -993,7 +1004,7 @@ tabular-parser (MinIO) → inbox → маппинг → `GET /files/{id}` = `PRO
```bash
docker compose -f ../docs/dev/docker-compose.yml -p arm-ka --profile services up -d --build --no-deps tabular-parser
curl -s localhost:8001/healthcheck # {"uptime": "...", "version": "...", "nats": "ok"}
curl -s localhost:8001/healthcheck # {"uptime": "...", "version": "...", "nats": "ok", "subjects": {"tabular_request": "...", "result": "..."}}
make e2e # PASSED
docker compose -f ../docs/dev/docker-compose.yml -p arm-ka --profile services stop tabular-parser
```

View File

@ -37,6 +37,10 @@ OPENBAO__MINIO__MINIO_SECURE=false
OPENBAO__NATS__NATS_URL=nats://localhost:4222
OPENBAO__NATS__NATS_SEED=
OPENBAO__NATS__NATS_JWT=
# subject'ы обмена — по окружению (прод/тест), у трёх сервисов одинаковые; умолчания — имена контракта
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
# SMB необязателен (ядру в Срезе 0 не нужен): пустой SMB_HOST — сессия SMB не регистрируется
OPENBAO__SMB__SMB_HOST=
@ -78,3 +82,6 @@ UPLOAD_MAX_BYTES=52428800
# Маппинг результата распознавания: период опроса готовых заданий (секунды) и число попыток до отметки ошибки
MAPPING_POLL_SECONDS=5
MAPPING_MAX_ATTEMPTS=3
# ВРЕМЕННО, для тестов на произвольных ОСВ (arm-core#103): ОСВ, загруженная в консолидацию, считается нарастающей за год и
# квартал консолидации, а не за период из шапки. В контуре — OPENBAO__APP__CONSOLIDATION_PERIOD_FROM_VERSION
CONSOLIDATION_PERIOD_FROM_VERSION=false

View File

@ -23,6 +23,9 @@ OPENBAO__MINIO__MINIO_SECURE=false
OPENBAO__NATS__NATS_URL=nats://localhost:4222
OPENBAO__NATS__NATS_SEED=
OPENBAO__NATS__NATS_JWT=
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
# SMB необязателен: пустой SMB_HOST — сессия SMB не регистрируется
OPENBAO__SMB__SMB_HOST=
@ -58,3 +61,5 @@ IMPORT_MAX_BYTES=20971520
UPLOAD_MAX_BYTES=52428800
MAPPING_POLL_SECONDS=5
MAPPING_MAX_ATTEMPTS=3
# временный режим «период ОСВ из версии консолидации» (arm-core#103); тесты включают его через monkeypatch
CONSOLIDATION_PERIOD_FROM_VERSION=false

View File

@ -2,7 +2,7 @@
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "arm-ka/recognition-request/v1",
"title": "Запрос распознавания файла (arm-core → tabular-parser / scan-gateway), контракт v1.0",
"description": "Тело сообщения на subject files.tabular_uploaded или files.img_uploaded; subject должен соответствовать полю pipeline (tabular → files.tabular_uploaded, image → files.img_uploaded). Заголовок Nats-Msg-Id = <job_id>:<attempt> (не поле тела). Один файл (stored_file) → один job.",
"description": "Тело сообщения на subject запроса; subject — настройка окружения и должен соответствовать полю pipeline (tabular → NATS_TABULAR_REQUEST_SUBJECT, по умолчанию files.tabular_uploaded; image → NATS_IMAGE_REQUEST_SUBJECT, по умолчанию files.img_uploaded). Заголовок Nats-Msg-Id = <job_id>:<attempt> (не поле тела). Один файл (stored_file) → один job.",
"$comment": "Версионирование: schema_version = MAJOR.MINOR. MINOR (1.1) — необязательные поля и новые значения enum; MAJOR (2.0) — удаление/переименование полей, смена типов или семантики. Потребитель принимает тот же MAJOR и любой MINOR. Новый MINOR раскатывается сначала на потребителей (ядро и сервисы обновляют копии схем раньше, чем производитель начнёт слать 1.1), потому что закрытые enum отвергают незнакомые значения.",
"type": "object",
"required": ["schema_version", "job_id", "file_id", "pipeline", "file_type", "original_name", "s3", "reply_deadline", "attempt"],

View File

@ -2,7 +2,7 @@
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "arm-ka/recognition-result-message/v1",
"title": "Сообщение результата распознавания (tabular-parser / scan-gateway → arm-core), контракт v1.0",
"description": "Тело сообщения на subject files.tabular_parsed. Заголовок Nats-Msg-Id (не поле тела) на этом subject имеет две формы: ack:<job_id>:<attempt> — для status = processing; result:<job_id>:<attempt> — для done, needs_check и error. Форма запроса <job_id>:<attempt> описана в схеме recognition-request. status = processing — ack «взял в работу», публикуется до чтения файла из S3. group_schema в конверт не входит: источник — файл результата (recognition-result).",
"description": "Тело сообщения на subject ответа (настройка окружения NATS_RESULT_SUBJECT, по умолчанию files.tabular_parsed). Заголовок Nats-Msg-Id (не поле тела) на этом subject имеет две формы: ack:<job_id>:<attempt> — для status = processing; result:<job_id>:<attempt> — для done, needs_check и error. Форма запроса <job_id>:<attempt> описана в схеме recognition-request. status = processing — ack «взял в работу», публикуется до чтения файла из S3. group_schema в конверт не входит: источник — файл результата (recognition-result).",
"$comment": "Версионирование: schema_version = MAJOR.MINOR. MINOR (1.1) — необязательные поля и новые значения enum; MAJOR (2.0) — удаление/переименование полей, смена типов или семантики. Потребитель принимает тот же MAJOR и любой MINOR. Новый MINOR раскатывается сначала на потребителей (ядро и сервисы обновляют копии схем раньше, чем производитель начнёт слать 1.1), потому что закрытые enum отвергают незнакомые значения.",
"type": "object",
"required": ["schema_version", "job_id", "file_id", "status", "service"],

View File

@ -1,17 +1,19 @@
"""Точка входа FastAPI. Каркас из шаблона команды: инфраструктура без доменного кода.
Что здесь есть: настройки, подключение S3 (MinIO; `MINIO_SECURE` — https или http), NATS (`src/core/nats.py`: core
NATS без JetStream, подключение в фоне при NATS_ENABLED=true независимо от DEBUG, подписка `files.tabular_parsed`
в queue group `arm-core`), SMB (горячие папки ИС Сканирование; необязателен — при пустом `SMB_HOST` не подключается),
middleware (gzip, proxy headers, CORS по `cors_options`; при DEBUG=False пакет РАИСА лишь условно импортируется,
middleware не подключается), healthcheck (uptime, version, nats: ok|down|off), раздача собранного SPA. Документация API
(`/docs`, `/redoc`, `/openapi.json`) — только при DEBUG (OQ-66). Приложение собирает `create_app(settings)`.
NATS без JetStream, подключение в фоне при NATS_ENABLED=true независимо от DEBUG, подписка на subject ответов (по
умолчанию `files.tabular_parsed`) в queue group `arm-core`), SMB (горячие папки ИС Сканирование; необязателен — при
пустом `SMB_HOST` не подключается), middleware (gzip, proxy headers, CORS по `cors_options`; при DEBUG=False пакет
РАИСА лишь условно импортируется, middleware не подключается), healthcheck (uptime, version, nats: ok|down|off,
subjects), раздача собранного SPA. Документация API (`/docs`, `/redoc`, `/openapi.json`) — только при DEBUG (OQ-66).
Приложение собирает `create_app(settings)`.
Недоступность S3 или NATS при старте не роняет и не задерживает приложение: ошибка логируется, healthcheck показывает
состояние. При NATS_ENABLED=true lifespan запускает фоновые задачи обмена `recognition_job` — outbox-публикатор,
вахтёр сроков (`src/services/recognition_exchange.py`, design D-6) и, если S3 подключено, цикл маппинга результатов
(`src/services/result_mapping.py`, Срез 2c D-1) — и останавливает их до закрытия NATS. Сообщения
`files.tabular_parsed` (ack и результаты сервисов) обрабатывает inbox `handle_message` того же модуля с фабрикой сессий
приложения (inbox меняет задание и статус файла; маппинг результата в доменные таблицы — отдельный цикл, Срез 2c D-1).
subject ответов (по умолчанию `files.tabular_parsed`; ack и результаты сервисов) обрабатывает inbox
`handle_message` того же модуля с фабрикой сессий приложения (inbox меняет задание и статус файла; маппинг
результата в доменные таблицы — отдельный цикл, Срез 2c D-1).
Остановка: фоновые задачи обмена → NATS → S3 → SMB → пул соединений БД; каждый шаг независим — ошибка одного
логируется и не отменяет остальные. Логгер настраивает `setup_logger` при импорте модуля: без значений локальных
переменных в трейсбеках (#63).
@ -43,7 +45,7 @@ from src.core.clients import S3Storage
from src.core.config import Settings, get_settings
from src.core.exception_handlers import register_exception_handlers
from src.core.logger import setup_logger
from src.core.nats import MessageHandler, NatsConnection
from src.core.nats import MessageHandler, NatsConnection, used_subjects
from src.db.db import async_session_maker, engine, session_transaction
from src.services.recognition_exchange import ExchangeTasks, handle_message
from src.services.session_audit import SessionAudit
@ -109,13 +111,14 @@ async def _connect_s3() -> S3Storage | None:
def _result_handler() -> MessageHandler:
"""Inbox `files.tabular_parsed`: `handle_message` с фабрикой сессий приложения (маппинг результата — отдельный цикл,
Срез 2c design D-1)."""
"""Inbox subject ответов (по умолчанию `files.tabular_parsed`): `handle_message` с фабрикой сессий приложения
(маппинг результата — отдельный цикл, Срез 2c design D-1)."""
return partial(handle_message, session_factory=async_session_maker)
async def _subscribe_results(nats: NatsConnection) -> None:
"""Подписки после первого подключения к NATS: `files.tabular_parsed` в queue group `arm-core`."""
"""Подписки после первого подключения к NATS: subject ответов (по умолчанию `files.tabular_parsed`) в queue
group `arm-core`."""
await nats.subscribe_results(_result_handler())
@ -183,6 +186,7 @@ async def lifespan(app: FastAPI):
app.state.nats = None
app.state.exchange = None
logger.info("Subject'ы NATS. {subjects}", subjects=used_subjects(settings.OPENBAO.NATS))
await _register_smb()
try:
@ -241,7 +245,12 @@ def create_app(app_settings: Settings, *, web_dir: str = "web") -> FastAPI:
if app_settings.NATS_ENABLED:
nc = getattr(request.app.state, "nats", None)
nats_status = "ok" if nc is not None and nc.is_connected else "down"
return {"uptime": UP_TIME, "version": VERSION, "nats": nats_status}
return {
"uptime": UP_TIME,
"version": VERSION,
"nats": nats_status,
"subjects": used_subjects(app_settings.OPENBAO.NATS),
}
application.include_router(build_api_router(app_settings))
# аудит входа (OQ-64): LOGIN первого запроса и LOGIN_FAILED — в отдельной транзакции, множество входов — на процесс

View File

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

View File

@ -3,19 +3,27 @@ import os
import re
from typing import Annotated, Any
from pydantic import NonNegativeInt, PositiveInt, field_validator, model_validator
from pydantic import NonNegativeInt, PositiveInt, SecretStr, ValidationInfo, field_validator, model_validator
from pydantic_settings import BaseSettings, NoDecode, PydanticBaseSettingsSource, SettingsConfigDict
from src.contract.subjects import IMAGE_REQUEST_SUBJECT, RESULT_SUBJECT, TABULAR_REQUEST_SUBJECT
from src.core.settings_sources import OpenBaoAppSettingsSource
DB_SCHEMA_RE = re.compile(r"[a-z_][a-z0-9_]{0,62}")
# 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 DBSettings(BaseSettings):
model_config = SettingsConfigDict(hide_input_in_errors=True)
DATABASE_URL: str
class MinioSettings(BaseSettings):
model_config = SettingsConfigDict(hide_input_in_errors=True)
MINIO_URL: str
MINIO_ADMIN_LOGIN: str
MINIO_ADMIN_PASSWORD: str
@ -24,15 +32,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 (локальная разработка, тесты)
NATS_JWT: str = ""
NATS_SEED: SecretStr = SecretStr("") # пусто — подключение к NATS без nkeys (локальная разработка, тесты)
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 (горячие папки ИС Сканирование, шаблон). Необязательны: ядру в Срезе 0 SMB не нужен; пустой `SMB_HOST` —
сессия SMB не регистрируется."""
model_config = SettingsConfigDict(hide_input_in_errors=True)
SMB_HOST: str = ""
SMB_PATH: str = ""
SMB_LOGIN: str = ""
@ -40,6 +90,8 @@ class SMBSettings(BaseSettings):
class OpenBAOSettings(BaseSettings):
model_config = SettingsConfigDict(hide_input_in_errors=True)
DATABASE: DBSettings
MINIO: MinioSettings
NATS: NATSSettings
@ -57,6 +109,8 @@ class OpenBAOSettings(BaseSettings):
class ServerSettings(BaseSettings):
model_config = SettingsConfigDict(hide_input_in_errors=True)
HOST: str = "0.0.0.0"
PORT: int = 8000
@ -112,6 +166,11 @@ class Settings(BaseSettings):
MAPPING_POLL_SECONDS: PositiveInt = 5
MAPPING_MAX_ATTEMPTS: PositiveInt = 3
# ВРЕМЕННЫЙ режим для тестов на произвольных ОСВ (arm-core#103): ОСВ, загруженная в консолидацию, считается
# нарастающей за год и квартал консолидации, а не за период из шапки (`domain/version_period.py`). В контуре —
# OPENBAO__APP__CONSOLIDATION_PERIOD_FROM_VERSION. Снять вместе с предупреждением о несовпадении периода
CONSOLIDATION_PERIOD_FROM_VERSION: bool = False
@classmethod
def settings_customise_sources(
cls,
@ -157,6 +216,7 @@ class Settings(BaseSettings):
env_file=os.environ.get("ENV_FILE", ".env"),
env_nested_delimiter="__",
nested_model_default_partial_update=False,
hide_input_in_errors=True,
)

View File

@ -4,8 +4,9 @@
- Аутентификация как в шаблоне: заданы `NATS_SEED` и `NATS_JWT` — nkeys/JWT; оба пусты — без учётных данных.
- `publish(subject, payload, msg_id)` — публикация с заголовком `Nats-Msg-Id` и `flush`: ошибка соединения проявляется
сразу исключением `NatsPublishError` (outbox оставляет задание в `queued`).
- `subscribe_results(handler)` — подписка на `files.tabular_parsed` в queue group `arm-core`: из нескольких инстансов
ядра сообщение получает ровно один. Исключения обработчика логируются, подписка не падает.
- `subscribe_results(handler)` — подписка на `NATS_RESULT_SUBJECT` настроек (умолчание `files.tabular_parsed`) в queue
group `arm-core`: из нескольких инстансов ядра сообщение получает ровно один. Исключения обработчика
логируются, подписка не падает.
- Stream'ов и consumer'ов JetStream нет. Переподключение после разрыва — встроенное в nats-py
(`max_reconnect_attempts=-1`, `reconnect_time_wait=2`), подписки nats-py восстанавливает сам.
- Первое подключение выполняет фоновый supervisor (`start()`): старт приложения не ждёт NATS. nats-py 2.14 с
@ -27,7 +28,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, RESULT_SUBJECT
from src.contract.subjects import NATS_MSG_ID_HEADER
from src.core.config import NATSSettings
RESULT_QUEUE_GROUP = "arm-core"
@ -60,14 +61,23 @@ class NatsPublishError(NatsError):
"""Публикация не подтверждена: соединения нет или сервер не ответил на flush."""
def used_subjects(settings: NATSSettings) -> dict[str, str]:
"""Subject'ы ядра для лога старта и `/healthcheck.subjects`: публикует оба запроса, подписано на ответ."""
return {
"tabular_request": settings.NATS_TABULAR_REQUEST_SUBJECT,
"image_request": settings.NATS_IMAGE_REQUEST_SUBJECT,
"result": settings.NATS_RESULT_SUBJECT,
}
def build_nats_auth(settings: NATSSettings) -> dict[str, Any]:
"""Параметры аутентификации для `Client.connect`.
Пустые `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:
@ -95,7 +105,7 @@ class NatsConnection:
`on_connected(conn)` вызывается один раз после первого успешного подключения (подписки); при исключении
повторяется с той же экспоненциальной паузой. `result_subject` и `queue` параметризуются для тестов —
по умолчанию `files.tabular_parsed` и `arm-core`.
по умолчанию `files.tabular_parsed` (фактический subject задаётся настройками окружения) и `arm-core`.
"""
def __init__( # noqa: PLR0913 - все параметры, кроме settings, именованные: подмены для тестов и тайминги
@ -103,7 +113,7 @@ class NatsConnection:
settings: NATSSettings,
*,
on_connected: Callable[["NatsConnection"], Awaitable[None]] | None = None,
result_subject: str = RESULT_SUBJECT,
result_subject: str | None = None,
queue: str = RESULT_QUEUE_GROUP,
client_factory: Callable[[], Any] = NATS,
sleep: SleepFn = asyncio.sleep,
@ -114,7 +124,7 @@ class NatsConnection:
) -> None:
self._settings = settings
self._on_connected = on_connected
self.result_subject = result_subject
self.result_subject = result_subject or settings.NATS_RESULT_SUBJECT
self.queue = queue
self._client_factory = client_factory
self._sleep = sleep

View File

@ -0,0 +1,94 @@
"""ВРЕМЕННЫЙ режим «период ОСВ из версии консолидации» (arm-core#103, настройка `CONSOLIDATION_PERIOD_FROM_VERSION`).
Для тестов на произвольных ОСВ: документ ОСВ, загруженный в консолидацию (`stored_file.consolidation_id`), период шапки
которого версии НЕ подходит, считается нарастающим за год и квартал этой консолидации. Документ, подходящий версии по
шапке, остаётся как есть — квартальные ОСВ по-прежнему склеиваются, уже привязанные не задваиваются. Первичка не
мутируется: подменяется только представление документа (`VersionPeriodDocument`) при привязке к версии (`fits`, ключ
замены R-1), сборке свода (`select_sources`), чек-листе и колонке «ОСВ» периметра; карточка документа показывает период
из шапки.
Следствие: подменённые ОСВ одного счёта одного ЮЛ занимают один слот привязки (нарастающий за квартал версии), побеждает
подтверждённая последней, и склейки кварталов у них нет. ББ/ОФР и файлы без консолидации (рабочий стол) режим не
затрагивает.
ASSUMPTION(OQ-161): правила в `05-domain-rules.md` нет — режим временный, по решению владельца 12.10.2026; по умолчанию
выключен, без настройки поведение §2 / D-5 Среза 4 не меняется.
"""
import datetime
import uuid
from dataclasses import dataclass
from typing import Protocol
from src.models.enums import DocumentStatus, DocumentType, PeriodQuarter
class DocumentFields(Protocol):
"""Поля документа, которые читают привязка, свод и чек-лист (структурный тип: `Document` подходит)."""
@property
def id(self) -> uuid.UUID: ...
@property
def legal_entity_id(self) -> uuid.UUID: ...
@property
def document_type(self) -> DocumentType: ...
@property
def osv_account_code(self) -> str | None: ...
@property
def period_year(self) -> int | None: ...
@property
def period_quarter(self) -> PeriodQuarter | None: ...
@property
def is_cumulative(self) -> bool | None: ...
@property
def status(self) -> DocumentStatus: ...
@property
def confirmed_at(self) -> datetime.datetime | None: ...
@dataclass(frozen=True)
class VersionPeriodDocument:
"""Документ ОСВ с периодом версии: поля документа как есть, период — год и квартал консолидации, нарастающий."""
id: uuid.UUID
legal_entity_id: uuid.UUID
document_type: DocumentType
osv_account_code: str | None
period_year: int
period_quarter: PeriodQuarter
is_cumulative: bool
status: DocumentStatus
confirmed_at: datetime.datetime | None
def takes_version_period(
*,
enabled: bool,
document_type: DocumentType,
file_consolidation_id: uuid.UUID | None,
consolidation_id: uuid.UUID,
fits_by_header: bool,
) -> bool:
"""Берёт ли документ период версии: режим включён, документ — ОСВ, его файл загружен в эту консолидацию, а период
шапки версии не подходит (`fits_by_header` — период шапки подходит версии, как в `fits` без проверки периметра)."""
return (
enabled
and document_type == DocumentType.OSV
and file_consolidation_id == consolidation_id
and not fits_by_header
)
def with_version_period(document: DocumentFields, *, year: int, quarter: PeriodQuarter) -> VersionPeriodDocument:
"""Представление `document` как нарастающего за `year`/`quarter` версии; сам документ не меняется."""
return VersionPeriodDocument(
id=document.id,
legal_entity_id=document.legal_entity_id,
document_type=document.document_type,
osv_account_code=document.osv_account_code,
period_year=year,
period_quarter=quarter,
is_cumulative=True,
status=document.status,
confirmed_at=document.confirmed_at,
)

View File

@ -10,7 +10,7 @@ import uuid
from collections import defaultdict
from collections.abc import Collection, Sequence
from sqlalchemy import Select, case, exists, false, func, or_, select
from sqlalchemy import ColumnElement, Select, and_, case, exists, false, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from src.domain.checklist import EntityFilesState
@ -381,19 +381,43 @@ async def rules_for(
return [(row[0], row[1]) for row in await session.execute(stmt)]
async def unconfirmed_documents(session: AsyncSession, entity_ids: Collection[uuid.UUID], year: int) -> list[Document]:
def _osv_of_consolidation(consolidation_id: uuid.UUID) -> ColumnElement[bool]:
"""ОСВ, файл которой загружен в консолидацию: в режиме «период из версии» (arm-core#103) — любого периода шапки."""
return and_(Document.document_type == DocumentType.OSV, StoredFile.consolidation_id == consolidation_id)
async def file_consolidation_ids(
session: AsyncSession, file_ids: Collection[uuid.UUID]
) -> dict[uuid.UUID, uuid.UUID | None]:
"""`stored_file.consolidation_id` по id файлов одним запросом (режим «период из версии», arm-core#103)."""
if not file_ids:
return {}
stmt = select(StoredFile.id, StoredFile.consolidation_id).where(StoredFile.id.in_(list(file_ids)))
return {file_id: consolidation_id for file_id, consolidation_id in await session.execute(stmt)}
async def unconfirmed_documents(
session: AsyncSession,
entity_ids: Collection[uuid.UUID],
year: int,
consolidation_id: uuid.UUID | None = None,
) -> list[Document]:
"""Неподтверждённые документы ЮЛ за год (`RECOGNIZED`/`NEEDS_CHECK`) из файлов не `SUPERSEDED`; любых типов (R-5).
`consolidation_id` (режим «период из версии», arm-core#103) — плюс ОСВ, загруженные в эту консолидацию, любого года.
Кандидаты чек-листа: «подходящесть» версии (`fits`) проверяет сервис. Порядок — по ЮЛ, типу, счёту, кварталу, `id`.
"""
if not entity_ids:
return []
in_period: ColumnElement[bool] = Document.period_year == year
if consolidation_id is not None:
in_period = or_(in_period, _osv_of_consolidation(consolidation_id))
stmt = (
select(Document)
.join(StoredFile, StoredFile.id == Document.stored_file_id)
.where(
Document.legal_entity_id.in_(list(entity_ids)),
Document.period_year == year,
in_period,
Document.status.in_(UNCONFIRMED_STATUSES),
StoredFile.status != FileStatus.SUPERSEDED,
)
@ -409,22 +433,29 @@ async def unconfirmed_documents(session: AsyncSession, entity_ids: Collection[uu
async def osv_accounts(
session: AsyncSession, entity_ids: Collection[uuid.UUID], year: int, quarters: Collection[PeriodQuarter]
session: AsyncSession,
entity_ids: Collection[uuid.UUID],
year: int,
quarters: Collection[PeriodQuarter],
consolidation_id: uuid.UUID | None = None,
) -> dict[uuid.UUID, list[str]]:
"""Счета ОСВ, загруженные к периоду версии (колонка «ОСВ» панели [34]; front-api-gaps D-3): уникальные
`osv_account_code` документов `OSV` ЮЛ за год `year` и кварталы `quarters`; документ не `ERROR`, файл не
`SUPERSEDED`/`ERROR`, квартал и код счёта заданы; статус подтверждения и привязка к версии не важны. Порядок —
натуральный (`account_sort_key`); ЮЛ без документов в словаре нет."""
натуральный (`account_sort_key`); ЮЛ без документов в словаре нет. `consolidation_id` (режим «период из версии»,
arm-core#103) — плюс ОСВ, загруженные в эту консолидацию, любого периода."""
if not entity_ids:
return {}
in_period: ColumnElement[bool] = and_(Document.period_year == year, Document.period_quarter.in_(list(quarters)))
if consolidation_id is not None:
in_period = or_(in_period, _osv_of_consolidation(consolidation_id))
stmt = (
select(Document.legal_entity_id, Document.osv_account_code)
.join(StoredFile, StoredFile.id == Document.stored_file_id)
.where(
Document.legal_entity_id.in_(list(entity_ids)),
Document.document_type == DocumentType.OSV,
Document.period_year == year,
Document.period_quarter.in_(list(quarters)),
in_period,
Document.status != DocumentStatus.ERROR,
StoredFile.status.not_in([FileStatus.SUPERSEDED, FileStatus.ERROR]),
Document.osv_account_code.is_not(None),

View File

@ -239,17 +239,24 @@ async def list_legal_entity_history(session: AsyncSession, entity_id: uuid.UUID)
return list((await session.execute(stmt)).scalars().all())
async def current_memberships(
session: AsyncSession, entity_id: uuid.UUID
) -> list[tuple[RelatedGroupMember, RelatedGroup]]:
"""Текущие членства ЮЛ в ГСК/ГСЗ вместе с группой (для карточки), по виду и коду группы."""
async def current_memberships_by_entities(
session: AsyncSession, entity_ids: Iterable[uuid.UUID]
) -> dict[uuid.UUID, list[tuple[RelatedGroupMember, RelatedGroup]]]:
"""Текущие членства ЮЛ в ГСК/ГСЗ вместе с группой по набору ЮЛ одним запросом (#109); у каждого ЮЛ — по виду и
коду группы; ЮЛ без членств в словаре нет."""
ids = set(entity_ids)
if not ids:
return {}
stmt = (
select(RelatedGroupMember, RelatedGroup)
.join(RelatedGroup, RelatedGroup.id == RelatedGroupMember.related_group_id)
.where(RelatedGroupMember.legal_entity_id == entity_id, RelatedGroupMember.is_current.is_(True))
.where(RelatedGroupMember.legal_entity_id.in_(ids), RelatedGroupMember.is_current.is_(True))
.order_by(RelatedGroupMember.kind, RelatedGroup.code)
)
return [(member, group) for member, group in (await session.execute(stmt)).all()]
result: dict[uuid.UUID, list[tuple[RelatedGroupMember, RelatedGroup]]] = {}
for member, group in (await session.execute(stmt)).all():
result.setdefault(member.legal_entity_id, []).append((member, group))
return result
async def close_memberships(

View File

@ -253,10 +253,21 @@ class LegalEntityWithdraw(BaseModel):
comment: str | None = None
class MembershipOut(BaseModel):
"""Текущее членство ЮЛ в ГСК/ГСЗ (список, карточка и ответы ЮЛ, #109)."""
related_group_id: uuid.UUID
code: str
kind: RelatedGroupKind
name: str
valid_from: datetime.datetime
class LegalEntityOut(BaseModel):
"""ЮЛ/ИП; `outside_group` — монозаёмщик (`company_group_id IS NULL`), `closed` — заполнен `close_date`.
`documents_count` — число документов файлов ЮЛ со статусом файла не SUPERSEDED (D-13).
`documents_count` — число документов файлов ЮЛ со статусом файла не SUPERSEDED (D-13); `memberships` — текущие
членства ГСК/ГСЗ (по виду и коду группы; после перевода и вывода из ГК — пусто, #109).
"""
model_config = ConfigDict(from_attributes=True)
@ -276,6 +287,7 @@ class LegalEntityOut(BaseModel):
created_at: datetime.datetime | None
updated_at: datetime.datetime | None
documents_count: int = 0
memberships: list[MembershipOut] = []
@computed_field # type: ignore[prop-decorator]
@property
@ -288,16 +300,6 @@ class LegalEntityOut(BaseModel):
return self.close_date is not None
class MembershipOut(BaseModel):
"""Текущее членство ЮЛ в ГСК/ГСЗ (для карточки ЮЛ)."""
related_group_id: uuid.UUID
code: str
kind: RelatedGroupKind
name: str
valid_from: datetime.datetime
class AliasCreate(BaseModel):
"""Тело `POST …/aliases`: `alias` сохраняется как есть (как в ОСВ), `normalized_alias` считает сервис (§3)."""
@ -316,9 +318,8 @@ class AliasOut(BaseModel):
class LegalEntityDetail(LegalEntityOut):
"""Карточка ЮЛ: текущие членства ГСК/ГСЗ и алиасы."""
"""Карточка ЮЛ: поля списка (в т. ч. `memberships`) и алиасы."""
memberships: list[MembershipOut] = []
aliases: list[AliasOut] = []
@ -360,33 +361,51 @@ class LegalEntityFilter(BaseModel):
class RelatedGroupCreate(BaseModel):
"""Создание ГСК/ГСЗ в ГК: `code` (`GSK-`/`GSZ-NNNNNN`) назначает сервис."""
"""Создание ГСК/ГСЗ в ГК: `code` (`GSK-`/`GSZ-NNNNNN`) назначает сервис.
`crm_id` («ID ГСК/ГСЗ в CRM» форм 51-03/51-04) принимается и отбрасывается: колонки в DBML нет (arm-core #112).
"""
model_config = ConfigDict(extra="forbid")
kind: RelatedGroupKind
name: str = Field(min_length=1, max_length=NAME_MAX_LENGTH)
created_date: datetime.date | None = None
crm_id: str | None = None
@field_validator("name")
@classmethod
def _strip_name(cls, value: str) -> str | None:
return _strip_required_name(value)
@field_validator("crm_id")
@classmethod
def _normalize_crm_id(cls, value: str | None) -> str | None:
return _blank_to_none(value)
class RelatedGroupUpdate(BaseModel):
"""Изменение ГСК/ГСЗ: `name`, `created_date` (вид и ГК не меняются); переданные поля — `model_fields_set`."""
"""Изменение ГСК/ГСЗ: `name`, `created_date` (вид и ГК не меняются); переданные поля — `model_fields_set`.
`crm_id` (форма 77-02) принимается и отбрасывается, как при создании (arm-core #112).
"""
model_config = ConfigDict(extra="forbid")
name: str | None = Field(default=None, min_length=1, max_length=NAME_MAX_LENGTH)
created_date: datetime.date | None = None
crm_id: str | None = None
@field_validator("name")
@classmethod
def _strip_name(cls, value: str | None) -> str | None:
return _strip_required_name(value)
@field_validator("crm_id")
@classmethod
def _normalize_crm_id(cls, value: str | None) -> str | None:
return _blank_to_none(value)
@model_validator(mode="after")
def _name_is_not_null(self) -> Self:
if "name" in self.model_fields_set and self.name is None:
@ -403,7 +422,7 @@ class MemberAdd(BaseModel):
class RelatedGroupOut(BaseModel):
"""ГСК/ГСЗ; `members_count` — число текущих членов."""
"""ГСК/ГСЗ; `members_count` — число текущих членов; `crm_id` — всегда `null`, пока нет колонки (arm-core #112)."""
model_config = ConfigDict(from_attributes=True)
@ -413,6 +432,7 @@ class RelatedGroupOut(BaseModel):
kind: RelatedGroupKind
name: str
created_date: datetime.date | None
crm_id: str | None = None
members_count: int = 0

View File

@ -24,10 +24,10 @@ from src.common.exceptions import DomainValidationError, NotFoundError
from src.domain import consolidated_osv as domain
from src.domain.aliases import normalize_alias
from src.domain.policy import VisibilityScope
from src.domain.version_period import DocumentFields
from src.models.consolidation import Consolidation, ConsolidationVersion
from src.models.directories import CompanyGroup, LegalEntity, RelatedGroup
from src.models.enums import AuditAction, AuditSection, DocumentType, RelatedGroupKind
from src.models.files import Document
from src.models.osv import OsvEntry
from src.repositories import consolidations as repo
from src.repositories.counterparties import CounterpartyIndex
@ -41,6 +41,7 @@ from src.schemas.consolidations import (
OsvSummaryOut,
RefOut,
)
from src.services import consolidation_documents
from src.services.consolidation_guard import VERSION_NOT_FOUND
from src.services.osv_export import build_workbook
@ -76,7 +77,7 @@ def _ref(item: CompanyGroup | RelatedGroup | None) -> RefOut | None:
return None if item is None else RefOut(id=item.id, name=item.name)
def _source_document(document: Document) -> domain.SourceDocument:
def _source_document(document: DocumentFields) -> domain.SourceDocument:
return domain.SourceDocument(
id=document.id,
legal_entity_id=document.legal_entity_id,
@ -142,11 +143,16 @@ class ConsolidatedOsvService:
raise NotFoundError(VERSION_NOT_FOUND)
version, consolidation = found
members = {member.id: _entity(member) for _, member in await repo.members(self._session, version.id)}
documents = [
_source_document(document)
bound = [
document
for document in await repo.bound_documents(self._session, version.id)
if document.document_type == DocumentType.OSV
]
# режим «период из версии» (arm-core#103): ОСВ, загруженные в консолидацию, — с периодом версии
documents = [
_source_document(document)
for document in await consolidation_documents.for_consolidation(self._session, bound, consolidation)
]
# ровно то, что отдал `load_osv_rows`: он оставляет `COUNTERPARTY`/`BU`+`SUM`, домен их не перефильтровывает
rows = [_source_row(entry) for entry, _, _ in await repo.load_osv_rows(self._session, version.id)]
index = await CounterpartyIndex.load(self._session)

View File

@ -25,12 +25,14 @@ from collections.abc import Collection, Iterable, Sequence
from typing import Any
from loguru import logger
from sqlalchemy import delete, select
from sqlalchemy import and_, delete, or_, select
from sqlalchemy.engine import CursorResult
from sqlalchemy.ext.asyncio import AsyncSession
from src.core.config import settings
from src.domain.consolidation_periods import QUARTER_ORDER, counts_as_cumulative, effective_cumulative
from src.domain.policy import IN_WORK_STATUSES
from src.domain.version_period import DocumentFields, takes_version_period, with_version_period
from src.models.consolidation import Consolidation, ConsolidationDocument, ConsolidationMember, ConsolidationVersion
from src.models.enums import ConsolidationStatus, DocumentStatus, DocumentType, FileStatus, PeriodQuarter
from src.models.files import Document, StoredFile
@ -41,9 +43,11 @@ __all__ = [
"bind_confirmed",
"bound_versions",
"fits",
"for_consolidation",
"on_document_confirmed",
"on_documents_mapped",
"on_documents_superseded",
"period_from_version",
"unbind_documents",
"unbind_entity",
]
@ -59,7 +63,36 @@ def _rowcount(result: Any) -> int:
return result.rowcount if isinstance(result, CursorResult) else 0
def _cumulative(document: Document) -> bool:
def period_from_version() -> bool:
"""Включён ли временный режим «период ОСВ из версии консолидации» (`domain/version_period.py`, arm-core#103)."""
return settings.CONSOLIDATION_PERIOD_FROM_VERSION
async def for_consolidation(
session: AsyncSession, documents: Sequence[Document], consolidation: Consolidation
) -> list[DocumentFields]:
"""Документы так, как их видит консолидация: при режиме `period_from_version` ОСВ, загруженные в неё и не подходящие
версии по периоду шапки, — с периодом версии (`with_version_period`), прочие — как есть. Режим выключен —
`documents` без запросов к БД; включён — один запрос `consolidation_id` файлов."""
if not period_from_version() or not documents:
return list(documents)
file_ids = {document.stored_file_id for document in documents}
file_consolidations = await repo.file_consolidation_ids(session, file_ids)
return [
with_version_period(document, year=consolidation.year, quarter=consolidation.quarter)
if takes_version_period(
enabled=True,
document_type=document.document_type,
file_consolidation_id=file_consolidations.get(document.stored_file_id),
consolidation_id=consolidation.id,
fits_by_header=_period_fits(document, consolidation),
)
else document
for document in documents
]
def _cumulative(document: DocumentFields) -> bool:
"""Нарастающий ли документ для привязки и ключа замены (R-1, решение №10, R-14, R-14a): NULL — нарастающий;
ОСВ за `Q1` — квартальная при любом флаге (`effective_cumulative`). Равнозначность `Q1` — только для ОСВ: ББ/ОФР
за `Q1` — отчёт на конец квартала, не источник версий `Q2`–`Q4` (`counts_as_cumulative`, всегда нарастающие)."""
@ -68,7 +101,7 @@ def _cumulative(document: Document) -> bool:
return counts_as_cumulative(document.is_cumulative)
def _key(document: Document) -> BindingKey:
def _key(document: DocumentFields) -> BindingKey:
"""Ключ замены привязки (R-1): ЮЛ, тип, счёт, квартал и признак нарастающего."""
return (
document.legal_entity_id,
@ -79,14 +112,14 @@ def _key(document: Document) -> BindingKey:
)
def _latest_first(document: Document) -> tuple[datetime.datetime, uuid.UUID]:
def _latest_first(document: DocumentFields) -> tuple[datetime.datetime, uuid.UUID]:
"""Порядок «самый поздний `confirmed_at` первым»; равные — по `id` (детерминированно)."""
confirmed_at = document.confirmed_at or datetime.datetime.min.replace(tzinfo=datetime.UTC)
return confirmed_at, document.id
def fits(
document: Document,
document: DocumentFields,
version: ConsolidationVersion, # noqa: ARG001 - сигнатура D-5: подходящесть — функция документа и версии
consolidation: Consolidation,
member_ids: Collection[uuid.UUID],
@ -94,8 +127,15 @@ def fits(
"""Подходит ли документ версии (D-5): ЮЛ — член периметра, год версии, квартал определён и: нарастающий (или
NULL) — квартал версии; квартальный — не позже квартала версии; ОСВ за `Q1` — любой версии года при любом флаге
(R-14; ББ/ОФР за `Q1` — только версии `Q1`, R-14a). Статус документа и файла не проверяется — это
дело вызывающего (привязываются только `CONFIRMED` незаменённых файлов). Чистая функция, без БД."""
if document.legal_entity_id not in member_ids or document.period_year != consolidation.year:
дело вызывающего (привязываются только `CONFIRMED` незаменённых файлов). Период — у переданного представления:
в режиме «период из версии» вызывающий передаёт `for_consolidation(...)`. Чистая функция, без БД."""
return document.legal_entity_id in member_ids and _period_fits(document, consolidation)
def _period_fits(document: DocumentFields, consolidation: Consolidation) -> bool:
"""Периодная часть `fits` (D-5, R-14, R-14a): год версии, квартал определён, нарастающий — за квартал версии,
квартальный — не позже него."""
if document.period_year != consolidation.year:
return False
if document.period_quarter is None:
return False
@ -136,9 +176,14 @@ async def _unbind(session: AsyncSession, version_id: uuid.UUID, document_ids: Co
return _rowcount(result)
async def _bind(session: AsyncSession, version_id: uuid.UUID, document: Document) -> None:
"""Привязывает `document` к версии, предварительно отвязав привязанные документы с тем же ключом (R-1)."""
bound = await _bound_documents(session, version_id, [document.legal_entity_id])
async def _bind(
session: AsyncSession, version_id: uuid.UUID, consolidation: Consolidation, document: DocumentFields
) -> None:
"""Привязывает `document` к версии, предварительно отвязав привязанные документы с тем же ключом (R-1); ключи —
по представлению консолидации (`for_consolidation`)."""
bound = await for_consolidation(
session, await _bound_documents(session, version_id, [document.legal_entity_id]), consolidation
)
await _unbind(session, version_id, [other.id for other in bound if _key(other) == _key(document)])
session.add(ConsolidationDocument(consolidation_version_id=version_id, document_id=document.id))
await session.flush()
@ -162,21 +207,28 @@ async def bind_confirmed(
if not entity_ids:
return 0
members = await _member_ids(session, version.id)
in_period = Document.period_year == consolidation.year
if period_from_version(): # ОСВ, загруженные в консолидацию, — любого периода из шапки
in_period = or_(
in_period,
and_(Document.document_type == DocumentType.OSV, StoredFile.consolidation_id == consolidation.id),
)
stmt = (
select(Document)
.join(StoredFile, StoredFile.id == Document.stored_file_id)
.where(
Document.legal_entity_id.in_(list(entity_ids)),
Document.status == DocumentStatus.CONFIRMED,
Document.period_year == consolidation.year,
in_period,
StoredFile.status != FileStatus.SUPERSEDED,
)
)
candidates = [doc for doc in (await session.execute(stmt)).scalars() if fits(doc, version, consolidation, members)]
winners: dict[BindingKey, Document] = {}
documents = await for_consolidation(session, list((await session.execute(stmt)).scalars()), consolidation)
candidates = [doc for doc in documents if fits(doc, version, consolidation, members)]
winners: dict[BindingKey, DocumentFields] = {}
for document in sorted(candidates, key=_latest_first, reverse=True):
winners.setdefault(_key(document), document)
bound = await _bound_documents(session, version.id, entity_ids)
bound = await for_consolidation(session, await _bound_documents(session, version.id, entity_ids), consolidation)
bound_ids = {document.id for document in bound}
replaced = [doc.id for doc in bound if _key(doc) in winners and winners[_key(doc)].id != doc.id]
await _unbind(session, version.id, replaced)
@ -261,9 +313,10 @@ async def on_document_confirmed( # noqa: PLR0913 - автор и коммент
consolidation = await session.get(Consolidation, version.consolidation_id)
if consolidation is None: # FK гарантирует наличие
continue
if not fits(document, version, consolidation, await _member_ids(session, version.id)):
[view] = await for_consolidation(session, [document], consolidation)
if not fits(view, version, consolidation, await _member_ids(session, version.id)):
continue
await _bind(session, version.id, document)
await _bind(session, version.id, consolidation, view)
if version.status in MOVES_TO_CORRECTION:
if changed_by is None:
logger.info(
@ -353,7 +406,8 @@ async def on_documents_mapped(
if consolidation is None: # FK гарантирует наличие
continue
members = await _member_ids(session, version.id)
if not any(fits(document, version, consolidation, members) for document in needs_check):
views = await for_consolidation(session, needs_check, consolidation)
if not any(fits(document, version, consolidation, members) for document in views):
continue
await transition(
session,

View File

@ -223,18 +223,31 @@ class ConsolidationService:
for rule, account in await repo.rules_for(self._session, version.approach)
)
entity_ids = [entity.id for entity in entities]
candidates = await repo.unconfirmed_documents(self._session, entity_ids, consolidation.year)
candidates = await consolidation_documents.for_consolidation(
self._session,
await repo.unconfirmed_documents(
self._session, entity_ids, consolidation.year, self._period_from_version(consolidation)
),
consolidation,
)
unbound = [doc for doc in candidates if consolidation_documents.fits(doc, version, consolidation, entity_ids)]
rows = checklist_domain.build(
rules,
version_quarter=consolidation.quarter,
members=entities,
bound=await repo.bound_documents(self._session, version.id),
bound=await consolidation_documents.for_consolidation(
self._session, await repo.bound_documents(self._session, version.id), consolidation
),
unbound=unbound,
files_state=await repo.entity_files_state(self._session, entity_ids),
)
return rules, rows
@staticmethod
def _period_from_version(consolidation: Consolidation) -> uuid.UUID | None:
"""`consolidation_id` для выборок кандидатов в режиме «период из версии» (arm-core#103); иначе `None`."""
return consolidation.id if consolidation_documents.period_from_version() else None
@staticmethod
def _cell_out(rule: checklist_domain.Rule, cell: checklist_domain.Cell) -> ChecklistCellOut:
"""Ячейка ответа с правилом (D-2); `rule` — правило того же столбца, что и `cell`."""
@ -397,6 +410,7 @@ class ConsolidationService:
[entity.id for _, entity in members],
consolidation.year,
quarters_through(consolidation.quarter),
self._period_from_version(consolidation),
)
return VersionCard(
**brief.model_dump(),

View File

@ -255,16 +255,7 @@ class LegalEntityService:
if entity.company_group_id is not None:
names = await repo.company_group_names_by_ids(self._session, [entity.company_group_id])
detail.company_group_name = names.get(entity.company_group_id)
detail.memberships = [
MembershipOut(
related_group_id=group.id,
code=group.code,
kind=group.kind,
name=group.name,
valid_from=member.valid_from,
)
for member, group in await repo.current_memberships(self._session, entity.id)
]
detail.memberships = (await self.memberships_by_entities([entity.id])).get(entity.id, [])
detail.aliases = [AliasOut.model_validate(alias) for alias in await repo.list_aliases(self._session, entity.id)]
return detail
@ -287,15 +278,34 @@ class LegalEntityService:
offices = await repo.regional_offices_by_ids(self._session, office_ids)
return {office_id: RegionalOfficeBrief.model_validate(office) for office_id, office in offices.items()}
async def memberships_by_entities(self, entity_ids: Iterable[uuid.UUID]) -> dict[uuid.UUID, list[MembershipOut]]:
"""Текущие членства ГСК/ГСЗ по набору ЮЛ — один запрос на набор, без N+1 (#109)."""
rows = await repo.current_memberships_by_entities(self._session, entity_ids)
return {
entity_id: [
MembershipOut(
related_group_id=group.id,
code=group.code,
kind=group.kind,
name=group.name,
valid_from=member.valid_from,
)
for member, group in pairs
]
for entity_id, pairs in rows.items()
}
async def to_outs(self, entities: Sequence[LegalEntity], counts: Mapping[uuid.UUID, int]) -> list[LegalEntityOut]:
"""Ответы по набору ЮЛ с числом документов; филиалы и имена ГК — по одному батч-запросу."""
"""Ответы по набору ЮЛ с числом документов; филиалы, имена ГК и членства ГСК/ГСЗ — по одному батч-запросу."""
offices = await self.offices_by_ids(entity.regional_office_id for entity in entities)
group_names = await repo.company_group_names_by_ids(self._session, (e.company_group_id for e in entities))
memberships = await self.memberships_by_entities(entity.id for entity in entities)
result: list[LegalEntityOut] = []
for entity in entities:
out = LegalEntityOut.model_validate(entity)
out.documents_count = counts.get(entity.id, 0)
out.regional_office = offices.get(entity.regional_office_id)
out.memberships = memberships.get(entity.id, [])
if entity.company_group_id is not None:
out.company_group_name = group_names.get(entity.company_group_id)
result.append(out)

View File

@ -480,7 +480,7 @@ class PrimaryFilesService:
id=job_id,
file_id=file_id,
pipeline=detected.pipeline,
request_subject=request_subject(detected.pipeline.value),
request_subject=request_subject(detected.pipeline.value, self._settings.OPENBAO.NATS.request_subjects()),
schema_version=CONTRACT_VERSION,
status=RecognitionStatus.queued,
attempt=1,

View File

@ -26,10 +26,10 @@ SKIP LOCKED` → публикация → `UPDATE` → `COMMIT`). Блокиро
Обмен не пишет `audit_event` (рулинг R-7, design #20; ASSUMPTION(OQ-55)).
Inbox (задача 24, рулинги R-12, R-22): `handle_message()` — обработчик подписки `files.tabular_parsed`
(`src/core/nats.py`, queue group `arm-core`). Одно сообщение — одна короткая транзакция без сетевого IO:
`SELECT … FOR UPDATE` строки задания (ЖДЁТ блокировку: ack, пришедший, пока публикатор держит строку до своего commit,
применяется после него, а не теряется) → проверки статуса → изменения → `COMMIT`.
Inbox (задача 24, рулинги R-12, R-22): `handle_message()` — обработчик подписки на subject ответов (по
умолчанию `files.tabular_parsed`; `src/core/nats.py`, queue group `arm-core`). Одно сообщение — одна короткая
транзакция без сетевого IO: `SELECT … FOR UPDATE` строки задания (ЖДЁТ блокировку: ack, пришедший, пока публикатор
держит строку до своего commit, применяется после него, а не теряется) → проверки статуса → изменения → `COMMIT`.
- ack (`status = processing`): `published` → `processing`, `started_at`, `service_*`, файл `UPLOADED → PROCESSING`
(R-5: только из `UPLOADED`); повторный ack для `processing` — тихо игнорируется; ack для другого статуса — warning,
задание и файл не меняются.
@ -64,6 +64,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from src.contract import (
CONTRACT_VERSION,
NATS_MSG_ID_HEADER,
REQUEST_SUBJECTS,
MessageStatus,
Metrics,
RecognitionRequest,
@ -211,6 +212,7 @@ async def _process_one( # noqa: PLR0913 - один шаг прохода: вс
ack_cutoff: datetime,
pass_started: datetime,
max_republish: int,
request_subjects: Mapping[str, str],
) -> bool:
"""Обработать одно задание в транзакции `session`; `False` — подходящих заданий больше нет."""
row = (await session.execute(_candidate_stmt(pass_started, ack_cutoff))).one_or_none()
@ -244,7 +246,7 @@ async def _process_one( # noqa: PLR0913 - один шаг прохода: вс
reply_deadline = row.reply_deadline or now + DEFAULT_REPLY_TIMEOUT[row.pipeline]
try:
request = _build_request(row, reply_deadline)
subject = request_subject(row.pipeline.value)
subject = request_subject(row.pipeline.value, request_subjects)
except (ValidationError, ValueError) as exc:
await session.execute(
_job_update(row.id, row.status).values(
@ -297,7 +299,7 @@ async def _process_one( # noqa: PLR0913 - один шаг прохода: вс
metrics = base.op("||", return_type=JSONB)(literal({REPUBLISH_COUNT_KEY: count}, JSONB))
await session.execute(
_job_update(row.id, RecognitionStatus.published).values(
published_at=now, reply_deadline=reply_deadline, metrics=metrics
published_at=now, reply_deadline=reply_deadline, metrics=metrics, request_subject=subject
)
)
report.republished += 1
@ -320,6 +322,7 @@ async def publish_pending( # noqa: PLR0913 - параметры прохода
max_republish: int,
batch_size: int = PUBLISH_BATCH_SIZE,
now: Clock | None = None,
request_subjects: Mapping[str, str] = REQUEST_SUBJECTS,
) -> PublishReport:
"""Один проход outbox-публикатора (spec «Outbox публикует задания из `queued`», «Запрос без ack публикуется
повторно, затем — `dlq`»; design D-6).
@ -346,6 +349,7 @@ async def publish_pending( # noqa: PLR0913 - параметры прохода
ack_cutoff=ack_cutoff,
pass_started=pass_started,
max_republish=max_republish,
request_subjects=request_subjects,
)
except NatsPublishError as exc:
report.failed += 1
@ -402,11 +406,18 @@ async def run_publisher_loop( # noqa: PLR0913 - настройки цикла
ack_timeout: timedelta,
max_republish: int,
sleep: SleepFn = asyncio.sleep,
request_subjects: Mapping[str, str] = REQUEST_SUBJECTS,
) -> None:
"""Бесконечный цикл публикатора с периодом `poll_seconds`; ошибка прохода логируется, цикл продолжается."""
while True:
try:
await publish_pending(session_factory, nats, ack_timeout=ack_timeout, max_republish=max_republish)
await publish_pending(
session_factory,
nats,
ack_timeout=ack_timeout,
max_republish=max_republish,
request_subjects=request_subjects,
)
except Exception as exc:
logger.exception(
"Ошибка прохода outbox-публикатора. error_category=outbox_error, error_type={error_type}, "
@ -459,6 +470,7 @@ class ExchangeTasks:
poll_seconds=settings.OUTBOX_POLL_SECONDS,
ack_timeout=timedelta(seconds=settings.ACK_TIMEOUT_SECONDS),
max_republish=settings.OUTBOX_MAX_REPUBLISH,
request_subjects=settings.OPENBAO.NATS.request_subjects(),
),
name="recognition-outbox-publisher",
)
@ -496,7 +508,7 @@ class ExchangeTasks:
)
# --- inbox: files.tabular_parsed ----------------------------------------------------------------------------------
# --- inbox: subject ответов ----------------------------------------------------------------------------------
# Финальные статусы для inbox (рулинг R-22, design #34): результат для них не применяется
FINAL_STATUSES: frozenset[RecognitionStatus] = frozenset(
@ -740,7 +752,7 @@ async def _apply_message(
job = await _lock_job(session, message.job_id)
if job is None:
logger.warning(
"Сообщение files.tabular_parsed для неизвестного задания, не применено. error_category=unknown_job, "
"Сообщение на subject ответа для неизвестного задания, не применено. error_category=unknown_job, "
"job_id={job_id}, status={status}",
job_id=message.job_id,
status=message.status.value,
@ -748,7 +760,7 @@ async def _apply_message(
return
if job.file_id != message.file_id:
logger.warning(
"Сообщение files.tabular_parsed: file_id не совпадает с заданием, не применено. "
"Сообщение на subject ответа: file_id не совпадает с заданием, не применено. "
"error_category=file_mismatch, job_id={job_id}, file_id={file_id}, message_file_id={message_file_id}",
job_id=job.id,
file_id=job.file_id,
@ -760,7 +772,7 @@ async def _apply_message(
if msg_id is None:
msg_id = ack_msg_id(job.id, job.attempt) if is_ack else result_msg_id(job.id, job.attempt)
logger.warning(
"Сообщение files.tabular_parsed без заголовка Nats-Msg-Id: использован производный id. "
"Сообщение на subject ответа без заголовка Nats-Msg-Id: использован производный id. "
"error_category=missing_msg_id, job_id={job_id}, msg_id={msg_id}",
job_id=job.id,
msg_id=msg_id,
@ -778,8 +790,9 @@ async def handle_message(
session_factory: async_sessionmaker[AsyncSession],
now: Clock | None = None,
) -> None:
"""Обработчик `files.tabular_parsed` (spec «Ack сервиса переводит задание в `processing`», «Inbox принимает
результат идемпотентно»; design D-6). Модель транзакции и правила — в docstring модуля.
"""Обработчик subject ответов (по умолчанию `files.tabular_parsed`; spec «Ack сервиса переводит задание в
`processing`», «Inbox принимает результат идемпотентно»; design D-6). Модель транзакции и правила — в docstring
модуля.
Исключения наружу не выходят: невалидное сообщение — warning; ошибка БД — откат транзакции (задание и файл),
`logger.exception`, без повторов (задание дойдёт до `timeout`). `audit_event` не пишется (R-7,

View File

@ -22,7 +22,7 @@ from sqlalchemy import func, select, update
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from src.api.dependencies.directories import get_app_settings
from src.core.config import get_settings
from src.core.config import NATSSettings, get_settings
from src.models.access import AppUser
from src.models.audit_ex import AuditEvent
from src.models.consolidation import Consolidation, ConsolidationStatusHistory, ConsolidationVersion
@ -210,6 +210,32 @@ async def test_real_osv_is_accepted_and_published_from_outbox(
# --- поля записей, имя файла, аудит ------------------------------------------------------------------------------
@pytest.mark.integration
async def test_job_request_subject_comes_from_environment_settings(
app: FastAPI, auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
) -> None:
"""Review Focus 4: `recognition_job.request_subject` берётся из настроек окружения, а не из констант контракта."""
base = get_settings()
nats = NATSSettings(
**{
**base.OPENBAO.NATS.model_dump(),
"NATS_TABULAR_REQUEST_SUBJECT": "test.files.tabular_uploaded",
}
)
app.dependency_overrides[get_app_settings] = lambda: base.model_copy(
update={"OPENBAO": base.OPENBAO.model_copy(update={"NATS": nats})}
)
entity = await _rostov_entity(seeded)
client = await auth_client(KA_RF)
response = await _post(client, entity.id, ("osv.xls", _xls("subject")))
assert response.status_code == HTTP_CREATED, response.text
job = await seeded.get(RecognitionJob, uuid.UUID(response.json()["accepted"][0]["job_id"]))
assert job is not None
assert job.request_subject == "test.files.tabular_uploaded"
@pytest.mark.integration
async def test_hostile_name_gives_flat_key_and_full_records(
auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage

View File

@ -0,0 +1,125 @@
"""Текущие членства ГСК/ГСЗ в ответах ЮЛ (arm-core#109): `memberships` в списке, `move`, `withdraw`, без N+1."""
import datetime
from collections.abc import Iterator
from typing import Any
import pytest
from sqlalchemy import event
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
from src.common.clock import moscow_date
from src.models.enums import RelatedGroupKind
from tests.conftest import AuthClientFactory
from tests.factories import make_company_group, make_legal_entity, make_membership, make_related_group
HTTP_OK = 200
SMALL = 2
ENTITIES = "/directories/legal-entities"
KA_GO = "ka_go@armka.local"
@pytest.fixture
def sql_log(db_engine: AsyncEngine) -> Iterator[list[str]]:
statements: list[str] = []
def record(_conn: Any, _cursor: Any, statement: str, *_rest: Any) -> None:
statements.append(statement)
event.listen(db_engine.sync_engine, "before_cursor_execute", record)
try:
yield statements
finally:
event.remove(db_engine.sync_engine, "before_cursor_execute", record)
def _member_selects(statements: list[str]) -> int:
return sum(
1 for s in statements if s.lstrip().upper().startswith("SELECT") and "FROM dkb.related_group_member " in s
)
def _today() -> str:
return moscow_date(datetime.datetime.now(datetime.UTC)).isoformat()
@pytest.mark.integration
async def test_list_carries_current_memberships(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
group = await make_company_group(seeded, name="ГК Членства")
both = await make_legal_entity(seeded, name="ООО Членства обе", group=group)
only_gsz = await make_legal_entity(seeded, name="ООО Членства ГСЗ", group=group)
none = await make_legal_entity(seeded, name="ООО Членства нет", group=group)
gsk = await make_related_group(seeded, group=group, name="ГСК Членства")
gsz = await make_related_group(seeded, group=group, kind=RelatedGroupKind.gsz, name="ГСЗ Членства")
old_gsk = await make_related_group(seeded, group=group, name="ГСК Прежняя")
await make_membership(seeded, related_group=gsk, entity=both)
await make_membership(seeded, related_group=gsz, entity=both)
await make_membership(seeded, related_group=gsz, entity=only_gsz)
await make_membership(seeded, related_group=old_gsk, entity=none, is_current=False)
client = await auth_client(KA_GO)
response = await client.get(ENTITIES, params={"company_group_id": str(group.id)})
assert response.status_code == HTTP_OK, response.text
by_id = {item["id"]: item for item in response.json()["items"]}
def groups(entity_id: Any) -> list[tuple[str, str, str, str]]:
return [(m["related_group_id"], m["code"], m["kind"], m["name"]) for m in by_id[str(entity_id)]["memberships"]]
assert groups(both.id) == [
(str(gsk.id), gsk.code, "gsk", "ГСК Членства"),
(str(gsz.id), gsz.code, "gsz", "ГСЗ Членства"),
]
assert groups(only_gsz.id) == [(str(gsz.id), gsz.code, "gsz", "ГСЗ Членства")]
assert groups(none.id) == [] # закрытое членство не отдаётся
assert all(m["valid_from"] for m in by_id[str(both.id)]["memberships"])
@pytest.mark.integration
async def test_move_and_withdraw_responses_have_no_memberships(
auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
group = await make_company_group(seeded, name="ГК Откуда членства")
new_group = await make_company_group(seeded, name="ГК Куда членства")
moved = await make_legal_entity(seeded, name="ООО Переезд членства", group=group)
withdrawn = await make_legal_entity(seeded, name="ООО Вывод членства", group=group)
gsk = await make_related_group(seeded, group=group, name="ГСК Откуда")
await make_membership(seeded, related_group=gsk, entity=moved)
await make_membership(seeded, related_group=gsk, entity=withdrawn)
client = await auth_client(KA_GO)
move = await client.post(
f"{ENTITIES}/{moved.id}/move", json={"company_group_id": str(new_group.id), "effective_date": _today()}
)
assert move.status_code == HTTP_OK, move.text
assert move.json()["memberships"] == []
withdraw = await client.post(
f"{ENTITIES}/{withdrawn.id}/withdraw", json={"event": "REMOVE_FROM_GROUP", "effective_date": _today()}
)
assert withdraw.status_code == HTTP_OK, withdraw.text
assert withdraw.json()["memberships"] == []
@pytest.mark.integration
async def test_list_loads_memberships_in_one_query(
auth_client: AuthClientFactory, seeded: AsyncSession, sql_log: list[str]
) -> None:
client = await auth_client(KA_GO)
await client.get(ENTITIES) # прогрев (первый запрос сессии тянет лишнее)
total_before = 0
for size in (2, 12):
group = await make_company_group(seeded, name=f"ГК членства список {size}")
gsk = await make_related_group(seeded, group=group, name=f"ГСК список {size}")
for n in range(size):
entity = await make_legal_entity(seeded, name=f"ООО членства список {size}-{n}", group=group)
await make_membership(seeded, related_group=gsk, entity=entity)
sql_log.clear()
response = await client.get(ENTITIES, params={"q": "членства список", "limit": 200})
assert response.status_code == HTTP_OK, response.text
items = response.json()["items"]
assert all(len(item["memberships"]) == 1 for item in items)
if size == SMALL:
total_before = len(sql_log)
else:
assert len(items) == 14 # noqa: PLR2004 - 2 + 12
assert len(sql_log) == total_before
assert _member_selects(sql_log) == 1

View File

@ -0,0 +1,85 @@
"""`crm_id` ГСК/ГСЗ в API без колонки в БД (arm-core #111; колонка — #112).
Формы 51-03 / 51-04 / 77-02 шлют «ID ГСК/ГСЗ в CRM»: POST/PATCH его принимают, значение отбрасывается,
в ответах `crm_id` всегда `null`, в аудит не попадает.
"""
from collections.abc import Sequence
from typing import Any
import httpx
import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from src.models.audit_ex import AuditEvent
from src.models.enums import AuditAction
from tests.conftest import AuthClientFactory
from tests.factories import make_company_group
HTTP_OK = 200
HTTP_CREATED = 201
HTTP_UNPROCESSABLE = 422
GROUPS = "/directories/company-groups"
RELATED = "/directories/related-groups"
GO = "ka_go@armka.local"
async def _events(session: AsyncSession) -> Sequence[AuditEvent]:
stmt = select(AuditEvent).where(
AuditEvent.action == AuditAction.MEMBERSHIP_CHANGE, AuditEvent.object_type == "related_group"
)
return (await session.execute(stmt.order_by(AuditEvent.occurred_at, AuditEvent.id))).scalars().all()
async def _create(client: httpx.AsyncClient, group_id: Any, **extra: Any) -> httpx.Response:
return await client.post(f"{GROUPS}/{group_id}/related-groups", json={"kind": "gsk", "name": "ГСК А", **extra})
@pytest.mark.integration
async def test_create_accepts_crm_id_and_drops_it(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
group = await make_company_group(seeded)
client = await auth_client(GO)
response = await _create(client, group.id, crm_id=" 3-9CGSK01 ")
assert response.status_code == HTTP_CREATED, response.text
body = response.json()
assert body["crm_id"] is None
assert body["name"] == "ГСК А"
events = await _events(seeded)
assert len(events) == 1
assert "crm_id" not in events[0].after_value
for value in ("", None):
assert (await _create(client, group.id, crm_id=value)).status_code == HTTP_CREATED
assert (await _create(client, group.id, crm_id=123)).status_code == HTTP_UNPROCESSABLE
@pytest.mark.integration
async def test_patch_accepts_crm_id_without_changes(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
group = await make_company_group(seeded)
client = await auth_client(GO)
gsk = (await _create(client, group.id)).json()
only_crm = await client.patch(f"{RELATED}/{gsk['id']}", json={"crm_id": "3-9CGSK01"})
assert only_crm.status_code == HTTP_OK, only_crm.text
assert only_crm.json()["crm_id"] is None
assert len(await _events(seeded)) == 1 # только создание: crm_id не изменение
with_name = await client.patch(f"{RELATED}/{gsk['id']}", json={"crm_id": None, "name": "ГСК Б"})
assert with_name.status_code == HTTP_OK, with_name.text
assert (with_name.json()["name"], with_name.json()["crm_id"]) == ("ГСК Б", None)
events = await _events(seeded)
assert len(events) == 2 # noqa: PLR2004 - создание и смена имени
assert events[1].after_value == {"name": "ГСК Б"}
@pytest.mark.integration
async def test_list_and_detail_return_null_crm_id(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
group = await make_company_group(seeded)
client = await auth_client(GO)
gsk = (await _create(client, group.id, crm_id="3-9CGSK01")).json()
listed = await client.get(f"{GROUPS}/{group.id}/related-groups")
assert [row["crm_id"] for row in listed.json()] == [None]
assert (await client.get(f"{RELATED}/{gsk['id']}")).json()["crm_id"] is None

View File

@ -83,6 +83,7 @@ async def test_create_gsk_and_gsz_codes_and_audit(auth_client: AuthClientFactory
"kind": "gsk",
"name": "ГСК Бисквит",
"created_date": "2025-01-15",
"crm_id": None,
"members_count": 0,
}
assert gsz_body["created_date"] is None

View File

@ -18,6 +18,7 @@ import pytest
from fastapi import Request
from fastapi.routing import APIRoute
from loguru import logger
from pydantic import SecretStr
from src.api.router import router
from src.core.config import Settings
@ -47,6 +48,49 @@ async def test_healthcheck_nats_off_when_disabled(client: httpx.AsyncClient, mon
assert {"uptime", "version"} <= set(body)
def _expected_subjects() -> dict[str, str]:
nats = app_module.settings.OPENBAO.NATS
return {
"tabular_request": nats.NATS_TABULAR_REQUEST_SUBJECT,
"image_request": nats.NATS_IMAGE_REQUEST_SUBJECT,
"result": nats.NATS_RESULT_SUBJECT,
}
@pytest.mark.parametrize("enabled", [False, True], ids=["when_nats_off", "when_nats_down"])
async def test_healthcheck_subjects(client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch, enabled: bool):
"""Review Focus 5: subjects отдаются при любом состоянии NATS, JWT в ответе нет."""
monkeypatch.setattr(app_module.settings, "NATS_ENABLED", enabled)
body = (await client.get("/healthcheck")).json()
assert body["subjects"] == _expected_subjects()
assert "NATS_JWT" not in str(body)
assert "jwt" not in str(body).lower()
@pytest.mark.parametrize("enabled", [False, True], ids=["when_nats_off", "when_nats_down"])
async def test_healthcheck_returns_non_default_subjects_without_jwt(
client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch, enabled: bool
):
"""T3.1: отдаются именно настроенные (не умолчательные) subject'ы; непустой JWT в тело не попадает."""
jwt = "JWTVALUEMARK"
nats = app_module.settings.OPENBAO.NATS
monkeypatch.setattr(app_module.settings, "NATS_ENABLED", enabled)
monkeypatch.setattr(nats, "NATS_TABULAR_REQUEST_SUBJECT", "env.files.tabular_uploaded")
monkeypatch.setattr(nats, "NATS_IMAGE_REQUEST_SUBJECT", "env.files.img_uploaded")
monkeypatch.setattr(nats, "NATS_RESULT_SUBJECT", "env.files.tabular_parsed")
monkeypatch.setattr(nats, "NATS_JWT", SecretStr(jwt))
assert nats.NATS_JWT.get_secret_value() == jwt
response = await client.get("/healthcheck")
assert response.json()["subjects"] == {
"tabular_request": "env.files.tabular_uploaded",
"image_request": "env.files.img_uploaded",
"result": "env.files.tabular_parsed",
}
assert jwt not in response.text
async def test_healthcheck_nats_down_when_enabled_without_connection(
client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch
):
@ -220,6 +264,26 @@ async def test_lifespan_does_not_wait_for_nats_and_healthcheck_is_down(monkeypat
assert not nats.is_connected
@pytest.mark.usefixtures("_no_smb")
async def test_lifespan_logs_nats_subjects_at_start(monkeypatch: pytest.MonkeyPatch):
"""Tasks 1.2: при старте в лог пишутся используемые subject'ы NATS."""
records: list[Any] = []
handler_id = logger.add(lambda message: records.append(message.record), level="INFO")
monkeypatch.setattr(app_module, "_connect_s3", _unavailable_s3)
monkeypatch.setattr(app_module, "_create_nats", lambda: None)
monkeypatch.setattr(app_module.settings.OPENBAO.NATS, "NATS_RESULT_SUBJECT", "env.files.tabular_parsed")
try:
async with app_module.lifespan(app_module.app):
pass
finally:
logger.remove(handler_id)
messages = [r["message"] for r in records if "Subject'ы NATS" in r["message"]]
assert len(messages) == 1
assert "env.files.tabular_parsed" in messages[0]
assert "files.tabular_uploaded" in messages[0]
def test_create_nats_disabled_returns_none(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(app_module.settings, "NATS_ENABLED", False)
assert app_module._create_nats() is None

View File

@ -36,15 +36,7 @@ def test_env_test_is_read(clean_env: pytest.MonkeyPatch) -> None:
def test_defaults(clean_env: pytest.MonkeyPatch) -> None:
"""Значения по умолчанию классов настроек: env-файла нет, заданы только обязательные OPENBAO__* (SMB не нужен)."""
required = {
"OPENBAO__DATABASE__DATABASE_URL": "postgresql+asyncpg://u:p@localhost:5432/d",
"OPENBAO__MINIO__MINIO_URL": "localhost:9000",
"OPENBAO__MINIO__MINIO_ADMIN_LOGIN": "login",
"OPENBAO__MINIO__MINIO_ADMIN_PASSWORD": "password",
"OPENBAO__MINIO__MINIO_BUCKET": "bucket",
"OPENBAO__NATS__NATS_URL": "nats://localhost:4222",
}
for key, value in required.items():
for key, value in REQUIRED_ENV.items():
clean_env.setenv(key, value)
s = Settings(_env_file=None) # type: ignore[call-arg]
assert s.DEBUG is False
@ -58,8 +50,8 @@ def test_defaults(clean_env: pytest.MonkeyPatch) -> None:
assert s.NATS_ENABLED is False
assert (s.OUTBOX_POLL_SECONDS, s.ACK_TIMEOUT_SECONDS, s.OUTBOX_MAX_REPUBLISH) == (5, 30, 3)
assert (s.UPLOAD_MAX_BYTES, s.MAPPING_POLL_SECONDS, s.MAPPING_MAX_ATTEMPTS) == (50 * 1024 * 1024, 5, 3)
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 s.OPENBAO.MINIO.MINIO_SECURE is True
assert (s.server.HOST, s.server.PORT) == ("0.0.0.0", 8000) # noqa: S104 - по умолчанию, как в контейнере
smb = s.OPENBAO.SMB
@ -69,8 +61,8 @@ def test_defaults(clean_env: pytest.MonkeyPatch) -> None:
def test_empty_nats_seed_and_jwt_allowed(clean_env: pytest.MonkeyPatch) -> None:
"""Пустые NATS_SEED/NATS_JWT допустимы (подключение без nkeys)."""
s = Settings(_env_file=API_DIR / ".env.example") # type: ignore[call-arg]
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 s.NATS_ENABLED is True
assert s.OPENBAO.SMB.SMB_HOST == ""
assert (s.server.HOST, s.server.PORT) == ("127.0.0.1", 8000) # dev-запуск — только loopback (M-11)
@ -155,3 +147,110 @@ def test_cors_origins_from_env(clean_env: pytest.MonkeyPatch, raw: str, expected
clean_env.setenv("CORS_ORIGINS", raw)
s = Settings(_env_file=API_DIR / ".env.test") # type: ignore[call-arg]
assert s.CORS_ORIGINS == expected
@pytest.fixture
def settings_from_env(clean_env: pytest.MonkeyPatch):
def make(**env: str | None) -> Settings:
for key, value in {**REQUIRED_ENV, **env}.items():
if value is None:
clean_env.delenv(key, raising=False)
else:
clean_env.setenv(key, value)
return Settings(_env_file=None) # type: ignore[call-arg]
return make
JWT = "eyJ0eXAiOiJKV1QiLCJhbGciOiJlZDI1NTE5LW5rZXkifQ.secret-payload.sig"
SEED = "SUACSSL3UAHUDXKFSNVUZRF5UHPMWZ6BFDTJ7M6USDXIEDNPPQYYYCU3VY"
def test_nats_subjects_default_to_contract_names(settings_from_env) -> None:
nats = settings_from_env().OPENBAO.NATS
assert nats.NATS_TABULAR_REQUEST_SUBJECT == "files.tabular_uploaded"
assert nats.NATS_IMAGE_REQUEST_SUBJECT == "files.img_uploaded"
assert nats.NATS_RESULT_SUBJECT == "files.tabular_parsed"
assert nats.request_subjects() == {"tabular": "files.tabular_uploaded", "image": "files.img_uploaded"}
def test_nats_subjects_from_openbao(settings_from_env) -> None:
nats = settings_from_env(
OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT="test.files.tabular_uploaded",
OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT="test.files.img_uploaded",
OPENBAO__NATS__NATS_RESULT_SUBJECT="test.files.tabular_parsed",
).OPENBAO.NATS
assert nats.request_subjects() == {"tabular": "test.files.tabular_uploaded", "image": "test.files.img_uploaded"}
assert nats.NATS_RESULT_SUBJECT == "test.files.tabular_parsed"
SUBJECT_KEYS = ("NATS_TABULAR_REQUEST_SUBJECT", "NATS_IMAGE_REQUEST_SUBJECT", "NATS_RESULT_SUBJECT")
@pytest.mark.parametrize("key", SUBJECT_KEYS)
@pytest.mark.parametrize(
"value",
["", "files.*", "files.>", ">", "files..tabular", ".files", "files.", "files tabular", "files.tabular_uploaded "],
)
def test_invalid_subject_rejected(settings_from_env, key: str, value: str) -> None:
with pytest.raises(ValidationError, match=key):
settings_from_env(**{f"OPENBAO__NATS__{key}": value})
@pytest.mark.parametrize(
("first", "second"),
[
("NATS_TABULAR_REQUEST_SUBJECT", "NATS_IMAGE_REQUEST_SUBJECT"),
("NATS_TABULAR_REQUEST_SUBJECT", "NATS_RESULT_SUBJECT"),
("NATS_IMAGE_REQUEST_SUBJECT", "NATS_RESULT_SUBJECT"),
],
)
def test_equal_subjects_rejected(settings_from_env, first: str, second: str) -> None:
with pytest.raises(ValidationError, match=f"{first} и {second} совпадают"):
settings_from_env(**{f"OPENBAO__NATS__{first}": "files.same", f"OPENBAO__NATS__{second}": "files.same"})
def test_invalid_subject_error_hides_nats_secrets(settings_from_env) -> None:
"""Ошибка валидации subject'а при заданных JWT/seed не печатает их (маркеры короткие, см. тест ниже)."""
jwt_mark, seed_mark = "JWTMARK", "SEEDMARK"
with pytest.raises(ValidationError) as exc_info:
settings_from_env(
OPENBAO__NATS__NATS_JWT=jwt_mark,
OPENBAO__NATS__NATS_SEED=seed_mark,
OPENBAO__NATS__NATS_RESULT_SUBJECT="files.*",
)
assert "NATS_RESULT_SUBJECT" in str(exc_info.value)
assert jwt_mark not in str(exc_info.value)
assert seed_mark not in str(exc_info.value)
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 обрезает `input_value` в тексте ошибки (около 50 символов по середине), и полный
JWT там не виден даже без скрытия — проверка на полном значении прошла бы вхолостую.
"""
jwt_mark, seed_mark = "JWTMARK", "SEEDMARK"
with pytest.raises(ValidationError) as exc_info:
settings_from_env(
OPENBAO__NATS__NATS_JWT=jwt_mark, OPENBAO__NATS__NATS_SEED=seed_mark, **dict.fromkeys(missing)
)
assert jwt_mark not in str(exc_info.value)
assert seed_mark not in str(exc_info.value)

View File

@ -626,6 +626,12 @@ def test_request_subject_by_pipeline() -> None:
assert NATS_MSG_ID_HEADER == "Nats-Msg-Id"
def test_request_subject_uses_environment_map() -> None:
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"
def test_request_subject_rejects_unknown_pipeline() -> None:
with pytest.raises(ValueError, match="pipeline"):
request_subject("ocr")

View File

@ -0,0 +1,89 @@
"""Временный режим «период ОСВ из версии консолидации» (arm-core#103): чистые функции `domain.version_period`."""
import datetime
import uuid
import pytest
from src.domain.version_period import takes_version_period, with_version_period
from src.models.consolidation import Consolidation, ConsolidationVersion
from src.models.enums import ConsolidationStatus, DocumentStatus, DocumentType, PeriodQuarter
from src.models.files import Document
from src.services.consolidation_documents import fits
CONSOLIDATION = uuid.uuid4()
OTHER = uuid.uuid4()
MEMBER = uuid.uuid4()
@pytest.mark.parametrize(
("enabled", "document_type", "file_consolidation_id", "fits_by_header", "expected"),
[
(True, DocumentType.OSV, CONSOLIDATION, False, True), # ОСВ «не того» периода загружена в эту консолидацию
(True, DocumentType.OSV, CONSOLIDATION, True, False), # период шапки и так подходит — склейка кварталов цела
(False, DocumentType.OSV, CONSOLIDATION, False, False), # режим выключен — период из шапки
(True, DocumentType.OSV, OTHER, False, False), # загружена в другую консолидацию
(True, DocumentType.OSV, None, False, False), # загружена с рабочего стола, без консолидации
(True, DocumentType.BALANCE, CONSOLIDATION, False, False), # ББ/ОФР — только период из шапки
(True, DocumentType.INCOME, CONSOLIDATION, False, False),
],
)
def test_takes_version_period(
enabled: bool,
document_type: DocumentType,
file_consolidation_id: uuid.UUID | None,
fits_by_header: bool,
expected: bool,
) -> None:
assert (
takes_version_period(
enabled=enabled,
document_type=document_type,
file_consolidation_id=file_consolidation_id,
consolidation_id=CONSOLIDATION,
fits_by_header=fits_by_header,
)
is expected
)
def _document(*, year: int | None = 2025, quarter: PeriodQuarter | None = PeriodQuarter.Q1) -> Document:
return Document(
id=uuid.uuid4(),
legal_entity_id=MEMBER,
document_type=DocumentType.OSV,
osv_account_code="60",
period_year=year,
period_quarter=quarter,
is_cumulative=False,
status=DocumentStatus.CONFIRMED,
confirmed_at=datetime.datetime(2026, 10, 12, tzinfo=datetime.UTC),
)
def test_with_version_period_takes_year_and_quarter_of_version_as_cumulative() -> None:
document = _document()
view = with_version_period(document, year=2026, quarter=PeriodQuarter.Q3)
assert (view.period_year, view.period_quarter, view.is_cumulative) == (2026, PeriodQuarter.Q3, True)
assert (view.id, view.legal_entity_id, view.document_type, view.osv_account_code) == (
document.id,
MEMBER,
DocumentType.OSV,
"60",
)
assert (view.status, view.confirmed_at) == (document.status, document.confirmed_at)
# первичка не мутируется: документ сохраняет период из шапки
assert (document.period_year, document.period_quarter, document.is_cumulative) == (2025, PeriodQuarter.Q1, False)
@pytest.mark.parametrize("quarter", [None, PeriodQuarter.Q4])
def test_version_period_fits_version_whatever_header_period(quarter: PeriodQuarter | None) -> None:
"""Документ без квартала или за другой год/квартал в представлении версии подходит ей (если ЮЛ — член)."""
consolidation = Consolidation(year=2026, quarter=PeriodQuarter.Q2)
version = ConsolidationVersion(status=ConsolidationStatus.INITIATED)
view = with_version_period(_document(year=2024, quarter=quarter), year=2026, quarter=PeriodQuarter.Q2)
assert fits(view, version, consolidation, {MEMBER}) is True
assert fits(view, version, consolidation, {uuid.uuid4()}) is False

View File

@ -25,6 +25,7 @@ from src.core.nats import (
NatsNotConnectedError,
NatsPublishError,
build_nats_auth,
used_subjects,
)
SRC_DIR = Path(__file__).resolve().parent.parent / "src"
@ -377,3 +378,33 @@ def test_src_does_not_use_jetstream():
if alias.name.startswith("nats.js")
)
assert offenders == []
def test_used_subjects_core() -> None:
nats = NATSSettings(NATS_URL="nats://x:4222", NATS_RESULT_SUBJECT="test.files.tabular_parsed")
assert used_subjects(nats) == {
"tabular_request": "files.tabular_uploaded",
"image_request": "files.img_uploaded",
"result": "test.files.tabular_parsed",
}
def test_connection_subscribes_to_configured_result_subject() -> None:
nats = NATSSettings(NATS_URL="nats://x:4222", NATS_RESULT_SUBJECT="test.files.tabular_parsed")
assert NatsConnection(nats).result_subject == "test.files.tabular_parsed"
assert NatsConnection(nats, result_subject="override").result_subject == "override"
async def test_connection_subscribes_with_configured_result_subject() -> None:
"""T3.2: subscribe вызывается с subject'ом из настроек (без override) и группой очереди."""
nats = NATSSettings(NATS_URL="nats://x:4222", NATS_RESULT_SUBJECT="test.files.tabular_parsed")
world = _FakeWorld()
conn = NatsConnection(nats, client_factory=world, sleep=_RecordingSleep(), queue="g")
await conn.start()
assert await conn.wait_ready(timeout=2)
await conn.subscribe_results(_handler)
subscription = world.clients[-1].subscriptions[0]
assert (subscription["subject"], subscription["queue"]) == ("test.files.tabular_parsed", "g")
await conn.stop()

View File

@ -75,7 +75,8 @@ class Clock:
class RequestSubscriber:
"""Тестовый подписчик на subject'ы запросов: «голый» клиент nats-py, сообщения фильтруются по `job_id`."""
def __init__(self) -> None:
def __init__(self, subjects: tuple[str, ...] = REQUEST_SUBJECTS) -> None:
self.subjects = subjects
self.messages: list[Msg] = []
self._nc = NATS()
@ -89,7 +90,7 @@ class RequestSubscriber:
async def collect(msg: Msg) -> None:
self.messages.append(msg)
for subject in REQUEST_SUBJECTS:
for subject in self.subjects:
await self._nc.subscribe(subject, cb=collect)
await self._nc.flush()
@ -261,6 +262,67 @@ async def test_publish_request_to_service( # noqa: PLR0913,PLR0917 - фикст
assert job.attempt == 1
@pytest.mark.integration
async def test_publish_uses_configured_request_subject(
session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, nats: NatsConnection
) -> None:
"""Review Focus 4: subject — из настроек окружения, `request_subject` задания — фактический subject публикации."""
run = uuid.uuid4().hex[:8]
subjects = {"tabular": f"test-{run}.files.tabular_uploaded", "image": f"test-{run}.files.img_uploaded"}
custom = RequestSubscriber(subjects=tuple(subjects.values()))
await custom.start()
try:
job_id = await jobs.make_queued_job(
pipeline=RecognitionPipeline.tabular, file_type=FileType.XLSX, original_name="ОСВ 60.xlsx"
)
await publish_pending(
session_factory, nats, ack_timeout=ACK, max_republish=MAX_REPUBLISH, request_subjects=subjects
)
(msg,) = await custom.wait_for(job_id, 1)
finally:
await custom.stop()
assert msg.subject == subjects["tabular"]
assert (await jobs.get(job_id)).request_subject == subjects["tabular"]
@pytest.mark.integration
async def test_republish_updates_request_subject(
session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, nats: NatsConnection
) -> None:
"""I-1: повторная публикация уходит на актуальный subject и обновляет `request_subject` задания."""
run = uuid.uuid4().hex[:8]
subjects = {"tabular": f"test-{run}.files.tabular_uploaded", "image": f"test-{run}.files.img_uploaded"}
clock = Clock()
job_id = await jobs.make_queued_job(
pipeline=RecognitionPipeline.tabular,
file_type=FileType.XLSX,
original_name="ОСВ 61.xlsx",
reply_deadline=clock() + timedelta(hours=1),
)
assert (await _publish(session_factory, nats, clock)).published == 1
old_subject = (await jobs.get(job_id)).request_subject
assert old_subject != subjects["tabular"]
custom = RequestSubscriber(subjects=tuple(subjects.values()))
await custom.start()
try:
clock.advance(ACK + timedelta(seconds=1))
report = await publish_pending(
session_factory,
nats,
ack_timeout=ACK,
max_republish=MAX_REPUBLISH,
now=clock,
request_subjects=subjects,
)
assert report.republished == 1
(msg,) = await custom.wait_for(job_id, 1)
finally:
await custom.stop()
assert msg.subject == subjects["tabular"]
assert (await jobs.get(job_id)).request_subject == subjects["tabular"]
@pytest.mark.integration
async def test_publish_keeps_existing_reply_deadline(
session_factory: async_sessionmaker[AsyncSession],
@ -819,7 +881,12 @@ async def test_exchange_tasks_start_two_loops_and_stop(monkeypatch: pytest.Monke
await _wait_until(lambda: len(started) == EXCHANGE_TASKS)
assert len(exchange.tasks) == EXCHANGE_TASKS
assert started["publisher"] == {"poll_seconds": 7, "ack_timeout": timedelta(seconds=11), "max_republish": 2}
assert started["publisher"] == {
"poll_seconds": 7,
"ack_timeout": timedelta(seconds=11),
"max_republish": 2,
"request_subjects": {"tabular": "files.tabular_uploaded", "image": "files.img_uploaded"},
}
assert started["watchdog"] == {"period_seconds": WATCHDOG_PERIOD_SECONDS}
await exchange.stop()
assert all(task.done() for task in exchange.tasks)

View File

@ -0,0 +1,237 @@
"""Временный режим «период ОСВ из версии консолидации» (arm-core#103, `CONSOLIDATION_PERIOD_FROM_VERSION`).
ОСВ, загруженная в консолидацию (`stored_file.consolidation_id`), при включённом режиме считается нарастающей за год и
квартал консолидации: привязывается при подтверждении и подхвате, попадает в свод, чек-лист и колонку «ОСВ» периметра.
Выключенный режим, файлы вне консолидации и ББ/ОФР — период из шапки, как раньше.
"""
import datetime
import uuid
from collections.abc import Sequence
import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from src.core.config import settings
from src.models.access import AppUser
from src.models.consolidation import (
Consolidation,
ConsolidationDocument,
ConsolidationMember,
ConsolidationVersion,
)
from src.models.directories import LegalEntity, RegionalOffice
from src.models.enums import DocumentStatus, PeriodQuarter
from src.models.files import Document, StoredFile
from src.services import consolidation_documents
from tests.conftest import AuthClientFactory
from tests.factories import make_company_group, make_confirmed_osv, make_consolidation, make_legal_entity
HTTP_OK = 200
KA_RF = "ka_rf@armka.local"
ROSTOV = "Ростовский региональный филиал"
VERSION_YEAR = 2026 # консолидация 2026 Q1
HEADER_YEAR = 2025 # а ОСВ в шапке — за 1 кв. 2025
ACCOUNT = "60"
ROWS = [{"name": "ООО Контрагент", "inn": "7701234567", "closing_debit": 10}]
Q1, Q2 = PeriodQuarter.Q1, PeriodQuarter.Q2
@pytest.fixture
def period_from_version(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(settings, "CONSOLIDATION_PERIOD_FROM_VERSION", True)
async def _world(
session: AsyncSession, members: int = 2
) -> tuple[Consolidation, ConsolidationVersion, list[LegalEntity]]:
office = (await session.execute(select(RegionalOffice).where(RegionalOffice.name == ROSTOV))).scalar_one()
group = await make_company_group(session, name=f"ГК {uuid.uuid4().hex[:6]}", office=office)
entities = [await make_legal_entity(session, name=f"ООО Член {n}", group=group) for n in range(members + 1)]
user = (await session.execute(select(AppUser).where(AppUser.email == KA_RF))).scalar_one()
consolidation, version = await make_consolidation(
session, group=group, year=VERSION_YEAR, quarter=Q1, members=entities[:members], initiator=user
)
return consolidation, version, entities
async def _osv( # noqa: PLR0913 - тестовая фабрика: ЮЛ, консолидация файла, период, статус, время подтверждения
session: AsyncSession,
entity: LegalEntity,
consolidation: Consolidation | None,
*,
quarter: PeriodQuarter = Q1,
cumulative: bool = True,
status: DocumentStatus = DocumentStatus.NEEDS_CHECK,
confirmed_at: datetime.datetime | None = None,
) -> Document:
"""ОСВ 60 за `HEADER_YEAR` со строкой контрагента; файл загружен в `consolidation` (`None` — рабочий стол)."""
document = await make_confirmed_osv(
session, entity=entity, account=ACCOUNT, year=HEADER_YEAR, quarter=quarter, is_cumulative=cumulative, rows=ROWS
)
file = await session.get(StoredFile, document.stored_file_id)
assert file is not None
file.consolidation_id = consolidation.id if consolidation is not None else None
document.status = status
document.confirmed_at = confirmed_at if status is DocumentStatus.CONFIRMED else None
await session.flush()
return document
async def _bound(session: AsyncSession, version_id: uuid.UUID) -> set[uuid.UUID]:
stmt = select(ConsolidationDocument.document_id).where(ConsolidationDocument.consolidation_version_id == version_id)
return set((await session.execute(stmt)).scalars())
async def _confirm(auth_client: AuthClientFactory, documents: Sequence[Document]) -> None:
client = await auth_client(KA_RF)
for document in documents:
response = await client.post(f"/documents/{document.id}/confirm")
assert response.status_code == HTTP_OK, response.text
@pytest.mark.integration
@pytest.mark.usefixtures("period_from_version")
async def test_confirmed_osv_of_other_period_goes_to_version_and_summary(
seeded: AsyncSession, auth_client: AuthClientFactory
) -> None:
"""Сценарий владельца 12.10.2026: ОСВ за 1 кв. 2025 в консолидации 2026 Q1 после подтверждения — в своде."""
consolidation, version, (first, second, _) = await _world(seeded)
documents = [await _osv(seeded, entity, consolidation) for entity in (first, second)]
await _confirm(auth_client, documents)
assert await _bound(seeded, version.id) == {document.id for document in documents}
client = await auth_client(KA_RF)
summary = await client.get(f"/consolidations/versions/{version.id}/osv?limit=100")
assert summary.status_code == HTTP_OK, summary.text
assert summary.json()["total"] == len(documents) * len(ROWS)
# первичка не мутируется: период документа — из шапки
document = await seeded.get(Document, documents[0].id, populate_existing=True)
assert document is not None
assert (document.period_year, document.period_quarter) == (HEADER_YEAR, Q1)
@pytest.mark.integration
async def test_without_mode_osv_of_other_period_stays_unbound(
seeded: AsyncSession, auth_client: AuthClientFactory
) -> None:
consolidation, version, (first, _, _) = await _world(seeded)
document = await _osv(seeded, first, consolidation)
await _confirm(auth_client, [document])
assert await _bound(seeded, version.id) == set()
@pytest.mark.integration
@pytest.mark.usefixtures("period_from_version")
async def test_mode_ignores_osv_uploaded_outside_consolidation(
seeded: AsyncSession, auth_client: AuthClientFactory
) -> None:
"""Файл с рабочего стола (`consolidation_id` пуст) — период из шапки: 2025 не подходит версии 2026."""
_, version, (first, _, _) = await _world(seeded)
document = await _osv(seeded, first, None)
await _confirm(auth_client, [document])
assert await _bound(seeded, version.id) == set()
@pytest.mark.integration
@pytest.mark.usefixtures("period_from_version")
async def test_checklist_and_card_see_unconfirmed_osv_of_other_period(
seeded: AsyncSession, auth_client: AuthClientFactory
) -> None:
consolidation, version, (first, _, _) = await _world(seeded)
await _osv(seeded, first, consolidation)
client = await auth_client(KA_RF)
checklist = (await client.get(f"/consolidations/versions/{version.id}/checklist")).json()
card = (await client.get(f"/consolidations/versions/{version.id}")).json()
rules = [(rule["document_type"], rule["account"]) for rule in checklist["rules"]]
row = next(row for row in checklist["rows"] if row["legal_entity_id"] == str(first.id))
assert row["cells"][rules.index(("OSV", ACCOUNT))]["status"] == "NEEDS_CHECK"
member = next(member for member in card["members"] if member["id"] == str(first.id))
assert member["osv_accounts"] == [ACCOUNT]
@pytest.mark.integration
@pytest.mark.usefixtures("period_from_version")
async def test_perimeter_pickup_binds_one_osv_per_account_latest_confirmed(seeded: AsyncSession) -> None:
"""Подхват (`bind_confirmed`): все ОСВ счёта одного ЮЛ в консолидации — один слот, побеждает поздняя."""
consolidation, version, (_, _, newcomer) = await _world(seeded)
earlier = datetime.datetime(2026, 10, 12, 9, tzinfo=datetime.UTC)
later = earlier + datetime.timedelta(hours=1)
await _osv(
seeded,
newcomer,
consolidation,
quarter=Q1,
cumulative=False,
status=DocumentStatus.CONFIRMED,
confirmed_at=earlier,
)
latest = await _osv(
seeded,
newcomer,
consolidation,
quarter=Q2,
cumulative=False,
status=DocumentStatus.CONFIRMED,
confirmed_at=later,
)
seeded.add(
ConsolidationMember(consolidation_version_id=version.id, legal_entity_id=newcomer.id, is_head_entity=False)
)
await seeded.flush()
bound = await consolidation_documents.bind_confirmed(
seeded, version=version, consolidation=consolidation, entity_ids=[newcomer.id]
)
assert bound == 1
assert await _bound(seeded, version.id) == {latest.id}
@pytest.mark.integration
async def test_mode_keeps_header_period_when_it_fits_version(
seeded: AsyncSession, auth_client: AuthClientFactory, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Регрессия: квартальные ОСВ Q1–Q3, загруженные в консолидацию Q3 и привязанные до включения режима, в режиме
по-прежнему склеиваются в одну строку свода (их период из шапки подходит версии — подмены нет), а не идут «как есть»
трижды как нарастающие за Q3."""
office = (await seeded.execute(select(RegionalOffice).where(RegionalOffice.name == ROSTOV))).scalar_one()
group = await make_company_group(seeded, name=f"ГК {uuid.uuid4().hex[:6]}", office=office)
entities = [await make_legal_entity(seeded, name=f"ООО Член {n}", group=group) for n in range(2)]
user = (await seeded.execute(select(AppUser).where(AppUser.email == KA_RF))).scalar_one()
consolidation, version = await make_consolidation(
seeded, group=group, year=HEADER_YEAR, quarter=PeriodQuarter.Q3, members=entities, initiator=user
)
quarters = (Q1, Q2, PeriodQuarter.Q3)
for quarter in quarters:
document = await make_confirmed_osv(
seeded,
entity=entities[0],
account=ACCOUNT,
year=HEADER_YEAR,
quarter=quarter,
is_cumulative=False,
rows=[{"name": "ООО Контрагент", "inn": "7701234567", "turnover_debit": 10}],
)
file = await seeded.get(StoredFile, document.stored_file_id)
assert file is not None
file.consolidation_id = consolidation.id
await seeded.flush()
assert await consolidation_documents.bind_confirmed(
seeded, version=version, consolidation=consolidation, entity_ids=[entities[0].id]
) == len(quarters) # режим выключен: три квартальных источника
monkeypatch.setattr(settings, "CONSOLIDATION_PERIOD_FROM_VERSION", True)
client = await auth_client(KA_RF)
body = (await client.get(f"/consolidations/versions/{version.id}/osv?limit=100")).json()
assert body["total"] == 1, body
assert body["items"][0]["turnover_debit"] == "30.00"

View File

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

View File

@ -0,0 +1,20 @@
## Context
Дизайн сквозной: `../docs/openspec/changes/nats-env-config/design.md` (решения 1–7, R-1…R-6). Здесь — только то, что
специфично для `arm-core`.
## Decisions
1. Ядро публикует запросы в `tabular_request` и `image_request`, подписан на `result` (queue group `arm-core`); в `/healthcheck.subjects` — `tabular_request`, `image_request`, `result`.
2. Проверяются все три subject'а, включая неиспользуемые сервисом (docs design, решение 1 и R-3).
3. Выбор subject'а и подписки получают имена из `settings.OPENBAO.NATS`; константы `contract/subjects.py` остаются
умолчаниями полей настроек. Форма передачи (аргумент, объект subject'ов) — в плане.
## Risks / Trade-offs
См. docs design: разъехавшиеся настройки — тишина на шине; JWT без прав на заданные subject'ы — ошибка прав NATS в логе.
## Decisions made without discussion
Решения без обсуждения ведутся в сквозном `../docs/openspec/changes/nats-env-config/design.md` (R-1…R-6); решения,
принятые при исполнении в этом репозитории, дописываются сюда с маркером `R-n` сквозной нумерации.

View File

@ -0,0 +1,40 @@
## Why
Прод и тест будут работать с разными наборами subject'ов NATS и разными JWT; платформа передаёт в под только
`OPENBAO__*`. Сейчас subject'ы — константы `api/src/contract/subjects.py`. Решение владельца 12.10.2026: subject'ы и
JWT задаются по окружению секретами группы `OPENBAO__NATS__`, как в `RSHB/armka_file_worker`. Сквозной change —
`../docs/openspec/changes/nats-env-config` (RSHB/arm-ka-docs#80); здесь — его часть для `arm-core` (RSHB/arm-core#106).
Срез: вне плана срезов `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` и `image_request`, подписан на `result` (queue group `arm-core`) — по настройкам; константы `contract/subjects.py` — только умолчания.
- Лог при старте и `/healthcheck.subjects` (`tabular_request`, `image_request`, `result`), независимо от состояния NATS.
- e2e (`tests/e2e/`) остаётся на subject'ах по умолчанию: ему нужен контейнер `tabular-parser` стенда (docs design R-7).
- `.env.example`, README.
## Non-goals
- `schema_version` контракта и копии схем не меняются (кроме побайтной синхронизации `description` из docs, если она
изменится).
- Queue group'ы — константы; механизм nkeys/JWT не меняется.
- Пароли MinIO/SMB в `SecretStr` не переводятся.
- Настройка через `OPENBAO__APP__*` не вводится.
## Capabilities
### New Capabilities
### Modified Capabilities
- `recognition-job-exchange`: subject'ы обмена и учётные данные NATS — настройки окружения; `/healthcheck.subjects`.
## Impact
Файлы: `api/src/core/config.py`, `api/src/contract/subjects.py`, `api/src/core/nats.py`, `api/src/services/recognition_exchange.py`, `api/src/services/primary_files.py`, `api/src/app.py` (`/healthcheck`, лог старта), тесты (`tests/test_config*.py`, `tests/test_nats_*.py`, `tests/e2e/`), `api/.env.example`, README. Выкат обратно совместим (умолчания — текущие имена); набор окружения задаётся одинаково всем трём
сервисам.

View File

@ -0,0 +1,46 @@
## 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` и `image_request`, подписан на `result` (queue group `arm-core`).
При старте SHALL проверяться все три имени (непустое, без пробельных символов, без `*` и `>`, без пустых токенов,
попарно различны); при нарушении приложение MUST NOT стартовать. Значения `NATS_JWT` и `NATS_SEED` MUST NOT попадать в
логи, `repr` настроек и тексты ошибок валидации. Используемые subject'ы SHALL писаться в лог при старте и отдаваться в
`/healthcheck` объектом `subjects` с ключами `tabular_request`, `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` настроек показывает их скрытыми
## MODIFIED Requirements
### Requirement: Соединение с NATS по шаблону, без JetStream
Ядро SHALL подключаться к NATS с nkeys-seed и JWT, если заданы `OPENBAO__NATS__NATS_SEED/NATS_JWT`, и без учётных данных,
если они пусты; подписываться на subject ответа `NATS_RESULT_SUBJECT` (по умолчанию `files.tabular_parsed`) в queue group `arm-core`; переподключаться автоматически.
Stream'ы и consumer'ы JetStream MUST NOT требоваться. Фоновые задачи SHALL запускаться при `NATS_ENABLED=true`.
#### Scenario: Dev-окружение compose
- **WHEN** ядро стартует с `DEBUG=true`, `NATS_ENABLED=true` и пустыми учётными данными против NATS без JetStream
- **THEN** подписка активна, `/healthcheck` отдаёт `nats = ok`, в логах нет ошибок подключения
#### Scenario: Два инстанса ядра
- **WHEN** запущены два инстанса с `NATS_ENABLED=true` и опубликован один результат
- **THEN** результат обработан ровно одним инстансом (queue group), строка задания изменена один раз
#### Scenario: NATS выключен
- **WHEN** `NATS_ENABLED=false`
- **THEN** приложение стартует, фоновые задачи не запускаются, `/healthcheck` отвечает 200 с `nats = off`

View File

@ -0,0 +1,18 @@
## 1. Тесты
- [x] 1.1 Unit: умолчания subject'ов, чтение `OPENBAO__NATS__NATS_*_SUBJECT`, отказ на `*`/`>`/пробеле/пустом токене и
совпадающих именах; `SecretStr` для `NATS_JWT`/`NATS_SEED`, отсутствие значений в ошибке валидации и `repr`
- [x] 1.2 Unit: `/healthcheck.subjects` (`tabular_request`, `image_request`, `result`) при `nats = ok | down | off`; лог старта с subject'ами
- [x] 1.3 Интеграционный тест: приложение с 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` зелёный

View File

@ -63,7 +63,10 @@ API справочников ядра: группы компаний (ГК), ю
членствами ГСК/ГСЗ и алиасами; `GET …/{id}/history` — снимки. В списке и карточке SHALL быть `documents_count` — число
документов файлов ЮЛ, кроме файлов `SUPERSEDED`. Во всех ответах ЮЛ, кроме снимков истории, SHALL быть `company_group_name` —
текущее имя ГК `company_group_id` (архивной тоже) или `null` у монозаёмщика; имена набора ЮЛ — одним запросом.
Ответ `POST` и `PATCH` — карточка (`LegalEntityDetail`: поля списка + `memberships`, `aliases`); поля формы
Во всех ответах ЮЛ, кроме снимков истории, SHALL быть `memberships` — текущие членства ЮЛ в ГСК/ГСЗ
(`related_group_id`, `code`, `kind`, `name`, `valid_from`; по виду и коду группы; закрытые не отдаются; нет — `[]`);
членства набора ЮЛ — одним запросом (#109: фильтр ЮЛ по ГСК/ГСЗ в карточке ГК без запроса карточки каждого ЮЛ).
Ответ `POST` и `PATCH` — карточка (`LegalEntityDetail`: поля списка + `aliases`); поля формы
`aliases`, `gsk_id`, `gsz_id` — требование «Форма ЮЛ одним запросом».
#### Scenario: Неверная длина ИНН
@ -82,6 +85,10 @@ API справочников ядра: группы компаний (ГК), ю
- **WHEN** ЮЛ создано без `company_group_id`
- **THEN** оно отдаётся в списке с `outside_group = true` и видно пользователям филиала `regional_office_id`
#### Scenario: Членства в списке
- **WHEN** ЮЛ ГК состоит в ГСК и ГСЗ, другое ЮЛ той же ГК — только в закрытом членстве ГСК
- **THEN** в `GET …?company_group_id=…` у первого `memberships` — ГСК и ГСЗ, у второго — `[]`
#### Scenario: Число документов
- **WHEN** у монозаёмщика файл с книгой НПЗ (ББ и ОФР) и заменённый файл с одной ОСВ
- **THEN** `documents_count = 2`
@ -156,7 +163,13 @@ SHALL установить `company_group_id = NULL`, при `CLOSE` — ещё
`DIFFERENT_COMPANY_GROUP`); если у ЮЛ уже есть текущее членство в группе того же вида — 409 `MEMBER_OF_OTHER_GROUP`
(с `related_group_id`); повтор в той же группе — 409 `MEMBER_EXISTS`. `DELETE …/members/{legal_entity_id}` SHALL закрывать
членство (`valid_to`, `is_current = false`). Все операции пишут `MEMBERSHIP_CHANGE`. `GET …/related-groups` и
`GET …/related-groups/{id}` SHALL отдавать группы с числом текущих членов и списком членов.
`GET …/related-groups/{id}` SHALL отдавать группы с числом текущих членов и списком членов. Тела `POST` и `PATCH` SHALL
принимать необязательный `crm_id` («ID ГСК/ГСЗ в CRM» форм 51-03/51-04/77-02) и отбрасывать его: колонки у `related_group`
нет (arm-core #112); в ответах `crm_id` всегда `null`, в аудит и в список изменений `PATCH` не попадает.
#### Scenario: `crm_id` ГСК принимается и не сохраняется
- **WHEN** ГСК создаётся с `crm_id = "3-9CGSK01"`, затем `PATCH` передаёт только `crm_id`
- **THEN** создание — 201 с `crm_id = null`; `PATCH` — 200 без изменений и без события аудита
#### Scenario: Одновременно в ГСК и ГСЗ
- **WHEN** ЮЛ добавлено в ГСК своей ГК, затем в ГСЗ той же ГК

View File

@ -91,7 +91,7 @@
### Requirement: Соединение с NATS по шаблону, без JetStream
Ядро SHALL подключаться к NATS с nkeys-seed и JWT, если заданы `OPENBAO__NATS__NATS_SEED/NATS_JWT`, и без учётных данных,
если они пусты; подписываться на `files.tabular_parsed` в queue group `arm-core`; переподключаться автоматически.
если они пусты; подписываться на subject ответа `NATS_RESULT_SUBJECT` (по умолчанию `files.tabular_parsed`) в queue group `arm-core`; переподключаться автоматически.
Stream'ы и consumer'ы JetStream MUST NOT требоваться. Фоновые задачи SHALL запускаться при `NATS_ENABLED=true`.
#### Scenario: Dev-окружение compose
@ -105,3 +105,29 @@ Stream'ы и consumer'ы JetStream MUST NOT требоваться. Фоновы
#### Scenario: NATS выключен
- **WHEN** `NATS_ENABLED=false`
- **THEN** приложение стартует, фоновые задачи не запускаются, `/healthcheck` отвечает 200 с `nats = off`
### 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` и `image_request`, подписан на `result` (queue group `arm-core`).
При старте SHALL проверяться все три имени (непустое, без пробельных символов, без `*` и `>`, без пустых токенов,
попарно различны); при нарушении приложение MUST NOT стартовать. Значения `NATS_JWT` и `NATS_SEED` MUST NOT попадать в
логи, `repr` настроек и тексты ошибок валидации. Используемые subject'ы SHALL писаться в лог при старте и отдаваться в
`/healthcheck` объектом `subjects` с ключами `tabular_request`, `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` настроек показывает их скрытыми