389 lines
12 KiB
Python
389 lines
12 KiB
Python
# tests/unit/test_config.py
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from logging import DEBUG, INFO
|
||
from unittest.mock import patch
|
||
import pytest
|
||
|
||
from dataloader.config import (
|
||
BaseAppSettings,
|
||
AppSettings,
|
||
LogSettings,
|
||
PGSettings,
|
||
WorkerSettings,
|
||
Secrets,
|
||
)
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestBaseAppSettings:
|
||
"""
|
||
Unit тесты для BaseAppSettings.
|
||
"""
|
||
|
||
def test_default_values(self):
|
||
"""
|
||
Тест дефолтных значений.
|
||
"""
|
||
settings = BaseAppSettings()
|
||
|
||
assert settings.local is False
|
||
assert settings.debug is False
|
||
|
||
def test_protocol_returns_http_when_not_local(self):
|
||
"""
|
||
Тест, что protocol возвращает http для не-локального режима.
|
||
"""
|
||
with patch.dict("os.environ", {"LOCAL": "false"}):
|
||
settings = BaseAppSettings()
|
||
|
||
assert settings.protocol == "http"
|
||
|
||
def test_protocol_returns_https_when_local(self):
|
||
"""
|
||
Тест, что protocol возвращает https для локального режима.
|
||
"""
|
||
with patch.dict("os.environ", {"LOCAL": "true"}):
|
||
settings = BaseAppSettings()
|
||
|
||
assert settings.protocol == "https"
|
||
|
||
def test_loads_from_env(self):
|
||
"""
|
||
Тест загрузки из переменных окружения.
|
||
"""
|
||
with patch.dict("os.environ", {"LOCAL": "true", "DEBUG": "true"}):
|
||
settings = BaseAppSettings()
|
||
|
||
assert settings.local is True
|
||
assert settings.debug is True
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestAppSettings:
|
||
"""
|
||
Unit тесты для AppSettings.
|
||
"""
|
||
|
||
def test_default_values(self):
|
||
"""
|
||
Тест дефолтных значений.
|
||
"""
|
||
settings = AppSettings()
|
||
|
||
assert settings.app_host == "0.0.0.0"
|
||
assert settings.app_port == 8081
|
||
assert settings.kube_net_name == "AIGATEWAY"
|
||
assert settings.timezone == "Europe/Moscow"
|
||
|
||
def test_loads_from_env(self):
|
||
"""
|
||
Тест загрузки из переменных окружения.
|
||
"""
|
||
with patch.dict("os.environ", {
|
||
"APP_HOST": "127.0.0.1",
|
||
"APP_PORT": "9000",
|
||
"PROJECT_NAME": "TestProject",
|
||
"TIMEZONE": "UTC"
|
||
}):
|
||
settings = AppSettings()
|
||
|
||
assert settings.app_host == "127.0.0.1"
|
||
assert settings.app_port == 9000
|
||
assert settings.kube_net_name == "TestProject"
|
||
assert settings.timezone == "UTC"
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestLogSettings:
|
||
"""
|
||
Unit тесты для LogSettings.
|
||
"""
|
||
|
||
def test_default_values(self):
|
||
"""
|
||
Тест дефолтных значений.
|
||
"""
|
||
settings = LogSettings()
|
||
|
||
assert settings.private_log_file_name == "app.log"
|
||
assert settings.log_rotation == "10 MB"
|
||
assert settings.private_metric_file_name == "app-metric.log"
|
||
assert settings.private_audit_file_name == "events.log"
|
||
assert settings.audit_host_ip == "127.0.0.1"
|
||
|
||
def test_get_file_abs_path_joins_path_and_file(self):
|
||
"""
|
||
Тест объединения пути и имени файла.
|
||
"""
|
||
path = LogSettings.get_file_abs_path("/var/log/", "app.log")
|
||
|
||
assert "app.log" in path
|
||
assert path.startswith("var")
|
||
|
||
def test_get_file_abs_path_handles_trailing_slashes(self):
|
||
"""
|
||
Тест обработки слэшей в путях.
|
||
"""
|
||
path = LogSettings.get_file_abs_path("/var/log///", "///app.log")
|
||
|
||
assert "app.log" in path
|
||
assert path.startswith("var")
|
||
assert not path.startswith("/")
|
||
|
||
def test_log_file_abs_path_property(self):
|
||
"""
|
||
Тест свойства log_file_abs_path.
|
||
"""
|
||
with patch.dict("os.environ", {"LOG_PATH": "/var/log", "LOG_FILE_NAME": "test.log"}):
|
||
settings = LogSettings()
|
||
|
||
assert "test.log" in settings.log_file_abs_path
|
||
assert settings.log_file_abs_path.startswith("var")
|
||
|
||
def test_metric_file_abs_path_property(self):
|
||
"""
|
||
Тест свойства metric_file_abs_path.
|
||
"""
|
||
with patch.dict("os.environ", {"METRIC_PATH": "/var/metrics", "METRIC_FILE_NAME": "metrics.log"}):
|
||
settings = LogSettings()
|
||
|
||
assert "metrics.log" in settings.metric_file_abs_path
|
||
assert settings.metric_file_abs_path.startswith("var")
|
||
|
||
def test_audit_file_abs_path_property(self):
|
||
"""
|
||
Тест свойства audit_file_abs_path.
|
||
"""
|
||
with patch.dict("os.environ", {"AUDIT_LOG_PATH": "/var/audit", "AUDIT_LOG_FILE_NAME": "audit.log"}):
|
||
settings = LogSettings()
|
||
|
||
assert "audit.log" in settings.audit_file_abs_path
|
||
assert settings.audit_file_abs_path.startswith("var")
|
||
|
||
def test_log_lvl_returns_debug_when_debug_enabled(self):
|
||
"""
|
||
Тест, что log_lvl возвращает DEBUG в debug-режиме.
|
||
"""
|
||
with patch.dict("os.environ", {"DEBUG": "true"}):
|
||
settings = LogSettings()
|
||
|
||
assert settings.log_lvl == DEBUG
|
||
|
||
def test_log_lvl_returns_info_when_debug_disabled(self):
|
||
"""
|
||
Тест, что log_lvl возвращает INFO в обычном режиме.
|
||
"""
|
||
with patch.dict("os.environ", {"DEBUG": "false"}):
|
||
settings = LogSettings()
|
||
|
||
assert settings.log_lvl == INFO
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestPGSettings:
|
||
"""
|
||
Unit тесты для PGSettings.
|
||
"""
|
||
|
||
def test_default_values(self):
|
||
"""
|
||
Тест дефолтных значений.
|
||
"""
|
||
with patch.dict("os.environ", {}, clear=True):
|
||
settings = PGSettings()
|
||
|
||
assert settings.host == "localhost"
|
||
assert settings.port == 5432
|
||
assert settings.user == "postgres"
|
||
assert settings.password == ""
|
||
assert settings.database == "postgres"
|
||
assert settings.schema_queue == "public"
|
||
assert settings.use_pool is True
|
||
assert settings.pool_size == 5
|
||
assert settings.max_overflow == 10
|
||
|
||
def test_url_property_returns_connection_string(self):
|
||
"""
|
||
Тест формирования строки подключения.
|
||
"""
|
||
with patch.dict("os.environ", {
|
||
"PG_HOST": "db.example.com",
|
||
"PG_PORT": "5433",
|
||
"PG_USER": "testuser",
|
||
"PG_PASSWORD": "testpass",
|
||
"PG_DATABASE": "testdb"
|
||
}):
|
||
settings = PGSettings()
|
||
|
||
expected = "postgresql+asyncpg://testuser:testpass@db.example.com:5433/testdb"
|
||
assert settings.url == expected
|
||
|
||
def test_url_property_with_empty_password(self):
|
||
"""
|
||
Тест строки подключения с пустым паролем.
|
||
"""
|
||
with patch.dict("os.environ", {
|
||
"PG_HOST": "localhost",
|
||
"PG_PORT": "5432",
|
||
"PG_USER": "postgres",
|
||
"PG_PASSWORD": "",
|
||
"PG_DATABASE": "testdb"
|
||
}):
|
||
settings = PGSettings()
|
||
|
||
expected = "postgresql+asyncpg://postgres:@localhost:5432/testdb"
|
||
assert settings.url == expected
|
||
|
||
def test_loads_from_env(self):
|
||
"""
|
||
Тест загрузки из переменных окружения.
|
||
"""
|
||
with patch.dict("os.environ", {
|
||
"PG_HOST": "testhost",
|
||
"PG_PORT": "5433",
|
||
"PG_USER": "testuser",
|
||
"PG_PASSWORD": "testpass",
|
||
"PG_DATABASE": "testdb",
|
||
"PG_SCHEMA_QUEUE": "queue_schema",
|
||
"PG_POOL_SIZE": "20"
|
||
}):
|
||
settings = PGSettings()
|
||
|
||
assert settings.host == "testhost"
|
||
assert settings.port == 5433
|
||
assert settings.user == "testuser"
|
||
assert settings.password == "testpass"
|
||
assert settings.database == "testdb"
|
||
assert settings.schema_queue == "queue_schema"
|
||
assert settings.pool_size == 20
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestWorkerSettings:
|
||
"""
|
||
Unit тесты для WorkerSettings.
|
||
"""
|
||
|
||
def test_default_values(self):
|
||
"""
|
||
Тест дефолтных значений.
|
||
"""
|
||
with patch.dict("os.environ", {"WORKERS_JSON": "[]"}, clear=True):
|
||
settings = WorkerSettings()
|
||
|
||
assert settings.workers_json == "[]"
|
||
assert settings.heartbeat_sec == 10
|
||
assert settings.default_lease_ttl_sec == 60
|
||
assert settings.reaper_period_sec == 10
|
||
assert settings.claim_backoff_sec == 15
|
||
|
||
def test_parsed_workers_returns_empty_list_for_default(self):
|
||
"""
|
||
Тест, что parsed_workers возвращает пустой список по умолчанию.
|
||
"""
|
||
with patch.dict("os.environ", {"WORKERS_JSON": "[]"}):
|
||
settings = WorkerSettings()
|
||
|
||
assert settings.parsed_workers() == []
|
||
|
||
def test_parsed_workers_parses_valid_json(self):
|
||
"""
|
||
Тест парсинга валидного JSON.
|
||
"""
|
||
workers_json = json.dumps([
|
||
{"queue": "queue1", "concurrency": 2},
|
||
{"queue": "queue2", "concurrency": 3}
|
||
])
|
||
with patch.dict("os.environ", {"WORKERS_JSON": workers_json}):
|
||
settings = WorkerSettings()
|
||
|
||
workers = settings.parsed_workers()
|
||
|
||
assert len(workers) == 2
|
||
assert workers[0]["queue"] == "queue1"
|
||
assert workers[0]["concurrency"] == 2
|
||
assert workers[1]["queue"] == "queue2"
|
||
assert workers[1]["concurrency"] == 3
|
||
|
||
def test_parsed_workers_filters_non_dict_items(self):
|
||
"""
|
||
Тест фильтрации не-словарей из JSON.
|
||
"""
|
||
workers_json = json.dumps([
|
||
{"queue": "queue1", "concurrency": 2},
|
||
"invalid_item",
|
||
123,
|
||
{"queue": "queue2", "concurrency": 3}
|
||
])
|
||
with patch.dict("os.environ", {"WORKERS_JSON": workers_json}):
|
||
settings = WorkerSettings()
|
||
|
||
workers = settings.parsed_workers()
|
||
|
||
assert len(workers) == 2
|
||
assert all(isinstance(w, dict) for w in workers)
|
||
|
||
def test_parsed_workers_handles_invalid_json(self):
|
||
"""
|
||
Тест обработки невалидного JSON.
|
||
"""
|
||
with patch.dict("os.environ", {"WORKERS_JSON": "not valid json"}):
|
||
settings = WorkerSettings()
|
||
|
||
workers = settings.parsed_workers()
|
||
|
||
assert workers == []
|
||
|
||
def test_parsed_workers_handles_empty_string(self):
|
||
"""
|
||
Тест обработки пустой строки.
|
||
"""
|
||
with patch.dict("os.environ", {"WORKERS_JSON": ""}):
|
||
settings = WorkerSettings()
|
||
|
||
workers = settings.parsed_workers()
|
||
|
||
assert workers == []
|
||
|
||
def test_parsed_workers_handles_null_json(self):
|
||
"""
|
||
Тест обработки null в JSON.
|
||
"""
|
||
with patch.dict("os.environ", {"WORKERS_JSON": "null"}):
|
||
settings = WorkerSettings()
|
||
|
||
workers = settings.parsed_workers()
|
||
|
||
assert workers == []
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestSecrets:
|
||
"""
|
||
Unit тесты для Secrets.
|
||
"""
|
||
|
||
def test_initializes_all_settings(self):
|
||
"""
|
||
Тест, что Secrets инициализирует все настройки.
|
||
"""
|
||
secrets = Secrets()
|
||
|
||
assert isinstance(secrets.app, AppSettings)
|
||
assert isinstance(secrets.log, LogSettings)
|
||
assert isinstance(secrets.pg, PGSettings)
|
||
assert isinstance(secrets.worker, WorkerSettings)
|
||
|
||
def test_all_settings_have_default_values(self):
|
||
"""
|
||
Тест, что все настройки имеют дефолтные значения.
|
||
"""
|
||
secrets = Secrets()
|
||
|
||
assert secrets.app.app_host == "0.0.0.0"
|
||
assert secrets.log.private_log_file_name == "app.log"
|
||
assert secrets.pg.host == "localhost"
|
||
assert secrets.worker.heartbeat_sec == 10
|