290 lines
12 KiB
Python
290 lines
12 KiB
Python
import os
|
||
from collections.abc import Callable
|
||
|
||
import pytest
|
||
from pydantic import ValidationError
|
||
|
||
from src.core.config import Settings, get_settings
|
||
|
||
SMB_ENV = {
|
||
"OPENBAO__SMB__SMB_HOST": "smb.example",
|
||
"OPENBAO__SMB__SMB_PATH": "share",
|
||
"OPENBAO__SMB__SMB_LOGIN": "user",
|
||
"OPENBAO__SMB__SMB_PASSWORD": "secret",
|
||
}
|
||
|
||
|
||
def _clean_env(monkeypatch):
|
||
"""Убрать из окружения всё, что относится к SMB/NATS-учётным данным и режиму горячей папки."""
|
||
for key in list(SMB_ENV) + ["OPENBAO__NATS__NATS_SEED", "OPENBAO__NATS__NATS_JWT", "SCAN_HOTFOLDER_MODE"]:
|
||
monkeypatch.delenv(key, raising=False)
|
||
|
||
|
||
def test_defaults_local_mode_without_smb_and_nats_credentials(monkeypatch, minimal_env_file):
|
||
_clean_env(monkeypatch)
|
||
|
||
s = get_settings()
|
||
|
||
assert s.SERVICE_NAME == "scan-gateway"
|
||
assert s.NATS_ENABLED is True
|
||
assert s.PUBLISH_RETRIES == 3
|
||
assert s.DONE_CACHE_SIZE == 1000
|
||
assert s.SCAN_HOTFOLDER_MODE == "local"
|
||
assert s.SCAN_INPUT_DIR == "Input"
|
||
assert s.SCAN_OUTPUT_DIR == "Output"
|
||
assert not hasattr(s, "SCAN_TMP_DIR") # служебный каталог только у эмулятора (change hotfolder-direct-write)
|
||
assert s.SCAN_SETTLE_SECONDS == 2
|
||
assert s.SCAN_POLL_SECONDS == 5
|
||
assert s.SCAN_TIMEOUT_SECONDS == 1200
|
||
assert s.SCAN_MAX_INFLIGHT == 4
|
||
assert s.OPENBAO.SMB.SMB_HOST == ""
|
||
assert s.OPENBAO.NATS.NATS_SEED.get_secret_value() == ""
|
||
assert s.OPENBAO.NATS.NATS_JWT.get_secret_value() == ""
|
||
assert not hasattr(s.OPENBAO, "DATABASE")
|
||
|
||
|
||
def test_local_mode_requires_hotfolder_path(monkeypatch):
|
||
_clean_env(monkeypatch)
|
||
monkeypatch.setenv("SCAN_HOTFOLDER_PATH", "")
|
||
|
||
with pytest.raises(ValidationError, match="SCAN_HOTFOLDER_PATH"):
|
||
get_settings()
|
||
|
||
|
||
def test_smb_mode_without_smb_block_fails_with_clear_error(monkeypatch):
|
||
_clean_env(monkeypatch)
|
||
monkeypatch.setenv("SCAN_HOTFOLDER_MODE", "smb")
|
||
|
||
with pytest.raises(ValidationError, match="OPENBAO__SMB"):
|
||
get_settings()
|
||
|
||
|
||
def test_smb_mode_with_smb_block_ok(monkeypatch):
|
||
_clean_env(monkeypatch)
|
||
monkeypatch.setenv("SCAN_HOTFOLDER_MODE", "smb")
|
||
for key, value in SMB_ENV.items():
|
||
monkeypatch.setenv(key, value)
|
||
|
||
s = get_settings()
|
||
|
||
assert s.OPENBAO.SMB.SMB_HOST == "smb.example"
|
||
|
||
|
||
def test_settle_seconds_zero_is_allowed(monkeypatch):
|
||
monkeypatch.setenv("SCAN_SETTLE_SECONDS", "0")
|
||
|
||
assert get_settings().SCAN_SETTLE_SECONDS == 0
|
||
|
||
|
||
def test_negative_settle_seconds_rejected(monkeypatch):
|
||
monkeypatch.setenv("SCAN_SETTLE_SECONDS", "-1")
|
||
|
||
with pytest.raises(ValidationError, match="SCAN_SETTLE_SECONDS"):
|
||
get_settings()
|
||
|
||
|
||
def test_legacy_scan_tmp_dir_is_ignored(monkeypatch, minimal_env_file):
|
||
"""Старый `.env` с `SCAN_TMP_DIR` (до change hotfolder-direct-write) не ломает старт: `extra="ignore"`."""
|
||
_clean_env(monkeypatch)
|
||
monkeypatch.setenv("SCAN_TMP_DIR", "Tmp")
|
||
|
||
s = get_settings()
|
||
|
||
assert not hasattr(s, "SCAN_TMP_DIR")
|
||
|
||
|
||
@pytest.mark.parametrize("settle", ["10", "11"])
|
||
def test_settle_not_less_than_timeout_rejected(monkeypatch, settle):
|
||
"""M-4: набор не может настояться за срок, который короче или равен settle — каждое задание давало бы TIMEOUT."""
|
||
monkeypatch.setenv("SCAN_TIMEOUT_SECONDS", "10")
|
||
monkeypatch.setenv("SCAN_SETTLE_SECONDS", settle)
|
||
|
||
with pytest.raises(ValidationError, match="SCAN_SETTLE_SECONDS .* меньше SCAN_TIMEOUT_SECONDS"):
|
||
get_settings()
|
||
|
||
|
||
def test_settle_less_than_timeout_is_allowed(monkeypatch):
|
||
monkeypatch.setenv("SCAN_TIMEOUT_SECONDS", "10")
|
||
monkeypatch.setenv("SCAN_SETTLE_SECONDS", "9.9")
|
||
|
||
assert get_settings().SCAN_SETTLE_SECONDS == 9.9
|
||
|
||
|
||
def test_unknown_hotfolder_mode_rejected(monkeypatch):
|
||
monkeypatch.setenv("SCAN_HOTFOLDER_MODE", "ftp")
|
||
|
||
with pytest.raises(ValidationError):
|
||
get_settings()
|
||
|
||
|
||
def test_env_file_selected_by_env_file_variable(monkeypatch, tmp_path):
|
||
_clean_env(monkeypatch)
|
||
for key in (
|
||
"SCAN_HOTFOLDER_PATH",
|
||
"SCAN_POLL_SECONDS",
|
||
"SERVER__PORT",
|
||
"OPENBAO__NATS__NATS_URL",
|
||
"OPENBAO__MINIO__MINIO_URL",
|
||
"OPENBAO__MINIO__MINIO_ADMIN_LOGIN",
|
||
"OPENBAO__MINIO__MINIO_ADMIN_PASSWORD",
|
||
"OPENBAO__MINIO__MINIO_BUCKET",
|
||
):
|
||
monkeypatch.delenv(key, raising=False)
|
||
env_file = tmp_path / "custom.env"
|
||
env_file.write_text(
|
||
"\n".join(
|
||
[
|
||
"OPENBAO__MINIO__MINIO_URL=minio.test:9000",
|
||
"OPENBAO__MINIO__MINIO_ADMIN_LOGIN=a",
|
||
"OPENBAO__MINIO__MINIO_ADMIN_PASSWORD=b",
|
||
"OPENBAO__MINIO__MINIO_BUCKET=bucket",
|
||
"OPENBAO__NATS__NATS_URL=nats://nats.test:4222",
|
||
"SCAN_HOTFOLDER_PATH=/data/hot",
|
||
"SCAN_POLL_SECONDS=1",
|
||
"SERVER__PORT=8123",
|
||
]
|
||
)
|
||
)
|
||
monkeypatch.setenv("ENV_FILE", str(env_file))
|
||
|
||
s = get_settings()
|
||
|
||
assert s.OPENBAO.MINIO.MINIO_URL == "minio.test:9000"
|
||
assert s.OPENBAO.NATS.NATS_URL == "nats://nats.test:4222"
|
||
assert s.SCAN_HOTFOLDER_PATH == "/data/hot"
|
||
assert s.SCAN_POLL_SECONDS == 1
|
||
assert s.server.PORT == 8123
|
||
|
||
|
||
def test_get_settings_is_cached(monkeypatch):
|
||
assert get_settings() is get_settings()
|
||
assert isinstance(get_settings(), Settings)
|
||
|
||
|
||
def test_env_test_file_loads_for_tests():
|
||
"""pytest.ini выбирает `.env.test`: он должен давать рабочие настройки local-режима с укороченными интервалами."""
|
||
s = get_settings()
|
||
|
||
assert s.SCAN_HOTFOLDER_MODE == "local"
|
||
assert s.SCAN_HOTFOLDER_PATH == "./.hotfolder-test"
|
||
assert s.SCAN_POLL_SECONDS == 0.1
|
||
assert s.SCAN_TIMEOUT_SECONDS == 5
|
||
assert not hasattr(s, "SCAN_TMP_DIR")
|
||
assert s.SCAN_SETTLE_SECONDS == 0 # R-23: тесты воркера не ждут стабилизации набора
|
||
assert s.OPENBAO.MINIO.MINIO_URL == "127.0.0.1:9000"
|
||
assert s.OPENBAO.MINIO.MINIO_SECURE is False
|
||
assert s.OPENBAO.NATS.NATS_URL == "nats://127.0.0.1:4222"
|
||
assert s.OPENBAO.NATS.NATS_SEED.get_secret_value() == ""
|
||
|
||
|
||
def test_env_example_loads(monkeypatch):
|
||
"""`.env.example` — рабочий dev-набор: Settings читает его без ошибок."""
|
||
monkeypatch.setenv("ENV_FILE", ".env.example")
|
||
|
||
s = get_settings()
|
||
|
||
assert s.SCAN_HOTFOLDER_MODE == "local"
|
||
assert s.SCAN_INPUT_DIR == "Input"
|
||
assert not hasattr(s, "SCAN_TMP_DIR")
|
||
assert s.SCAN_SETTLE_SECONDS == 2
|
||
assert s.OPENBAO.MINIO.MINIO_BUCKET == "arm-ka"
|
||
assert s.OPENBAO.NATS.NATS_JWT.get_secret_value() == ""
|
||
|
||
|
||
# --- настройки NATS: subject'ы и секреты окружения OPENBAO__NATS__* (change nats-env-config) ----------------------
|
||
|
||
REQUIRED_OPENBAO_ENV = {
|
||
"OPENBAO__MINIO__MINIO_URL": "localhost:9000",
|
||
"OPENBAO__MINIO__MINIO_ADMIN_LOGIN": "minioadmin",
|
||
"OPENBAO__MINIO__MINIO_ADMIN_PASSWORD": "minioadmin",
|
||
"OPENBAO__MINIO__MINIO_BUCKET": "arm-ka",
|
||
"OPENBAO__NATS__NATS_URL": "nats://localhost:4222",
|
||
}
|
||
|
||
|
||
@pytest.fixture
|
||
def settings_from_env(monkeypatch, tmp_path) -> Callable[..., Settings]:
|
||
"""Фабрика `Settings(_env_file=None)`: минимальный набор обязательных `OPENBAO__*` плюс `env` поверх; значение
|
||
`None` удаляет переменную. Файл окружения не читается: иначе удалённая переменная осталась бы в нём."""
|
||
|
||
def build(**env: str | None) -> Settings:
|
||
_clean_env(monkeypatch)
|
||
for key in [key for key in os.environ if key.startswith("OPENBAO__")]:
|
||
monkeypatch.delenv(key)
|
||
values = {**REQUIRED_OPENBAO_ENV, "SCAN_HOTFOLDER_PATH": str(tmp_path), **env}
|
||
for key, value in values.items():
|
||
if value is None:
|
||
monkeypatch.delenv(key, raising=False)
|
||
else:
|
||
monkeypatch.setenv(key, value)
|
||
return Settings(_env_file=None) # type: ignore[call-arg]
|
||
|
||
return build
|
||
|
||
|
||
JWT = "eyJ0eXAiOiJKV1QiLCJhbGciOiJlZDI1NTE5LW5rZXkifQ.secret-payload.sig"
|
||
SEED = "SUACSSL3UAHUDXKFSNVUZRF5UHPMWZ6BFDTJ7M6USDXIEDNPPQYYYCU3VY"
|
||
|
||
|
||
def test_nats_subjects_default_to_contract_names(settings_from_env) -> None:
|
||
nats = settings_from_env().OPENBAO.NATS
|
||
assert nats.NATS_TABULAR_REQUEST_SUBJECT == "files.tabular_uploaded"
|
||
assert nats.NATS_IMAGE_REQUEST_SUBJECT == "files.img_uploaded"
|
||
assert nats.NATS_RESULT_SUBJECT == "files.tabular_parsed"
|
||
assert nats.request_subjects() == {"tabular": "files.tabular_uploaded", "image": "files.img_uploaded"}
|
||
|
||
|
||
def test_nats_subjects_from_openbao(settings_from_env) -> None:
|
||
nats = settings_from_env(
|
||
OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT="test.files.tabular_uploaded",
|
||
OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT="test.files.img_uploaded",
|
||
OPENBAO__NATS__NATS_RESULT_SUBJECT="test.files.tabular_parsed",
|
||
).OPENBAO.NATS
|
||
assert nats.request_subjects() == {"tabular": "test.files.tabular_uploaded", "image": "test.files.img_uploaded"}
|
||
assert nats.NATS_RESULT_SUBJECT == "test.files.tabular_parsed"
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"value",
|
||
["", "files.*", "files.>", ">", "files..tabular", ".files", "files.", "files tabular", "files.tabular_uploaded "],
|
||
)
|
||
def test_invalid_subject_rejected(settings_from_env, value: str) -> None:
|
||
with pytest.raises(ValidationError, match="NATS_RESULT_SUBJECT"):
|
||
settings_from_env(OPENBAO__NATS__NATS_RESULT_SUBJECT=value)
|
||
|
||
|
||
def test_equal_subjects_rejected(settings_from_env) -> None:
|
||
with pytest.raises(ValidationError, match="NATS_TABULAR_REQUEST_SUBJECT и NATS_RESULT_SUBJECT совпадают"):
|
||
settings_from_env(OPENBAO__NATS__NATS_RESULT_SUBJECT="files.tabular_uploaded")
|
||
|
||
|
||
def test_nats_secrets_hidden_in_repr(settings_from_env) -> None:
|
||
settings = settings_from_env(OPENBAO__NATS__NATS_JWT=JWT, OPENBAO__NATS__NATS_SEED=SEED)
|
||
assert settings.OPENBAO.NATS.NATS_JWT.get_secret_value() == JWT
|
||
assert JWT not in repr(settings) and SEED not in repr(settings)
|
||
assert JWT not in str(settings) and SEED not in str(settings)
|
||
|
||
|
||
MINIO_KEYS = ("MINIO_URL", "MINIO_ADMIN_LOGIN", "MINIO_ADMIN_PASSWORD", "MINIO_BUCKET", "MINIO_SECURE")
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"missing",
|
||
[
|
||
("OPENBAO__NATS__NATS_URL",), # нет поля в группе NATS: в ошибке — вход группы NATS (с JWT)
|
||
tuple(f"OPENBAO__MINIO__{key}" for key in MINIO_KEYS), # нет группы MINIO: в ошибке — вход всего OPENBAO
|
||
],
|
||
)
|
||
def test_validation_error_hides_nats_secrets(settings_from_env, missing: tuple[str, ...]) -> None:
|
||
"""Ошибка «нет поля» печатает вход всей группы — JWT и seed туда попадать не должны (Review Focus 1).
|
||
|
||
pydantic урезает длинные значения в тексте ошибки: видны начало и конец секрета, а не он целиком. Поэтому проверяем
|
||
и голову, и хвост каждого секрета (без скрытия входа в тексте видны оба куска)."""
|
||
with pytest.raises(ValidationError) as exc_info:
|
||
settings_from_env(OPENBAO__NATS__NATS_JWT=JWT, OPENBAO__NATS__NATS_SEED=SEED, **dict.fromkeys(missing))
|
||
text = str(exc_info.value)
|
||
for secret in (JWT, SEED):
|
||
assert secret not in text
|
||
assert secret[:10] not in text
|
||
assert secret[-10:] not in text
|