132 lines
4.1 KiB
Python
132 lines
4.1 KiB
Python
import unittest
|
|
from unittest.mock import patch, mock_open, AsyncMock
|
|
import asyncio
|
|
import logging
|
|
import io
|
|
import json
|
|
import yaml
|
|
|
|
import sys
|
|
import os
|
|
|
|
sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'src'))
|
|
|
|
from basic_application.basic_application import ConfigManager
|
|
|
|
|
|
class TestConfigManager(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self):
|
|
self.json_data = json.dumps({
|
|
"work_interval": 1,
|
|
"update_interval": 1,
|
|
"logging": {
|
|
"version": 1,
|
|
"handlers": {"console": {"class": "logging.StreamHandler", "level": "DEBUG"}},
|
|
"root": {"handlers": ["console"], "level": "DEBUG"}
|
|
},
|
|
"some_key": "some_value"
|
|
})
|
|
self.yaml_data = """
|
|
work_interval: 1
|
|
update_interval: 1
|
|
logging:
|
|
version: 1
|
|
handlers:
|
|
console:
|
|
class: logging.StreamHandler
|
|
level: DEBUG
|
|
root:
|
|
handlers: [console]
|
|
level: DEBUG
|
|
some_key: some_value
|
|
"""
|
|
|
|
@patch("builtins.open", new_callable=mock_open, read_data="")
|
|
async def test_read_file_async_json(self, mock_file):
|
|
mock_file.return_value.read = lambda: self.json_data
|
|
cm = ConfigManager("config.json")
|
|
content = await cm._read_file_async()
|
|
self.assertEqual(content, self.json_data)
|
|
|
|
@patch("builtins.open", new_callable=mock_open, read_data="")
|
|
async def test_read_file_async_yaml(self, mock_file):
|
|
mock_file.return_value.read = lambda: self.yaml_data
|
|
cm = ConfigManager("config.yaml")
|
|
content = await cm._read_file_async()
|
|
self.assertEqual(content, self.yaml_data)
|
|
|
|
def test_parse_json(self):
|
|
cm = ConfigManager("config.json")
|
|
parsed = cm._parse_config(self.json_data)
|
|
self.assertIsInstance(parsed, dict)
|
|
self.assertEqual(parsed["some_key"], "some_value")
|
|
|
|
def test_parse_yaml(self):
|
|
cm = ConfigManager("config.yaml")
|
|
parsed = cm._parse_config(self.yaml_data)
|
|
self.assertIsInstance(parsed, dict)
|
|
self.assertEqual(parsed["some_key"], "some_value")
|
|
|
|
@patch("basic_application.basic_application.logging.config.dictConfig")
|
|
def test_apply_logging_config(self, mock_dict_config):
|
|
cm = ConfigManager("config.json")
|
|
cm._apply_logging_config({"logging": {"version": 1}})
|
|
mock_dict_config.assert_called_once()
|
|
|
|
async def test_update_config_changes_config_and_intervals(self):
|
|
# Мокаем чтение файла
|
|
m = mock_open(read_data=self.json_data)
|
|
with patch("builtins.open", m):
|
|
cm = ConfigManager("config.json")
|
|
|
|
# Проверяем исходные интервалы
|
|
self.assertEqual(cm.update_interval, cm.DEFAULT_UPDATE_INTERVAL)
|
|
self.assertEqual(cm.work_interval, cm.DEFAULT_WORK_INTERVAL)
|
|
|
|
await cm._update_config()
|
|
|
|
# После обновления данные заполнены
|
|
self.assertIsInstance(cm.config, dict)
|
|
self.assertEqual(cm.update_interval, 1.0)
|
|
self.assertEqual(cm.work_interval, 1.0)
|
|
|
|
async def test_execute_called_in_worker_loop(self):
|
|
called = False
|
|
|
|
class TestCM(ConfigManager):
|
|
def execute(self2):
|
|
nonlocal called
|
|
called = True
|
|
|
|
cm = TestCM("config.json")
|
|
|
|
async def stop_after_delay():
|
|
await asyncio.sleep(0.1)
|
|
cm.stop()
|
|
|
|
# Запускаем worker_loop и через 0.1 сек останавливаем
|
|
await asyncio.gather(cm._worker_loop(), stop_after_delay())
|
|
|
|
self.assertTrue(called)
|
|
|
|
async def test_periodic_update_loop_runs(self):
|
|
count = 0
|
|
|
|
class TestCM(ConfigManager):
|
|
async def _update_config(self2):
|
|
nonlocal count
|
|
count += 1
|
|
if count >= 2:
|
|
self2.stop()
|
|
|
|
cm = TestCM("config.json")
|
|
|
|
await cm._periodic_update_loop()
|
|
|
|
self.assertGreaterEqual(count, 2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
logging.basicConfig(level=logging.WARNING) # отключаем логи во время тестов
|
|
unittest.main()
|