no message
This commit is contained in:
53
tests/config.yaml
Normal file
53
tests/config.yaml
Normal file
@@ -0,0 +1,53 @@
|
||||
# === Раздел с общими конфигурационными параметрами ===
|
||||
runtime: 5
|
||||
|
||||
# === Логирование ===
|
||||
log:
|
||||
version: 1
|
||||
disable_existing_loggers: False
|
||||
|
||||
formatters:
|
||||
standard:
|
||||
format: '%(asctime)s %(module)15s [%(levelname)8s]: %(message)s'
|
||||
telegram:
|
||||
format: '%(message)s'
|
||||
|
||||
handlers:
|
||||
console:
|
||||
level: DEBUG
|
||||
formatter: standard
|
||||
class: logging.StreamHandler
|
||||
stream: ext://sys.stdout # Default is stderr
|
||||
|
||||
file:
|
||||
level: DEBUG
|
||||
formatter: standard
|
||||
class: logging.handlers.RotatingFileHandler
|
||||
filename: logs/log.log
|
||||
mode: a
|
||||
maxBytes: 500000
|
||||
backupCount: 15
|
||||
|
||||
#telegram:
|
||||
# level: CRITICAL
|
||||
# formatter: telegram
|
||||
# class: logging_telegram_handler.TelegramHandler
|
||||
# chat_id: 211945135
|
||||
# alias: "PDC"
|
||||
|
||||
|
||||
# -- Логгеры --
|
||||
loggers:
|
||||
'':
|
||||
handlers: [console, file]
|
||||
level: INFO
|
||||
propagate: False
|
||||
|
||||
__main__:
|
||||
handlers: [console, file]
|
||||
level: DEBUG
|
||||
propagate: False
|
||||
|
||||
config_manager:
|
||||
handlers: [console, file]
|
||||
level: DEBUG
|
||||
145
tests/test.py
145
tests/test.py
@@ -1,131 +1,28 @@
|
||||
import unittest
|
||||
from unittest.mock import patch, mock_open, AsyncMock
|
||||
import asyncio
|
||||
from config_manager import ConfigManager
|
||||
import logging
|
||||
import io
|
||||
import json
|
||||
import yaml
|
||||
import asyncio
|
||||
from typing import Optional
|
||||
#import os
|
||||
#os.chdir(os.path.dirname(__file__))
|
||||
|
||||
import sys
|
||||
import os
|
||||
logger = logging.getLogger()
|
||||
|
||||
sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'src'))
|
||||
class MyApp(ConfigManager):
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.iter = 0
|
||||
|
||||
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)
|
||||
def execute(self) -> None:
|
||||
logger.info(f"current iteration {self.iter}")
|
||||
self.iter += 1
|
||||
|
||||
async def main():
|
||||
app = MyApp("config.yaml")
|
||||
app.start()
|
||||
logger.info("App started")
|
||||
await asyncio.sleep(20)
|
||||
await app.stop()
|
||||
|
||||
if __name__ == "__main__":
|
||||
logging.basicConfig(level=logging.WARNING) # отключаем логи во время тестов
|
||||
unittest.main()
|
||||
asyncio.run(main())
|
||||
|
||||
Reference in New Issue
Block a user