tests.unit.test_core.test_logs.TestLogManager.test_set_debug_mode() - Code Metrics - Inspection of "Merge pull request #1078 from gleybersonandrade/ta..." - kytos/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( e601a7...eb7115 )

by Humberto

created 2020-06-18 18:26 UTC

TestLogManager.test_set_debug_mode() A

↳ Parent: tests.unit.test_core.test_logs

Complexity

Conditions

Size

Total Lines	19
Code Lines	14

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	14
nop	4
dl	0
loc	19
rs	9.7
c	0
b	0
f	0

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

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


class LogTester(TestCase):
    """Common code to test loggers."""

    def setUp(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(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]
        self.assertTrue(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)
        self.assertIn(handler, 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
        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)
        self.assertIn(handler, 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 lib/python3.6/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)
            self.assertEqual(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 lib/python3.6/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.
        imp.reload(logs)
        old_handler.addFilter.assert_called_once_with(
            logs.LogManager.filter_session_disconnected)

        logging.root.removeHandler(old_handler)


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

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

    def tearDown(self):
        """Undo mocking."""
        super().tearDown()
        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()
        self.assertEqual(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()
        self.assertEqual(root_logger.name, napp_logger.name)


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

kytos / kytos

Push — master ( e601a7...eb7115 )

TestLogManager.test_set_debug_mode() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like