b6eecfcf34
- app.py: create_app(settings), SMB-сессия только при SCAN_HOTFOLDER_MODE=smb, без SPA/CORS/РАИСА/nkeys; /healthcheck отдаёт uptime и version - config.py: ENV_FILE, get_settings() с lru_cache, параметры горячих папок/NATS, блок OPENBAO__SMB необязателен, валидация режимов local/smb; убраны DBSettings и OPENBAO.DATABASE - exception_handlers.py: убрана зависимость от sqlalchemy - requirements.txt: убраны asyncpg, SQLAlchemy, alembic, argon2-cffi(+bindings), raisa_*, greenlet, Mako - api/.env.example: минимальный local-режим против compose; unit-тесты config/app Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
123 lines
4.0 KiB
Python
123 lines
4.0 KiB
Python
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):
|
|
_clean_env(monkeypatch)
|
|
monkeypatch.setenv("ENV_FILE", "/nonexistent/.env")
|
|
|
|
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 s.SCAN_POLL_SECONDS == 5
|
|
assert s.SCAN_TIMEOUT_SECONDS == 300
|
|
assert s.SCAN_MAX_INFLIGHT == 4
|
|
assert s.OPENBAO.SMB.SMB_HOST == ""
|
|
assert s.OPENBAO.NATS.NATS_SEED == ""
|
|
assert s.OPENBAO.NATS.NATS_JWT == ""
|
|
assert not hasattr(s.OPENBAO, "DATABASE")
|
|
|
|
|
|
def test_local_mode_requires_hotfolder_path(monkeypatch):
|
|
_clean_env(monkeypatch)
|
|
monkeypatch.setenv("ENV_FILE", "/nonexistent/.env")
|
|
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("ENV_FILE", "/nonexistent/.env")
|
|
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("ENV_FILE", "/nonexistent/.env")
|
|
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_unknown_hotfolder_mode_rejected(monkeypatch):
|
|
monkeypatch.setenv("ENV_FILE", "/nonexistent/.env")
|
|
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):
|
|
monkeypatch.setenv("ENV_FILE", "/nonexistent/.env")
|
|
assert get_settings() is get_settings()
|
|
assert isinstance(get_settings(), Settings)
|