"""Настройки: чтение 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 не нужен).""" 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(): 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 == "" assert s.OPENBAO.NATS.NATS_JWT == "" 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 == "" assert s.OPENBAO.NATS.NATS_JWT == "" 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