1 # Copyright (c) 2020 Deutsche Telekom.
2 # Licensed under the Apache License, Version 2.0 (the "License");
3 # you may not use this file except in compliance with the License.
4 # You may obtain a copy of the License at:
5 # http://www.apache.org/licenses/LICENSE-2.0
7 # Unless required by applicable law or agreed to in writing, software
8 # distributed under the License is distributed on an "AS IS" BASIS,
9 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15 if sys.version_info[0] < 3:
16 from mock import MagicMock, patch
17 if sys.version_info[0] >= 3:
18 from unittest.mock import MagicMock, patch
22 from onaplogging.mdcContext import (
36 class TestMDCContext(unittest.TestCase):
39 super(TestMDCContext, self).setUp()
42 self.TEST_VALUE = "value"
44 self.mdc_context = MDCContext()
46 def test_mdc_context(self):
48 self.assertTrue(self.mdc_context.isEmpty())
49 self.assertIsNone(self.mdc_context.get(self.TEST_KEY))
50 self.mdc_context.remove(self.TEST_KEY)
51 self.mdc_context.put(self.TEST_KEY, self.TEST_VALUE)
52 self.assertFalse(self.mdc_context.isEmpty())
53 self.assertEqual(self.mdc_context.get(self.TEST_KEY), self.TEST_VALUE)
54 self.assertDictEqual(self.mdc_context.result(), {self.TEST_KEY: self.TEST_VALUE})
55 self.mdc_context.remove(self.TEST_KEY)
56 self.assertTrue(self.mdc_context.isEmpty())
57 self.assertDictEqual(self.mdc_context.result(), {})
58 self.mdc_context.put(self.TEST_KEY, self.TEST_VALUE)
59 self.assertFalse(self.mdc_context.isEmpty())
60 self.assertEqual(self.mdc_context.get(self.TEST_KEY), self.TEST_VALUE)
61 self.assertDictEqual(self.mdc_context.result(), {self.TEST_KEY: self.TEST_VALUE})
62 self.mdc_context.clear()
63 self.assertTrue(self.mdc_context.isEmpty())
64 self.assertDictEqual(self.mdc_context.result(), {})
66 def test_getmdcs(self):
67 with patch("onaplogging.mdcContext.MDC", self.mdc_context):
68 self.assertIsNone(_getmdcs(None))
69 self.mdc_context.put(self.TEST_KEY, self.TEST_VALUE)
70 self.assertDictEqual(_getmdcs(None), {"mdc": {self.TEST_KEY: self.TEST_VALUE}})
71 self.assertDictEqual(_getmdcs({"test": "value"}), {"mdc": {self.TEST_KEY: self.TEST_VALUE}, "test": "value"})
72 with pytest.raises(KeyError):
73 _getmdcs({self.TEST_KEY: self.TEST_VALUE})
74 with pytest.raises(KeyError):
75 _getmdcs({"mdc": "exception"})
77 def test_fetchkeys_info(self):
78 with patch("onaplogging.mdcContext.MDC", self.mdc_context):
79 test_self = MagicMock()
80 test_self.isEnabledFor.return_value = False
81 info(test_self, "msg")
82 test_self._log.assert_not_called()
83 test_self.isEnabledFor.return_value = True
84 info(test_self, "msg")
85 test_self._log.assert_called_once_with(logging.INFO, "msg", (), extra=None)
86 test_self._log.reset_mock()
87 self.mdc_context.put(self.TEST_KEY, self.TEST_VALUE)
88 info(test_self, "msg")
89 test_self._log.assert_called_once_with(logging.INFO, "msg", (), extra={"mdc": {self.TEST_KEY: self.TEST_VALUE}})
91 def test_fetchkeys_debug(self):
92 with patch("onaplogging.mdcContext.MDC", self.mdc_context):
93 test_self = MagicMock()
94 test_self.isEnabledFor.return_value = False
95 debug(test_self, "msg")
96 test_self._log.assert_not_called()
97 test_self.isEnabledFor.return_value = True
98 debug(test_self, "msg")
99 test_self._log.assert_called_once_with(logging.DEBUG, "msg", (), extra=None)
100 test_self._log.reset_mock()
101 self.mdc_context.put(self.TEST_KEY, self.TEST_VALUE)
102 debug(test_self, "msg")
103 test_self._log.assert_called_once_with(logging.DEBUG, "msg", (), extra={"mdc": {self.TEST_KEY: self.TEST_VALUE}})
105 def test_fetchkeys_warning(self):
106 with patch("onaplogging.mdcContext.MDC", self.mdc_context):
107 test_self = MagicMock()
108 test_self.isEnabledFor.return_value = False
109 warning(test_self, "msg")
110 test_self._log.assert_not_called()
111 test_self.isEnabledFor.return_value = True
112 warning(test_self, "msg")
113 test_self._log.assert_called_once_with(logging.WARNING, "msg", (), extra=None)
114 test_self._log.reset_mock()
115 self.mdc_context.put(self.TEST_KEY, self.TEST_VALUE)
116 warning(test_self, "msg")
117 test_self._log.assert_called_once_with(logging.WARNING, "msg", (), extra={"mdc": {self.TEST_KEY: self.TEST_VALUE}})
119 def test_fetchkeys_exception(self):
120 with patch("onaplogging.mdcContext.MDC", self.mdc_context):
121 test_self = MagicMock()
122 test_self.isEnabledFor.return_value = False
123 exception(test_self, "msg")
124 test_self.error.assert_called_once_with("msg", exc_info=1, extra=None)
126 def test_fetchkeys_critical(self):
127 with patch("onaplogging.mdcContext.MDC", self.mdc_context):
128 test_self = MagicMock()
129 test_self.isEnabledFor.return_value = False
130 critical(test_self, "msg")
131 test_self._log.assert_not_called()
132 test_self.isEnabledFor.return_value = True
133 critical(test_self, "msg")
134 test_self._log.assert_called_once_with(logging.CRITICAL, "msg", (), extra=None)
135 test_self._log.reset_mock()
136 self.mdc_context.put(self.TEST_KEY, self.TEST_VALUE)
137 critical(test_self, "msg")
138 test_self._log.assert_called_once_with(logging.CRITICAL, "msg", (), extra={"mdc": {self.TEST_KEY: self.TEST_VALUE}})
140 def test_fetchkeys_error(self):
141 with patch("onaplogging.mdcContext.MDC", self.mdc_context):
142 test_self = MagicMock()
143 test_self.isEnabledFor.return_value = False
144 error(test_self, "msg")
145 test_self._log.assert_not_called()
146 test_self.isEnabledFor.return_value = True
147 error(test_self, "msg")
148 test_self._log.assert_called_once_with(logging.ERROR, "msg", (), extra=None)
149 test_self._log.reset_mock()
150 self.mdc_context.put(self.TEST_KEY, self.TEST_VALUE)
151 error(test_self, "msg")
152 test_self._log.assert_called_once_with(logging.ERROR, "msg", (), extra={"mdc": {self.TEST_KEY: self.TEST_VALUE}})
154 def test_fetchkeys_log(self):
155 with patch("onaplogging.mdcContext.MDC", self.mdc_context):
156 test_self = MagicMock()
157 test_self.isEnabledFor.return_value = False
158 logging.raiseExceptions = False
159 log(test_self, "invalid_level", "msg")
160 logging.raiseExceptions = True
161 with pytest.raises(TypeError):
162 log(test_self, "invalid_level", "msg")
163 log(test_self, logging.DEBUG, "msg")
164 test_self._log.assert_not_called()
165 test_self.isEnabledFor.return_value = True
166 log(test_self, logging.DEBUG, "msg")
167 test_self._log.assert_called_once()
169 def test_handle(self):
170 with patch("onaplogging.mdcContext.MDC", self.mdc_context):
171 test_self = MagicMock()
173 test_self.disabled = True
174 test_self.filter.return_value = False
175 handle(test_self, record)
176 test_self.callHandlers.assert_not_called()
178 test_self.disabled = False
179 test_self.filter.return_value = False
180 handle(test_self, record)
181 test_self.callHandlers.assert_not_called()
182 test_self.filter.assert_called_once_with(record)
184 test_self.filter.reset_mock()
185 test_self.disabled = False
186 test_self.filter.return_value = True
187 handle(test_self, record)
188 test_self.callHandlers.assert_called_once()
189 test_self.filter.assert_called_once_with(record)