mirror of
https://github.com/outbackdingo/patroni.git
synced 2026-01-27 18:20:05 +00:00
c1bfb0e6d6
- get rid from 2.7 specific modules: `six`, `ipaddress` - use Python3 unpacking operator - use `shutil.which()` instead of `find_executable()`
75 lines
2.6 KiB
Python
75 lines
2.6 KiB
Python
import logging
|
|
import os
|
|
import sys
|
|
import unittest
|
|
import yaml
|
|
|
|
from mock import Mock, patch
|
|
from patroni.config import Config
|
|
from patroni.log import PatroniLogger
|
|
from queue import Queue, Full
|
|
|
|
_LOG = logging.getLogger(__name__)
|
|
|
|
|
|
class TestPatroniLogger(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self._handlers = logging.getLogger().handlers[:]
|
|
|
|
def tearDown(self):
|
|
logging.getLogger().handlers[:] = self._handlers
|
|
|
|
@patch('logging.FileHandler._open', Mock())
|
|
def test_patroni_logger(self):
|
|
config = {
|
|
'log': {
|
|
'traceback_level': 'DEBUG',
|
|
'max_queue_size': 5,
|
|
'dir': 'foo',
|
|
'file_size': 4096,
|
|
'file_num': 5,
|
|
'loggers': {
|
|
'foo.bar': 'INFO'
|
|
}
|
|
},
|
|
'restapi': {}, 'postgresql': {'data_dir': 'foo'}
|
|
}
|
|
sys.argv = ['patroni.py']
|
|
os.environ[Config.PATRONI_CONFIG_VARIABLE] = yaml.dump(config, default_flow_style=False)
|
|
logger = PatroniLogger()
|
|
patroni_config = Config(None)
|
|
logger.reload_config(patroni_config['log'])
|
|
_LOG.exception('test')
|
|
logger.start()
|
|
|
|
with patch.object(logging.Handler, 'format', Mock(side_effect=Exception)),\
|
|
patch('_pytest.logging.LogCaptureHandler.emit', Mock()):
|
|
logging.error('test')
|
|
|
|
self.assertEqual(logger.log_handler.maxBytes, config['log']['file_size'])
|
|
self.assertEqual(logger.log_handler.backupCount, config['log']['file_num'])
|
|
|
|
config['log']['level'] = 'DEBUG'
|
|
config['log'].pop('dir')
|
|
with patch('logging.Handler.close', Mock(side_effect=Exception)):
|
|
logger.reload_config(config['log'])
|
|
with patch.object(logging.Logger, 'makeRecord',
|
|
Mock(side_effect=[logging.LogRecord('', logging.INFO, '', 0, '', (), None), Exception])):
|
|
logging.exception('test')
|
|
logging.error('test')
|
|
with patch.object(Queue, 'put_nowait', Mock(side_effect=Full)):
|
|
self.assertRaises(SystemExit, logger.shutdown)
|
|
self.assertRaises(Exception, logger.shutdown)
|
|
self.assertLessEqual(logger.queue_size, 2) # "Failed to close the old log handler" could be still in the queue
|
|
self.assertEqual(logger.records_lost, 0)
|
|
|
|
def test_interceptor(self):
|
|
logger = PatroniLogger()
|
|
logger.reload_config({'level': 'INFO'})
|
|
logger.start()
|
|
_LOG.info('Lock owner: ')
|
|
_LOG.info('blabla')
|
|
logger.shutdown()
|
|
self.assertEqual(logger.records_lost, 0)
|