tests.unit.test_core.test_logs.TestNAppLog.setup_method() - Code Metrics - Inspection of "Merge pull request #449 from kytos-ng/chore/pytest" - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 74a04b...775f07 )

by Vinicius

created 2024-02-08 12:50 UTC

TestNAppLog.setup_method() A

↳ Parent: tests.unit.test_core.test_logs

Complexity

Conditions

Size

Total Lines	4
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	3
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	3
nop	1
dl	0
loc	4
ccs	3
cts	3
cp	1
crap	1
rs	10
c	0
b	0
f	0

"""Test the logs module."""
import importlib
import logging
from copy import copy
from inspect import FrameInfo
from unittest.mock import Mock, patch

from kytos.core import logs
from kytos.core.logs import LogManager, NAppLog


class LogTester:
    """Common code to test loggers."""

    def setup_method(self):
        """Backup original handlers."""
        self._patcher = None
        self._logger = None
        # If we don't copy, it'll be a reference to a list that will be changed
        self._handlers_bak = copy(logging.getLogger().handlers)

    def teardown_method(self):
        """Undo mocking, restore handlers."""
        if self._patcher:
            self._patcher.stop()
        logging.getLogger().handlers = self._handlers_bak

    def _mock_logger(self):
        """Mock kytos.core.log and assign to self._logger.

        Calling this function is required for ``_has_string_in_log``.
        """
        self._patcher = patch('kytos.core.logs.LOG')
        self._logger = self._patcher.start()

    def _assert_string_in_logs(self, string):
        """Assert _string_ is in any message sent to the root logger.

        Search through all log messages since ``_mock_logger`` was called.
        """
        msgs = [call[1][0] for call in self._logger.mock_calls]
        assert any(string in msg for msg in msgs), \
               f'Message "{string}" not in {msgs}.'


