db8b0bcc13
- invalid_subject параметризован по трём ключам, equality покрывает три пары - кейс утечки JWT/seed при невалидном subject - /healthcheck с нестандартными subject'ами и проверкой отсутствия JWT - subscribe с настроенным subject'ом, лог старта с subject'ами - tasks.md: отмечены 1.1, 1.2, 2.x, 3.1 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
257 lines
12 KiB
Python
257 lines
12 KiB
Python
"""Настройки: чтение ENV_FILE и значения по умолчанию (без БД и сети)."""
|
||
|
||
import os
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from pydantic import ValidationError
|
||
|
||
from src.core.config import Settings
|
||
|
||
API_DIR = Path(__file__).resolve().parent.parent
|
||
|
||
|
||
@pytest.fixture
|
||
def clean_env(monkeypatch: pytest.MonkeyPatch) -> pytest.MonkeyPatch:
|
||
"""Убирает из окружения процесса переменные, которые перебили бы значения из env-файла."""
|
||
for key in list(os.environ):
|
||
if key.startswith(("OPENBAO__", "AUTH_", "OUTBOX_", "ACK_", "SERVER__", "UPLOAD_", "MAPPING_")) or key in {
|
||
"NATS_ENABLED",
|
||
"DEBUG",
|
||
}:
|
||
monkeypatch.delenv(key)
|
||
return monkeypatch
|
||
|
||
|
||
def test_env_test_is_read(clean_env: pytest.MonkeyPatch) -> None:
|
||
""".env.test читается: стандартные порты docker-compose из docs/dev."""
|
||
s = Settings(_env_file=API_DIR / ".env.test") # type: ignore[call-arg]
|
||
assert s.OPENBAO.DATABASE.DATABASE_URL.endswith("localhost:5432/armka")
|
||
assert s.OPENBAO.NATS.NATS_URL == "nats://localhost:4222"
|
||
assert s.AUTH_DEV_ISSUER is True
|
||
assert s.NATS_ENABLED is False
|
||
assert s.OPENBAO.MINIO.MINIO_SECURE is False
|
||
assert s.OPENBAO.SMB.SMB_HOST == "" # SMB выключен: в логе запуска нет трейсбэка SMB
|
||
|
||
|
||
def test_defaults(clean_env: pytest.MonkeyPatch) -> None:
|
||
"""Значения по умолчанию классов настроек: env-файла нет, заданы только обязательные OPENBAO__* (SMB не нужен)."""
|
||
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
|
||
assert s.AUTH_DEV_ISSUER is False
|
||
assert s.AUTH_DEV_SECRET == ""
|
||
assert s.AUTH_JWKS_URL == ""
|
||
assert s.AUTH_PUBLIC_KEY == ""
|
||
assert s.AUTH_ISSUER == ""
|
||
assert s.AUTH_AUDIENCE == ""
|
||
assert s.AUTH_EMAIL_CLAIM == "email"
|
||
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.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
|
||
assert (smb.SMB_HOST, smb.SMB_PATH, smb.SMB_LOGIN, smb.SMB_PASSWORD) == ("", "", "", "")
|
||
|
||
|
||
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.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)
|
||
|
||
|
||
REQUIRED_ENV = {
|
||
"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",
|
||
}
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("key", "value"),
|
||
[
|
||
("OUTBOX_POLL_SECONDS", "0"),
|
||
("OUTBOX_POLL_SECONDS", "-1"),
|
||
("ACK_TIMEOUT_SECONDS", "0"),
|
||
("ACK_TIMEOUT_SECONDS", "-5"),
|
||
("OUTBOX_MAX_REPUBLISH", "-1"),
|
||
("UPLOAD_MAX_BYTES", "0"),
|
||
("MAPPING_POLL_SECONDS", "0"),
|
||
("MAPPING_MAX_ATTEMPTS", "0"),
|
||
],
|
||
)
|
||
def test_exchange_settings_reject_invalid(clean_env: pytest.MonkeyPatch, key: str, value: str) -> None:
|
||
"""Период опроса и ack-таймаут — строго положительные, число повторов — неотрицательное (финальное ревью M-4)."""
|
||
for name, required in REQUIRED_ENV.items():
|
||
clean_env.setenv(name, required)
|
||
clean_env.setenv(key, value)
|
||
with pytest.raises(ValidationError, match=key):
|
||
Settings(_env_file=None) # type: ignore[call-arg]
|
||
|
||
|
||
def test_zero_republish_allowed(clean_env: pytest.MonkeyPatch) -> None:
|
||
"""OUTBOX_MAX_REPUBLISH=0 допустим: без повторов, сразу dlq после ack-таймаута."""
|
||
for name, required in REQUIRED_ENV.items():
|
||
clean_env.setenv(name, required)
|
||
clean_env.setenv("OUTBOX_MAX_REPUBLISH", "0")
|
||
assert Settings(_env_file=None).OUTBOX_MAX_REPUBLISH == 0 # type: ignore[call-arg]
|
||
|
||
|
||
def _env_keys(name: str) -> set[str]:
|
||
lines = (API_DIR / name).read_text(encoding="utf-8").splitlines()
|
||
return {line.split("=", 1)[0] for line in lines if line.strip() and not line.lstrip().startswith("#")}
|
||
|
||
|
||
def test_env_example_and_env_test_have_same_keys() -> None:
|
||
"""`.env.example` и `.env.test` перечисляют одни и те же переменные: новая настройка не забывается ни в одном."""
|
||
assert _env_keys(".env.example") == _env_keys(".env.test")
|
||
|
||
|
||
def test_env_example_listens_on_loopback(clean_env: pytest.MonkeyPatch) -> None:
|
||
"""Dev-запуск по `.env.example` слушает только 127.0.0.1 (финальное ревью M-11)."""
|
||
s = Settings(_env_file=API_DIR / ".env.example") # type: ignore[call-arg]
|
||
assert (s.server.HOST, s.server.PORT) == ("127.0.0.1", 8000)
|
||
|
||
|
||
# --- CORS_ORIGINS (OQ-65, issue #9 п.2) ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_cors_origins_default_is_empty(clean_env: pytest.MonkeyPatch) -> None:
|
||
clean_env.delenv("CORS_ORIGINS", raising=False)
|
||
s = Settings(_env_file=API_DIR / ".env.test") # type: ignore[call-arg]
|
||
assert s.CORS_ORIGINS == []
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("raw", "expected"),
|
||
[
|
||
("https://a.example.ru,https://b.example.ru", ["https://a.example.ru", "https://b.example.ru"]),
|
||
(" https://a.example.ru , ", ["https://a.example.ru"]),
|
||
('["https://a.example.ru", "https://b.example.ru"]', ["https://a.example.ru", "https://b.example.ru"]),
|
||
("", []),
|
||
],
|
||
ids=["comma", "spaces-and-trailing-comma", "json", "empty"],
|
||
)
|
||
def test_cors_origins_from_env(clean_env: pytest.MonkeyPatch, raw: str, expected: list[str]) -> None:
|
||
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)
|