4f0636f627
T4.1, T2.1, T2.2, T5.1, T5.2, T5.3 (тесты); T5.6 (subscribe_requests: пустая строка не подменяется умолчанием). Отмечены пункты tasks.md 1.1, 1.2, 2.x, 3.1. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
177 lines
8.7 KiB
Python
177 lines
8.7 KiB
Python
"""Unit-тесты настроек `src/core/config.py` (spec «Старт с минимальным окружением»; R-35)."""
|
||
|
||
from collections.abc import Callable
|
||
|
||
import pytest
|
||
from pydantic import ValidationError
|
||
|
||
from src.core.config import Settings
|
||
|
||
SettingsFactory = Callable[..., Settings]
|
||
SettingsFromEnvFactory = Callable[..., Settings]
|
||
|
||
# Минимальный набор обязательных переменных (как MINIMAL_ENV в conftest): окружение процесса, без .env
|
||
MINIMAL_ENV_VARS = {
|
||
"DEBUG": "true",
|
||
"OPENBAO__MINIO__MINIO_URL": "localhost:9000",
|
||
"OPENBAO__MINIO__MINIO_ADMIN_LOGIN": "minioadmin",
|
||
"OPENBAO__MINIO__MINIO_ADMIN_PASSWORD": "minioadmin",
|
||
"OPENBAO__MINIO__MINIO_BUCKET": "arm-ka",
|
||
"OPENBAO__NATS__NATS_URL": "nats://localhost:4222",
|
||
}
|
||
|
||
|
||
@pytest.fixture
|
||
def settings_from_env(minimal_settings: SettingsFactory, monkeypatch: pytest.MonkeyPatch) -> SettingsFromEnvFactory:
|
||
"""Фабрика `settings_from_env(**env)`: Settings из переменных окружения поверх минимального набора.
|
||
|
||
Окружение процесса очищено фикстурой `minimal_settings` (она же убирает `OPENBAO__*`); значение `None` — удалить
|
||
переменную. `.env` не читается (`_env_file=None`).
|
||
"""
|
||
|
||
def build(**env: str | None) -> Settings:
|
||
for name, value in MINIMAL_ENV_VARS.items():
|
||
monkeypatch.setenv(name, value)
|
||
for name, value in env.items():
|
||
if value is None:
|
||
monkeypatch.delenv(name, raising=False)
|
||
else:
|
||
monkeypatch.setenv(name, value)
|
||
return Settings(_env_file=None) # type: ignore[call-arg]
|
||
|
||
return build
|
||
|
||
|
||
JWT = "eyJ0eXAiOiJKV1QiLCJhbGciOiJlZDI1NTE5LW5rZXkifQ.secret-payload.sig"
|
||
SEED = "SUACSSL3UAHUDXKFSNVUZRF5UHPMWZ6BFDTJ7M6USDXIEDNPPQYYYCU3VY"
|
||
|
||
|
||
def test_minimal_environment_gives_defaults(minimal_settings: SettingsFactory) -> None:
|
||
settings = minimal_settings()
|
||
|
||
assert settings.DEBUG is True
|
||
assert settings.OPENBAO.MINIO.MINIO_BUCKET == "arm-ka"
|
||
assert settings.OPENBAO.NATS.NATS_URL == "nats://localhost:4222"
|
||
assert settings.OPENBAO.NATS.NATS_SEED.get_secret_value() == ""
|
||
assert settings.OPENBAO.NATS.NATS_JWT.get_secret_value() == ""
|
||
assert settings.NATS_ENABLED is True
|
||
assert settings.SERVICE_NAME == "tabular-parser"
|
||
assert settings.WORKER_CONCURRENCY == 1
|
||
assert settings.PUBLISH_RETRIES == 3 # noqa: PLR2004 - значение по умолчанию из spec
|
||
assert settings.DONE_CACHE_SIZE == 1000 # noqa: PLR2004 - значение по умолчанию из spec
|
||
assert settings.PARSE_MAX_BYTES == 50 * 1024 * 1024 # design D-8: 50 МиБ
|
||
assert settings.PARSE_TIMEOUT_SECONDS == 120 # noqa: PLR2004 - значение по умолчанию из design D-8
|
||
|
||
|
||
def test_parse_limits_must_be_positive(minimal_settings: SettingsFactory) -> None:
|
||
with pytest.raises(ValidationError, match="PARSE_MAX_BYTES"):
|
||
minimal_settings("PARSE_MAX_BYTES=0\n")
|
||
with pytest.raises(ValidationError, match="PARSE_TIMEOUT_SECONDS"):
|
||
minimal_settings("PARSE_TIMEOUT_SECONDS=0\n")
|
||
|
||
|
||
def test_publish_retries_zero_is_allowed(minimal_settings: SettingsFactory) -> None:
|
||
"""0 — без повторов: одна попытка публикации (NatsConnection принимает publish_retries ≥ 0)."""
|
||
assert minimal_settings("PUBLISH_RETRIES=0\n").PUBLISH_RETRIES == 0
|
||
|
||
|
||
def test_publish_retries_negative_is_rejected(minimal_settings: SettingsFactory) -> None:
|
||
with pytest.raises(ValidationError, match="PUBLISH_RETRIES"):
|
||
minimal_settings("PUBLISH_RETRIES=-1\n")
|
||
|
||
|
||
def test_service_name_other_than_contract_value_is_rejected(minimal_settings: SettingsFactory) -> None:
|
||
"""SERVICE_NAME — `service.name` конвертов контракта: другое значение — ошибка конфигурации при старте."""
|
||
with pytest.raises(ValidationError, match="SERVICE_NAME"):
|
||
minimal_settings("SERVICE_NAME=scan-gateway\n")
|
||
|
||
|
||
def test_nats_subjects_default_to_contract_names(settings_from_env: SettingsFromEnvFactory) -> None:
|
||
nats = settings_from_env().OPENBAO.NATS
|
||
assert nats.NATS_TABULAR_REQUEST_SUBJECT == "files.tabular_uploaded"
|
||
assert nats.NATS_IMAGE_REQUEST_SUBJECT == "files.img_uploaded"
|
||
assert nats.NATS_RESULT_SUBJECT == "files.tabular_parsed"
|
||
assert nats.request_subjects() == {"tabular": "files.tabular_uploaded", "image": "files.img_uploaded"}
|
||
|
||
|
||
def test_nats_subjects_from_openbao(settings_from_env: SettingsFromEnvFactory) -> None:
|
||
nats = settings_from_env(
|
||
OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT="test.files.tabular_uploaded",
|
||
OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT="test.files.img_uploaded",
|
||
OPENBAO__NATS__NATS_RESULT_SUBJECT="test.files.tabular_parsed",
|
||
).OPENBAO.NATS
|
||
assert nats.request_subjects() == {"tabular": "test.files.tabular_uploaded", "image": "test.files.img_uploaded"}
|
||
assert nats.NATS_RESULT_SUBJECT == "test.files.tabular_parsed"
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"value",
|
||
["", "files.*", "files.>", ">", "files..tabular", ".files", "files.", "files tabular", "files.tabular_uploaded "],
|
||
)
|
||
@pytest.mark.parametrize("key", ["NATS_TABULAR_REQUEST_SUBJECT", "NATS_IMAGE_REQUEST_SUBJECT", "NATS_RESULT_SUBJECT"])
|
||
def test_invalid_subject_rejected(settings_from_env: SettingsFromEnvFactory, key: str, value: str) -> None:
|
||
with pytest.raises(ValidationError, match=key):
|
||
settings_from_env(**{f"OPENBAO__NATS__{key}": value})
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("first", "second"),
|
||
[
|
||
("NATS_TABULAR_REQUEST_SUBJECT", "NATS_IMAGE_REQUEST_SUBJECT"),
|
||
("NATS_TABULAR_REQUEST_SUBJECT", "NATS_RESULT_SUBJECT"),
|
||
("NATS_IMAGE_REQUEST_SUBJECT", "NATS_RESULT_SUBJECT"),
|
||
],
|
||
)
|
||
def test_equal_subjects_rejected(settings_from_env: SettingsFromEnvFactory, first: str, second: str) -> None:
|
||
"""Любая пара из трёх subject'ов не может совпадать; сообщение называет оба ключа."""
|
||
with pytest.raises(ValidationError, match=f"{first} и {second} совпадают"):
|
||
settings_from_env(
|
||
**{f"OPENBAO__NATS__{first}": "test.same_subject", f"OPENBAO__NATS__{second}": "test.same_subject"}
|
||
)
|
||
|
||
|
||
def test_nats_secrets_hidden_in_repr(settings_from_env: SettingsFromEnvFactory) -> None:
|
||
settings = settings_from_env(OPENBAO__NATS__NATS_JWT=JWT, OPENBAO__NATS__NATS_SEED=SEED)
|
||
assert settings.OPENBAO.NATS.NATS_JWT.get_secret_value() == JWT
|
||
assert JWT not in repr(settings) and SEED not in repr(settings)
|
||
assert JWT not in str(settings) and SEED not in str(settings)
|
||
|
||
|
||
MINIO_KEYS = ("MINIO_URL", "MINIO_ADMIN_LOGIN", "MINIO_ADMIN_PASSWORD", "MINIO_BUCKET", "MINIO_SECURE")
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"missing",
|
||
[
|
||
("OPENBAO__NATS__NATS_URL",), # нет поля в группе NATS: в ошибке — вход группы NATS (с JWT)
|
||
tuple(f"OPENBAO__MINIO__{key}" for key in MINIO_KEYS), # нет группы MINIO: в ошибке — вход всего OPENBAO
|
||
],
|
||
)
|
||
def test_validation_error_hides_nats_secrets(
|
||
settings_from_env: SettingsFromEnvFactory, missing: tuple[str, ...]
|
||
) -> None:
|
||
"""Ошибка «нет поля» печатает вход всей группы — JWT и seed туда попадать не должны (Review Focus 1).
|
||
|
||
pydantic в тексте ошибки усекает значения (`'SUACSSL3UA...PQYYYCU3VY'`), поэтому полный секрет там не появляется
|
||
даже без скрытия; проверяем края секретов — они и попадают в текст при утечке.
|
||
"""
|
||
with pytest.raises(ValidationError) as exc_info:
|
||
settings_from_env(OPENBAO__NATS__NATS_JWT=JWT, OPENBAO__NATS__NATS_SEED=SEED, **dict.fromkeys(missing))
|
||
text = str(exc_info.value)
|
||
for secret in (JWT, SEED):
|
||
for fragment in (secret[:10], secret[-10:]):
|
||
assert fragment not in text
|
||
|
||
|
||
def test_invalid_subject_error_hides_nats_secrets(settings_from_env: SettingsFromEnvFactory) -> None:
|
||
"""Невалидный subject при заданных JWT/seed: ошибка валидации печатает вход группы NATS — секретов там нет."""
|
||
with pytest.raises(ValidationError) as exc_info:
|
||
settings_from_env(
|
||
OPENBAO__NATS__NATS_JWT=JWT, OPENBAO__NATS__NATS_SEED=SEED, OPENBAO__NATS__NATS_RESULT_SUBJECT="files.*"
|
||
)
|
||
text = str(exc_info.value)
|
||
assert "NATS_RESULT_SUBJECT" in text
|
||
for secret in (JWT, SEED):
|
||
for fragment in (secret[:10], secret[-10:]):
|
||
assert fragment not in text
|