class TestLogManager(LogTester):
    """Test LogManager class."""

    def test_add_handler_to_root(self):
        """Handler should be added to root logger."""
        handler = Mock()
        LogManager.add_handler(handler)
        assert handler in logging.root.handlers

    @staticmethod
    @patch('kytos.core.logs.LogManager._PARSER')
    @patch('kytos.core.logs.Path')
    def test_custom_formatter(path, parser):
        """Should use a custom formatter instead of Python's default."""
        path.return_value.exists.return_value = False
        # Make sure we have the custome formatter section
        parser.__contains__.return_value = True

        # Make 'parser' behave as a dict, this is necessary because the _PARSER
        # is being patched returning a MagicMock (inside add_handler) and in
        # 'logging' module (python 3.8) the 'Formatter' class includes a
        # 'validate' method that breaks when receives a MagicMock object.
        format_dict = {'formatter_console': {'format': None}}
        parser.__getitem__.side_effect = format_dict.__getitem__

        handler = Mock()

        LogManager.add_handler(handler)
        handler.setFormatter.assert_called_once()

    def test_add_websocket(self):
        """A stream should be used in a handler added to the root logger."""
        socket = Mock()
        handler = LogManager.enable_websocket(socket)
        assert handler in logging.root.handlers

    @staticmethod
    def test_parent_handler_usage():
        """Existent logger should use a new handler."""
        # First, get a child logger.
        old_logger = logging.getLogger('non-root logger')
        # Then, add a new handler.
        new_handler = Mock(level=logging.DEBUG)
        LogManager.add_handler(new_handler)

        old_logger.setLevel(logging.DEBUG)
        old_logger.debug('Handler should receive me.')
        new_handler.handle.assert_called_once()

    @patch('kytos.core.logs.Path')
    def test_non_existent_config_file(self, path):
        """If config file doesn't exist, warn instead of raising exception."""
        self._mock_logger()
        path.return_value.exists.return_value = False
        LogManager.load_config_file('non_existent_file')
        self._assert_string_in_logs('Log config file "%s" does not exist.')

    @patch.object(LogManager, '_PARSER')
    @patch('kytos.core.logs.config')
    @patch('kytos.core.logs.Path')
    def test_no_syslog(self, path, config, parser):
        """Must log when there's no syslog and try again without it."""
        path.return_value.exists.return_value = True
        config.fileConfig.side_effect = [OSError, None]
        parser.__contains__.return_value = True  # must have syslog section
        self._mock_logger()

        LogManager.load_config_file('existent_file')
        self._assert_string_in_logs('Trying to disable syslog')
        parser.remove_section.assert_called_once_with('handler_syslog')
        self._assert_string_in_logs('Logging config file "%s" loaded '
                                    'successfully.')

    @patch('kytos.core.logs.Path')
    @patch('kytos.core.logs.config')
    def test_config_file_exception(self, config, path):
        """Test other errors (e.g. /dev/log permission)."""
        path.return_value.exists.return_value = True
        config.fileConfig.side_effect = OSError

        self._mock_logger()
        LogManager.load_config_file('existent_file')
        self._assert_string_in_logs('Using default Python logging config')

    @patch.object(LogManager, '_PARSER')
    @patch('kytos.core.logs.Path')
    @patch('kytos.core.logs.config')
    def test_set_debug_mode(self, config, path, parser):
        """Test set_debug_mode with debug = True."""
        path.return_value.exists.return_value = True
        config.fileConfig.side_effect = OSError

        self._mock_logger()
        LogManager.load_config_file('existent_file', debug=True)

        expected_message = 'Setting log configuration with debug mode.'
        self._assert_string_in_logs(expected_message)

        parser.set('logger_root', 'level', 'DEBUG')
        parser.set.assert_called_with('logger_root', 'level', 'DEBUG')

        parser.set('logger_api_server', 'level', 'DEBUG')
        parser.set.assert_called_with('logger_api_server', 'level', 'DEBUG')

    @patch('kytos.core.logs.LogManager._PARSER')
    @patch('kytos.core.logs.Path')
    @patch('kytos.core.logs.config')
    def test_set_debug_mode_with_false(self, config, path, parser):
        """Test set_debug_mode with debug = False."""
        path.return_value.exists.return_value = True
        config.fileConfig.side_effect = OSError

        self._mock_logger()
        LogManager.load_config_file('existent_file', debug=False)

        parser.set.assert_not_called()

    def test_handler_filter(self):
        """Should not log harmless werkzeug "session is disconnected" msg."""
        logging.root.handlers = []

        handler = Mock(level=logging.WARNING)
        LogManager.add_handler(handler)

        # Message based on the log output that ends with traceback plaintext as
        # seen in site-packages/werkzeug/serving.py:225 of Werkzeug==0.12.1
        msg = "lorem ipsum KeyError: 'Session is disconnected'"
        logger = logging.getLogger('werkzeug')
        logger.setLevel(logging.ERROR)
        with patch.object(handler, 'emit'):
            logger.error('lorem ipsum %s', msg)
            assert 0 == handler.emit.call_count

    @staticmethod
    def test_old_handler_filter():
        """Should not log harmless werkzeug "session is disconnected" msg.

        The filter should be added to all root handlers, even the ones that
        already existed before importing the "logs" module.

        Message based on the log output that ends with traceback plaintext as
        seen in site-packages/werkzeug/serving.py:225 of Werkzeug==0.12.1:

            - logger name: werkzeug
            - level: ERROR
            - only argument: ends with "KeyError: 'Session is disconnected'"
        """
        old_handler = Mock()
        old_handler.filters = []

        logging.root.addHandler(old_handler)
        old_handler.addFilter.assert_not_called()
        # Importing the module should add the filter to existent root handlers.
        importlib.reload(logs)
        old_handler.addFilter.assert_called_once_with(
            logs.LogManager.filter_session_disconnected)

        logging.root.removeHandler(old_handler)

    def test_decorate_logger_class(self):
        """Check that setting the decorator works correctly"""
        class DummyWrapper(logging.Logger):
            """Dummy class for comparison"""
            pass

        LogManager.decorate_logger_class(lambda arg: DummyWrapper)

        assert logging.Logger.manager.loggerClass == DummyWrapper

        assert isinstance(logging.root, DummyWrapper)


class TestNAppLog(LogTester):
    """Test the log used by NApps."""

    def setup_method(self):
        """Initialize variables."""
        super().setup_method()
        self._inspect_patcher = None

    def teardown_method(self):
        """Undo mocking."""
        super().teardown_method()
        if self._inspect_patcher:
            self._inspect_patcher.stop()

    def _set_filename(self, filename):
        """Mock the NApp's main.py file path."""
        # Put the filename in the call stack
        frame = FrameInfo(None, filename, None, None, None, None)
        self._inspect_patcher = patch('kytos.core.logs.inspect')
        inspect = self._inspect_patcher.start()
        inspect.stack.return_value = [frame]

    def test_napp_id_detection(self):
        """Test NApp ID detection based on filename."""
        self._set_filename('/napps/username/name/main.py')
        expected_logger_name = 'kytos.napps.username/name'
        napp_logger = NAppLog()
        assert expected_logger_name == napp_logger.name

    def test_napp_id_not_found(self):
        """If NApp ID is not found, should use root logger."""
        self._set_filename('not/an/expected/NApp/path.py')
        root_logger = logging.getLogger("kytos.napps")
        napp_logger = NAppLog()
        assert root_logger.name == napp_logger.name


1		"""Test the logs module."""
2	1	import importlib
3	1	import logging
4	1	from copy import copy
5	1	from inspect import FrameInfo
6	1	from unittest.mock import Mock, patch
7
8	1	from kytos.core import logs
9	1	from kytos.core.logs import LogManager, NAppLog
10
11
12	1	class LogTester:
13		"""Common code to test loggers."""
14
15	1	def setup_method(self):
16		"""Backup original handlers."""
17	1	self._patcher = None
18	1	self._logger = None
19		# If we don't copy, it'll be a reference to a list that will be changed
20	1	self._handlers_bak = copy(logging.getLogger().handlers)
21
22	1	def teardown_method(self):
23		"""Undo mocking, restore handlers."""
24	1	if self._patcher:
25	1	self._patcher.stop()
26	1	logging.getLogger().handlers = self._handlers_bak
27
28	1	def _mock_logger(self):
29		"""Mock kytos.core.log and assign to self._logger.
30
31		Calling this function is required for ``_has_string_in_log``.
32		"""
33	1	self._patcher = patch('kytos.core.logs.LOG')
34	1	self._logger = self._patcher.start()
35
36	1	def _assert_string_in_logs(self, string):
37		"""Assert _string_ is in any message sent to the root logger.
38
39		Search through all log messages since ``_mock_logger`` was called.
40		"""
41	1	msgs = [call[1][0] for call in self._logger.mock_calls]
42	1	assert any(string in msg for msg in msgs), \
43		f'Message "{string}" not in {msgs}.'
44
45
46	1	class TestLogManager(LogTester):
47		"""Test LogManager class."""
48
49	1	def test_add_handler_to_root(self):
50		"""Handler should be added to root logger."""
51	1	handler = Mock()
52	1	LogManager.add_handler(handler)
53	1	assert handler in logging.root.handlers
54
55	1	@staticmethod
56	1	@patch('kytos.core.logs.LogManager._PARSER')
57	1	@patch('kytos.core.logs.Path')
58	1	def test_custom_formatter(path, parser):
59		"""Should use a custom formatter instead of Python's default."""
60	1	path.return_value.exists.return_value = False
61		# Make sure we have the custome formatter section
62	1	parser.__contains__.return_value = True
63
64		# Make 'parser' behave as a dict, this is necessary because the _PARSER
65		# is being patched returning a MagicMock (inside add_handler) and in
66		# 'logging' module (python 3.8) the 'Formatter' class includes a
67		# 'validate' method that breaks when receives a MagicMock object.
68	1	format_dict = {'formatter_console': {'format': None}}
69	1	parser.__getitem__.side_effect = format_dict.__getitem__
70
71	1	handler = Mock()
72
73	1	LogManager.add_handler(handler)
74	1	handler.setFormatter.assert_called_once()
75
76	1	def test_add_websocket(self):
77		"""A stream should be used in a handler added to the root logger."""
78	1	socket = Mock()
79	1	handler = LogManager.enable_websocket(socket)
80	1	assert handler in logging.root.handlers
81
82	1	@staticmethod
83	1	def test_parent_handler_usage():
84		"""Existent logger should use a new handler."""
85		# First, get a child logger.
86	1	old_logger = logging.getLogger('non-root logger')
87		# Then, add a new handler.
88	1	new_handler = Mock(level=logging.DEBUG)
89	1	LogManager.add_handler(new_handler)
90
91	1	old_logger.setLevel(logging.DEBUG)
92	1	old_logger.debug('Handler should receive me.')
93	1	new_handler.handle.assert_called_once()
94
95	1	@patch('kytos.core.logs.Path')
96	1	def test_non_existent_config_file(self, path):
97		"""If config file doesn't exist, warn instead of raising exception."""
98	1	self._mock_logger()
99	1	path.return_value.exists.return_value = False
100	1	LogManager.load_config_file('non_existent_file')
101	1	self._assert_string_in_logs('Log config file "%s" does not exist.')
102
103	1	@patch.object(LogManager, '_PARSER')
104	1	@patch('kytos.core.logs.config')
105	1	@patch('kytos.core.logs.Path')
106	1	def test_no_syslog(self, path, config, parser):
107		"""Must log when there's no syslog and try again without it."""
108	1	path.return_value.exists.return_value = True
109	1	config.fileConfig.side_effect = [OSError, None]
110	1	parser.__contains__.return_value = True # must have syslog section
111	1	self._mock_logger()
112
113	1	LogManager.load_config_file('existent_file')
114	1	self._assert_string_in_logs('Trying to disable syslog')
115	1	parser.remove_section.assert_called_once_with('handler_syslog')
116	1	self._assert_string_in_logs('Logging config file "%s" loaded '
117		'successfully.')
118
119	1	@patch('kytos.core.logs.Path')
120	1	@patch('kytos.core.logs.config')
121	1	def test_config_file_exception(self, config, path):
122		"""Test other errors (e.g. /dev/log permission)."""
123	1	path.return_value.exists.return_value = True
124	1	config.fileConfig.side_effect = OSError
125
126	1	self._mock_logger()
127	1	LogManager.load_config_file('existent_file')
128	1	self._assert_string_in_logs('Using default Python logging config')
129
130	1	@patch.object(LogManager, '_PARSER')
131	1	@patch('kytos.core.logs.Path')
132	1	@patch('kytos.core.logs.config')
133	1	def test_set_debug_mode(self, config, path, parser):
134		"""Test set_debug_mode with debug = True."""
135	1	path.return_value.exists.return_value = True
136	1	config.fileConfig.side_effect = OSError
137
138	1	self._mock_logger()
139	1	LogManager.load_config_file('existent_file', debug=True)
140
141	1	expected_message = 'Setting log configuration with debug mode.'
142	1	self._assert_string_in_logs(expected_message)
143
144	1	parser.set('logger_root', 'level', 'DEBUG')
145	1	parser.set.assert_called_with('logger_root', 'level', 'DEBUG')
146
147	1	parser.set('logger_api_server', 'level', 'DEBUG')
148	1	parser.set.assert_called_with('logger_api_server', 'level', 'DEBUG')
149
150	1	@patch('kytos.core.logs.LogManager._PARSER')
151	1	@patch('kytos.core.logs.Path')
152	1	@patch('kytos.core.logs.config')
153	1	def test_set_debug_mode_with_false(self, config, path, parser):
154		"""Test set_debug_mode with debug = False."""
155	1	path.return_value.exists.return_value = True
156	1	config.fileConfig.side_effect = OSError
157
158	1	self._mock_logger()
159	1	LogManager.load_config_file('existent_file', debug=False)
160
161	1	parser.set.assert_not_called()
162
163	1	def test_handler_filter(self):
164		"""Should not log harmless werkzeug "session is disconnected" msg."""
165	1	logging.root.handlers = []
166
167	1	handler = Mock(level=logging.WARNING)
168	1	LogManager.add_handler(handler)
169
170		# Message based on the log output that ends with traceback plaintext as
171		# seen in site-packages/werkzeug/serving.py:225 of Werkzeug==0.12.1
172	1	msg = "lorem ipsum KeyError: 'Session is disconnected'"
173	1	logger = logging.getLogger('werkzeug')
174	1	logger.setLevel(logging.ERROR)
175	1	with patch.object(handler, 'emit'):
176	1	logger.error('lorem ipsum %s', msg)
177	1	assert 0 == handler.emit.call_count
178
179	1	@staticmethod
180	1	def test_old_handler_filter():
181		"""Should not log harmless werkzeug "session is disconnected" msg.
182
183		The filter should be added to all root handlers, even the ones that
184		already existed before importing the "logs" module.
185
186		Message based on the log output that ends with traceback plaintext as
187		seen in site-packages/werkzeug/serving.py:225 of Werkzeug==0.12.1:
188
189		- logger name: werkzeug
190		- level: ERROR
191		- only argument: ends with "KeyError: 'Session is disconnected'"
192		"""
193	1	old_handler = Mock()
194	1	old_handler.filters = []
195
196	1	logging.root.addHandler(old_handler)
197	1	old_handler.addFilter.assert_not_called()
198		# Importing the module should add the filter to existent root handlers.
199	1	importlib.reload(logs)
200	1	old_handler.addFilter.assert_called_once_with(
201		logs.LogManager.filter_session_disconnected)
202
203	1	logging.root.removeHandler(old_handler)
204
205	1	def test_decorate_logger_class(self):
206		"""Check that setting the decorator works correctly"""
207	1	class DummyWrapper(logging.Logger):
208		"""Dummy class for comparison"""
209	1	pass
210
211	1	LogManager.decorate_logger_class(lambda arg: DummyWrapper)
212
213	1	assert logging.Logger.manager.loggerClass == DummyWrapper
214
215	1	assert isinstance(logging.root, DummyWrapper)
216
217
218	1	class TestNAppLog(LogTester):
219		"""Test the log used by NApps."""
220
221	1	def setup_method(self):
222		"""Initialize variables."""
223	1	super().setup_method()
224	1	self._inspect_patcher = None
225
226	1	def teardown_method(self):
227		"""Undo mocking."""
228	1	super().teardown_method()
229	1	if self._inspect_patcher:
230	1	self._inspect_patcher.stop()
231
232	1	def _set_filename(self, filename):
233		"""Mock the NApp's main.py file path."""
234		# Put the filename in the call stack
235	1	frame = FrameInfo(None, filename, None, None, None, None)
236	1	self._inspect_patcher = patch('kytos.core.logs.inspect')
237	1	inspect = self._inspect_patcher.start()
238	1	inspect.stack.return_value = [frame]
239
240	1	def test_napp_id_detection(self):
241		"""Test NApp ID detection based on filename."""
242	1	self._set_filename('/napps/username/name/main.py')
243	1	expected_logger_name = 'kytos.napps.username/name'
244	1	napp_logger = NAppLog()
245	1	assert expected_logger_name == napp_logger.name
246
247	1	def test_napp_id_not_found(self):
248		"""If NApp ID is not found, should use root logger."""
249	1	self._set_filename('not/an/expected/NApp/path.py')
250	1	root_logger = logging.getLogger("kytos.napps")
251	1	napp_logger = NAppLog()
252		assert root_logger.name == napp_logger.name
253

kytos-ng / kytos

Push — master ( 74a04b...775f07 )

TestNAppLog.setup_method() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like