diff options
Diffstat (limited to 'pylog')
| -rw-r--r-- | pylog/onaplogging/colorFormatter.py | 22 | ||||
| -rw-r--r-- | pylog/onaplogging/marker/markerHandler.py | 2 | ||||
| -rw-r--r-- | pylog/onaplogging/mdcContext.py | 2 | ||||
| -rw-r--r-- | pylog/onaplogging/mdcformatter.py | 16 | ||||
| -rw-r--r-- | pylog/onaplogging/utils.py | 31 | ||||
| -rw-r--r-- | pylog/test_requirements.txt | 3 | ||||
| -rw-r--r-- | pylog/tests/test_color_formatter.py | 133 | ||||
| -rw-r--r-- | pylog/tests/test_example.py | 18 | ||||
| -rw-r--r-- | pylog/tests/test_log_watchdog.py | 90 | ||||
| -rw-r--r-- | pylog/tests/test_marker.py | 179 | ||||
| -rw-r--r-- | pylog/tests/test_marker_formatter.py | 67 | ||||
| -rw-r--r-- | pylog/tests/test_marker_log_adaptor.py | 76 | ||||
| -rw-r--r-- | pylog/tests/test_mdc_context.py | 189 | ||||
| -rw-r--r-- | pylog/tests/test_mdc_formatter.py | 108 | ||||
| -rw-r--r-- | pylog/tests/test_monkey.py | 47 | ||||
| -rw-r--r-- | pylog/tests/test_utils.py | 45 | ||||
| -rw-r--r-- | pylog/tox.ini | 7 |
17 files changed, 990 insertions, 45 deletions
diff --git a/pylog/onaplogging/colorFormatter.py b/pylog/onaplogging/colorFormatter.py index 64e220a..ef713ea 100644 --- a/pylog/onaplogging/colorFormatter.py +++ b/pylog/onaplogging/colorFormatter.py @@ -14,9 +14,10 @@ import os import sys -import logging from logging import Formatter +from .utils import is_above_python_2_7, is_above_python_3_2 + ATTRIBUTES = { 'normal': 0, @@ -57,7 +58,8 @@ COLOR_TAG = "color" HIGHLIGHT_TAG = "highlight" ATTRIBUTE_TAG = "attribute" -RESET = '\033[0m' +RESET = "\033[0m" +FMT_STR = "\033[%dm%s" def colored(text, color=None, on_color=None, attrs=None): @@ -70,16 +72,15 @@ def colored(text, color=None, on_color=None, attrs=None): attrs = [attrs] if os.getenv('ANSI_COLORS_DISABLED', None) is None: - fmt_str = '\033[%dm%s' if color is not None and isinstance(color, str): - text = fmt_str % (COLORS.get(color, 0), text) + text = FMT_STR % (COLORS.get(color, 0), text) if on_color is not None and isinstance(on_color, str): - text = fmt_str % (HIGHLIGHTS.get(on_color, 0), text) + text = FMT_STR % (HIGHLIGHTS.get(on_color, 0), text) if attrs is not None: for attr in attrs: - text = fmt_str % (ATTRIBUTES.get(attr, 0), text) + text = FMT_STR % (ATTRIBUTES.get(attr, 0), text) # keep origin color for tail spaces text += RESET @@ -89,20 +90,15 @@ def colored(text, color=None, on_color=None, attrs=None): class BaseColorFormatter(Formatter): def __init__(self, fmt=None, datefmt=None, colorfmt=None, style="%"): - if sys.version_info > (3, 2): + if is_above_python_3_2(): super(BaseColorFormatter, self).__init__( fmt=fmt, datefmt=datefmt, style=style) - elif sys.version_info > (2, 7): + elif is_above_python_2_7(): super(BaseColorFormatter, self).__init__(fmt, datefmt) else: Formatter.__init__(self, fmt, datefmt) self.style = style - if sys.version_info > (3, 2): - if self.style not in logging._STYLES: - raise ValueError('Style must be one of: %s' % ','.join( - logging._STYLES.keys())) - self.colorfmt = colorfmt def _parseColor(self, record): diff --git a/pylog/onaplogging/marker/markerHandler.py b/pylog/onaplogging/marker/markerHandler.py index 12b5488..e9ce810 100644 --- a/pylog/onaplogging/marker/markerHandler.py +++ b/pylog/onaplogging/marker/markerHandler.py @@ -44,7 +44,7 @@ class MarkerNotifyHandler(SMTPHandler): if matchMarkerHelp(record, self.markers): if sys.version_info > (2, 7): - return super(SMTPHandler, self).handle(record) + return super(MarkerNotifyHandler, self).handle(record) else: return SMTPHandler.handle(self, record) diff --git a/pylog/onaplogging/mdcContext.py b/pylog/onaplogging/mdcContext.py index fa94536..ecdc2d9 100644 --- a/pylog/onaplogging/mdcContext.py +++ b/pylog/onaplogging/mdcContext.py @@ -48,7 +48,7 @@ class MDCContext(threading.local): def remove(self, key): - if key in self.localDict: + if key in self._localDict: del self._localDict[key] def clear(self): diff --git a/pylog/onaplogging/mdcformatter.py b/pylog/onaplogging/mdcformatter.py index 545a4c1..4cacbe8 100644 --- a/pylog/onaplogging/mdcformatter.py +++ b/pylog/onaplogging/mdcformatter.py @@ -12,9 +12,9 @@ # See the License for the specific language governing permissions and # limitations under the License. -import sys import logging from .markerFormatter import MarkerFormatter +from .utils import is_above_python_2_7, is_above_python_3_2 class MDCFormatter(MarkerFormatter): @@ -33,12 +33,12 @@ class MDCFormatter(MarkerFormatter): :param colorfmt: colored output with ANSI escape code on terminal :param style: style mapping keys in python3 """ - if sys.version_info > (3, 2): + if is_above_python_3_2(): super(MDCFormatter, self).__init__(fmt=fmt, datefmt=datefmt, colorfmt=colorfmt, style=style) - elif sys.version_info > (2, 7): + elif is_above_python_2_7(): super(MDCFormatter, self).__init__(fmt=fmt, datefmt=datefmt, colorfmt=colorfmt) @@ -112,7 +112,7 @@ class MDCFormatter(MarkerFormatter): """ mdcIndex = self._fmt.find(self._mdc_tag) if mdcIndex == -1: - if sys.version_info > (2, 7): + if is_above_python_2_7(): return super(MDCFormatter, self).format(record) else: return MarkerFormatter.format(self, record) @@ -121,10 +121,10 @@ class MDCFormatter(MarkerFormatter): if mdcFmtWords is None: self._fmt = self._fmt.replace(self._mdc_tag, "") - if sys.version_info > (3, 2): + if is_above_python_3_2(): self._style = logging._STYLES[self.style][0](self._fmt) - if sys.version_info > (2, 7): + if is_above_python_2_7(): return super(MDCFormatter, self).format(record) else: return MarkerFormatter.format(self, record) @@ -142,10 +142,10 @@ class MDCFormatter(MarkerFormatter): mdcstr = self._replaceStr(keys=mdcFmtkeys).format(**res) self._fmt = self._fmt.replace(self._mdc_tag, mdcstr) - if sys.version_info > (3, 2): + if is_above_python_3_2(): self._style = logging._STYLES[self.style][0](self._fmt) - if sys.version_info > (2, 7): + if is_above_python_2_7(): return super(MDCFormatter, self).format(record) else: return MarkerFormatter.format(self, record) diff --git a/pylog/onaplogging/utils.py b/pylog/onaplogging/utils.py new file mode 100644 index 0000000..5c96b2d --- /dev/null +++ b/pylog/onaplogging/utils.py @@ -0,0 +1,31 @@ +# Copyright (c) 2020 Deutsche Telekom. +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + +import sys + + +def is_above_python_3_2(): # type: () -> bool + """Check if code is running at least on Python 3.2 version. + + Returns: + bool: True if it's at least 3.2 version, False otherwise + + """ + return sys.version_info >= (3, 2, 0, "final", 0) + + +def is_above_python_2_7(): # type: () -> bool + """Check if code is running at least on Python 2.7 version. + + Returns: + bool: True if it's at least 2.7 version, False otherwise + + """ + return sys.version_info >= (2, 7, 0, "final", 0) diff --git a/pylog/test_requirements.txt b/pylog/test_requirements.txt new file mode 100644 index 0000000..d32736b --- /dev/null +++ b/pylog/test_requirements.txt @@ -0,0 +1,3 @@ +-r requirements.txt +pytest +mock
\ No newline at end of file diff --git a/pylog/tests/test_color_formatter.py b/pylog/tests/test_color_formatter.py new file mode 100644 index 0000000..9a9ae5a --- /dev/null +++ b/pylog/tests/test_color_formatter.py @@ -0,0 +1,133 @@ +# Copyright (c) 2020 Deutsche Telekom. +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + +import unittest +import sys +from logging import LogRecord + +if sys.version_info[0] < 3: + from mock import patch +if sys.version_info[0] >= 3: + from unittest.mock import patch +import pytest + +from onaplogging.colorFormatter import ( + ATTRIBUTES, + BaseColorFormatter, + colored, + COLORS, + HIGHLIGHTS, + FMT_STR, + RESET, +) +from onaplogging.utils import is_above_python_3_2 + + +class TestColorFormatter(unittest.TestCase): + + TEST_TEXT = "test" + + def test_colored_os_name_nt(self): + + with patch("onaplogging.colorFormatter.os.name", "nt"): + + text = colored(self.TEST_TEXT) + assert text == self.TEST_TEXT + + text = colored(self.TEST_TEXT, color="black") + assert text == self.TEST_TEXT + + text = colored(self.TEST_TEXT, on_color="black") + assert text == self.TEST_TEXT + + text = colored(self.TEST_TEXT, attrs="bold") + assert text == self.TEST_TEXT + + def test_colored_os_name_ce(self): + + with patch("onaplogging.colorFormatter.os.name", "ce"): + + text = colored(self.TEST_TEXT) + assert text == self.TEST_TEXT + + text = colored(self.TEST_TEXT, color="black") + assert text == self.TEST_TEXT + + text = colored(self.TEST_TEXT, on_color="black") + assert text == self.TEST_TEXT + + text = colored(self.TEST_TEXT, attrs="bold") + assert text == self.TEST_TEXT + + def test_colored_os_name_posix(self): + + with patch("onaplogging.colorFormatter.os.name", "posix"): + text = colored(self.TEST_TEXT) + assert text == self.TEST_TEXT + RESET + + text = colored(self.TEST_TEXT, color="black") + assert text == FMT_STR % (COLORS["black"], self.TEST_TEXT) + RESET + + text = colored(self.TEST_TEXT, color="invalid") + assert text == FMT_STR % (0, self.TEST_TEXT) + RESET + + text = colored(self.TEST_TEXT, on_color="red") + assert text == FMT_STR % (HIGHLIGHTS["red"], self.TEST_TEXT) + RESET + + text = colored(self.TEST_TEXT, on_color="invalid") + assert text == FMT_STR % (0, self.TEST_TEXT) + RESET + + text = colored(self.TEST_TEXT, attrs="bold") + assert text == FMT_STR % (ATTRIBUTES["bold"], self.TEST_TEXT) + RESET + + text = colored(self.TEST_TEXT, attrs=["bold", "blink"]) + assert ( + text + == FMT_STR % (ATTRIBUTES["blink"], FMT_STR % (ATTRIBUTES["bold"], self.TEST_TEXT)) + + RESET + ) + + text = colored(self.TEST_TEXT, attrs="invalid") + assert text == FMT_STR % (0, self.TEST_TEXT) + RESET + + def test_base_color_formatter(self): + + if is_above_python_3_2(): + with pytest.raises(ValueError): + BaseColorFormatter(style="!") + + TEST_MESSAGE = "TestMessage" + record = LogRecord( + name="TestName", + level=0, + pathname="TestPathName", + lineno=1, + msg=TEST_MESSAGE, + args=None, + exc_info=None, + ) + + base_formatter = BaseColorFormatter() + assert base_formatter.format(record) == TEST_MESSAGE + RESET + + base_formatter = BaseColorFormatter(fmt="TEST %(message)s") + assert base_formatter.format(record) == "TEST " + TEST_MESSAGE + RESET + + colorfmt = {record.levelname: {"color": "black", "highlight": "red", "attribute": "bold"}} + base_formatter = BaseColorFormatter(colorfmt=colorfmt) + assert ( + base_formatter.format(record) + == FMT_STR + % ( + ATTRIBUTES["bold"], + FMT_STR % (HIGHLIGHTS["red"], FMT_STR % (COLORS["black"], "TestMessage")), + ) + + RESET + ) diff --git a/pylog/tests/test_example.py b/pylog/tests/test_example.py deleted file mode 100644 index c0d97bf..0000000 --- a/pylog/tests/test_example.py +++ /dev/null @@ -1,18 +0,0 @@ -# Copyright (c) 2018-2019 VMware, Inc. -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at: -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - - -import unittest - - -class TestExample(unittest.TestCase): - - def test_mdcFormat(self): - return diff --git a/pylog/tests/test_log_watchdog.py b/pylog/tests/test_log_watchdog.py new file mode 100644 index 0000000..5f43138 --- /dev/null +++ b/pylog/tests/test_log_watchdog.py @@ -0,0 +1,90 @@ +# Copyright (c) 2020 Deutsche Telekom. +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + +import os +import sys +import unittest +from collections import namedtuple +from tempfile import NamedTemporaryFile + +if sys.version_info[0] < 3: + from mock import patch +if sys.version_info[0] >= 3: + from unittest.mock import patch + +import pytest +import yaml + +from onaplogging.logWatchDog import FileEventHandlers, _yaml2Dict, _yamlConfig + + +TestEvent = namedtuple("TestEvent", ["src_path"]) + + +class TestLogWatchdog(unittest.TestCase): + + TEST_DICT = { + "A": { + "B": "C" + } + } + + def setUp(self): + super(TestLogWatchdog, self).setUp() + + self.temp_file = NamedTemporaryFile(mode="w+t", delete=False) + self.temp_file.write(yaml.dump(self.TEST_DICT)) + self.temp_file.close() + + def tearDown(self): + super(TestLogWatchdog, self).tearDown() + + os.unlink(self.temp_file.name) + + def test_yaml2dict(self): + with pytest.raises(TypeError): + _yaml2Dict(None) + + self.assertDictEqual(self.TEST_DICT, _yaml2Dict(self.temp_file.name)) + + def test_file_event_handler(self): + + with patch("onaplogging.logWatchDog.config.dictConfig") as mock_config: + mock_config.side_effect = Exception + + feh = FileEventHandlers(self.temp_file.name) + self.assertIsNone(feh.currentConfig) + feh.on_modified(TestEvent(src_path=self.temp_file.name)) + self.assertIsNone(feh.currentConfig) + + with patch("onaplogging.logWatchDog.config"): + + feh = FileEventHandlers(self.temp_file.name) + self.assertIsNone(feh.currentConfig) + feh.on_modified(TestEvent(src_path=self.temp_file.name)) + self.assertIsNotNone(feh.currentConfig) + + def test_patch_yaml_config(self): + + with pytest.raises(TypeError): + _yamlConfig(filepath=None) + + with pytest.raises(OSError): + _yamlConfig(filepath="invalid path") + + with patch("onaplogging.logWatchDog.config.dictConfig") as mock_config: + _yamlConfig(filepath=self.temp_file.name) + mock_config.assert_called_once_with(self.TEST_DICT) + + with patch("onaplogging.logWatchDog.config.dictConfig") as mock_config: + with patch("onaplogging.logWatchDog.Observer.start") as mock_observer_start: + _yamlConfig(filepath=self.temp_file.name, watchDog=True) + mock_config.assert_called_once_with(self.TEST_DICT) + mock_observer_start.assert_called_once() diff --git a/pylog/tests/test_marker.py b/pylog/tests/test_marker.py new file mode 100644 index 0000000..c9e9f62 --- /dev/null +++ b/pylog/tests/test_marker.py @@ -0,0 +1,179 @@ +# Copyright (c) 2020 Deutsche Telekom. +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + +import sys +import unittest +from collections import namedtuple + +if sys.version_info[0] < 3: + from mock import patch +if sys.version_info[0] >= 3: + from unittest.mock import patch + +import pytest + +from onaplogging.marker import BaseMarker, matchMarkerHelp, MarkerFactory, MarkerFilter, MarkerNotifyHandler + + +class TestRecordMixin(object): + + Record = namedtuple("Record", "marker") + + +class TestNameMixin(object): + + TEST_NAME = "test_base" + + +class TestBaseMarker(unittest.TestCase, TestNameMixin): + + def setUp(self): + super(TestBaseMarker, self).setUp() + self.base_marker = BaseMarker(name=self.TEST_NAME) + + def test_base_marker_name(self): + with pytest.raises(TypeError): + BaseMarker(123) + + with pytest.raises(ValueError): + BaseMarker(name="") + + self.assertEqual(self.base_marker.getName(), self.TEST_NAME) + + def test_base_marker_contains(self): + self.assertTrue(self.base_marker.contains(self.base_marker)) + self.assertTrue(self.base_marker.contains(self.TEST_NAME)) + + def test_base_marker_compare(self): + self.assertNotEqual(self.base_marker, 3) + self.assertEqual(self.base_marker, self.base_marker) + other = BaseMarker("Other") + self.assertNotEqual(self.base_marker, other) + other = BaseMarker(self.TEST_NAME) + self.assertEqual(self.base_marker, other) + + def test_base_marker_child(self): + self.assertListEqual(list(iter(self.base_marker)), []) + self.assertFalse(self.base_marker.contains(3)) + with pytest.raises(TypeError): + self.base_marker.addChild(3) + with pytest.raises(TypeError): + self.base_marker.addChild("str") + with pytest.raises(TypeError): + self.base_marker.removeChild(3) + + self.base_marker.addChild(self.base_marker) + self.assertListEqual(list(iter(self.base_marker)), []) + + child1 = BaseMarker(name="child1") + self.assertFalse(self.base_marker.contains(child1)) + self.base_marker.addChild(child1) + self.assertListEqual(list(iter(self.base_marker)), [child1]) + self.assertTrue(self.base_marker.contains(child1)) + self.base_marker.addChild(child1) + self.assertListEqual(list(iter(self.base_marker)), [child1]) + + self.base_marker.removeChild(child1) + self.assertListEqual(list(iter(self.base_marker)), []) + self.assertFalse(self.base_marker.contains(child1)) + + child2 = BaseMarker(name="child2") + self.assertFalse(self.base_marker.contains(child2)) + + with pytest.raises(TypeError): + self.base_marker.addChilds(None) + self.base_marker.addChilds((child1, child2,)) + self.assertTrue(self.base_marker.contains(child1)) + self.assertTrue(self.base_marker.contains(child2)) + self.base_marker.removeChild(child1) + self.assertFalse(self.base_marker.contains(child1)) + self.assertTrue(self.base_marker.contains(child2)) + self.assertFalse(self.base_marker.contains("child1")) + self.assertTrue(self.base_marker.contains("child2")) + + +class TestMatchMarkerHelp(unittest.TestCase, TestRecordMixin, TestNameMixin): + CHILD_NAME = "child" + + def test_match_marker_help(self): + record = self.Record(None) + self.assertFalse(matchMarkerHelp(record, "anything")) + + record = self.Record("not_marker_instance") + self.assertFalse(matchMarkerHelp(record, "not_marker_instance")) + + marker = BaseMarker(self.TEST_NAME) + record = self.Record(marker) + self.assertFalse(matchMarkerHelp(record, "invalid_name")) + self.assertTrue(matchMarkerHelp(record, marker)) + self.assertTrue(matchMarkerHelp(record, self.TEST_NAME)) + + child = BaseMarker(self.CHILD_NAME) + marker.addChild(child) + self.assertTrue(matchMarkerHelp(record, [self.TEST_NAME, self.CHILD_NAME])) + self.assertTrue(matchMarkerHelp(record, [marker, self.CHILD_NAME])) + self.assertTrue(matchMarkerHelp(record, [marker, child])) + self.assertTrue(matchMarkerHelp(record, [marker, "invalid"])) + + +class TestMarkerFactory(unittest.TestCase, TestNameMixin): + + def setUp(self): + super(TestMarkerFactory, self).setUp() + self.marker_factory = MarkerFactory() + + def test_get_marker(self): + with pytest.raises(ValueError): + self.marker_factory.getMarker() + self.assertEqual(len(self.marker_factory._marker_map), 0) + marker = self.marker_factory.getMarker(self.TEST_NAME) + self.assertEqual(marker.getName(), self.TEST_NAME) + self.assertEqual(len(self.marker_factory._marker_map), 1) + marker = self.marker_factory.getMarker(self.TEST_NAME) + self.assertEqual(marker.getName(), self.TEST_NAME) + self.assertEqual(len(self.marker_factory._marker_map), 1) + + self.assertTrue(self.marker_factory.exist(marker.getName())) + + self.assertTrue(self.marker_factory.deleteMarker(marker.getName())) + self.assertFalse(self.marker_factory.exist(marker.getName())) + self.assertEqual(len(self.marker_factory._marker_map), 0) + + self.assertFalse(self.marker_factory.deleteMarker(marker.getName())) + + +class TestMarkerFilter(unittest.TestCase, TestRecordMixin, TestNameMixin): + + def test_marker_filter(self): + marker_filter = MarkerFilter() + + record = self.Record(BaseMarker(self.TEST_NAME)) + self.assertFalse(marker_filter.filter(record)) + + marker_filter = MarkerFilter(markers=BaseMarker(self.TEST_NAME)) + self.assertTrue(marker_filter.filter(record)) + + +class TestMarkerNotifyHandler(unittest.TestCase, TestRecordMixin, TestNameMixin): + + def test_marker_notify_handler(self): + record = self.Record(BaseMarker(self.TEST_NAME)) + + notify_handler = MarkerNotifyHandler("test_host", "fromaddr", "toaddrs", "subject") + self.assertIsNone(notify_handler.markers) + self.assertFalse(notify_handler.handle(record)) + + marker = BaseMarker(self.TEST_NAME) + notify_handler = MarkerNotifyHandler("test_host", "fromaddr", "toaddrs", "subject", markers=[marker]) + with patch("onaplogging.marker.markerHandler.SMTPHandler.handle") as mock_smtp_handler_handle: + mock_smtp_handler_handle.return_value = True + self.assertTrue(notify_handler.handle(record)) + record = self.Record(BaseMarker("other")) + self.assertFalse(notify_handler.handle(record)) diff --git a/pylog/tests/test_marker_formatter.py b/pylog/tests/test_marker_formatter.py new file mode 100644 index 0000000..bda7f24 --- /dev/null +++ b/pylog/tests/test_marker_formatter.py @@ -0,0 +1,67 @@ +# Copyright (c) 2020 Deutsche Telekom. +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + +import sys +import unittest +from collections import namedtuple + +if sys.version_info[0] < 3: + from mock import patch +if sys.version_info[0] >= 3: + from unittest.mock import patch + +import pytest + +from onaplogging.marker import BaseMarker +from onaplogging.markerFormatter import MarkerFormatter + + +class TestMarkerFormatter(unittest.TestCase): + + Record = namedtuple("Record", "marker") + + def test_marker_formatter_init(self): + marker_formatter = MarkerFormatter() + self.assertEqual(marker_formatter.style, "%") + self.assertEqual(marker_formatter._marker_tag, "%(marker)s") + + if sys.version_info[0] >= 3: + marker_formatter = MarkerFormatter(style="{") + self.assertEqual(marker_formatter.style, "{") + self.assertEqual(marker_formatter._marker_tag, "{marker}") + + marker_formatter = MarkerFormatter(style="$") + self.assertEqual(marker_formatter.style, "$") + self.assertEqual(marker_formatter._marker_tag, "${marker}") + + with pytest.raises(ValueError): + MarkerFormatter(style="*") + + def test_marker_formatter_format(self): + record = self.Record(BaseMarker("test")) + + with patch("onaplogging.markerFormatter.BaseColorFormatter.format") as mock_format: + marker_formatter = MarkerFormatter() + self.assertEqual(marker_formatter._fmt, "%(message)s") + self.assertEqual(marker_formatter._marker_tag, "%(marker)s") + marker_formatter.format(record) + mock_format.assert_called_once() + self.assertEqual(marker_formatter._fmt, "%(message)s") + self.assertEqual(marker_formatter._marker_tag, "%(marker)s") + + with patch("onaplogging.markerFormatter.BaseColorFormatter.format") as mock_format: + marker_formatter = MarkerFormatter(fmt="%(message)s %(marker)s") + self.assertEqual(marker_formatter._fmt, "%(message)s %(marker)s") + self.assertEqual(marker_formatter._marker_tag, "%(marker)s") + marker_formatter.format(record) + mock_format.assert_called_once() + self.assertEqual(marker_formatter._fmt, "%(message)s %(marker)s") + self.assertEqual(marker_formatter._marker_tag, "%(marker)s") + self.assertEqual(marker_formatter._tmpFmt, "%(message)s %(marker)s") diff --git a/pylog/tests/test_marker_log_adaptor.py b/pylog/tests/test_marker_log_adaptor.py new file mode 100644 index 0000000..35f852f --- /dev/null +++ b/pylog/tests/test_marker_log_adaptor.py @@ -0,0 +1,76 @@ +# Copyright (c) 2020 Deutsche Telekom. +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + +import sys +import unittest + +if sys.version_info[0] < 3: + from mock import MagicMock, patch +if sys.version_info[0] >= 3: + from unittest.mock import MagicMock, patch + +import pytest + +from onaplogging.marker import BaseMarker +from onaplogging.markerLogAdaptor import MarkerLogAdaptor + + +class TestMarkerLogAdaptor(unittest.TestCase): + + def test_process(self): + log_adaptor = MarkerLogAdaptor(MagicMock(), extra=None) + msg, kwargs = log_adaptor.process("test", {}) + self.assertEqual(msg, "test") + self.assertDictEqual(kwargs, {"extra": None}) + + log_adaptor = MarkerLogAdaptor(MagicMock(), extra={"A": "B"}) + msg, kwargs = log_adaptor.process("test", {}) + self.assertEqual(msg, "test") + self.assertDictEqual(kwargs, {"extra": {"A": "B"}}) + + # Commented out due to that: https://bugs.python.org/issue20239 + # Comment out if Jenkis build runs using Python > 3.6 + # def test_markers(self): + # log_adaptor = MarkerLogAdaptor(MagicMock(), extra=None) + + # with patch("onaplogging.markerLogAdaptor.LoggerAdapter.info") as mock_info: + # log_adaptor.infoMarker(BaseMarker("info_marker"), "test_message") + # mock_info.assert_called_once() + + # with patch("onaplogging.markerLogAdaptor.LoggerAdapter.debug") as mock_debug: + # log_adaptor.debugMarker(BaseMarker("info_marker"), "test_message") + # mock_debug.assert_called_once() + + # with patch("onaplogging.markerLogAdaptor.LoggerAdapter.warning") as mock_warning: + # log_adaptor.warningMarker(BaseMarker("info_marker"), "test_message") + # mock_warning.assert_called_once() + + # with patch("onaplogging.markerLogAdaptor.LoggerAdapter.error") as mock_error: + # log_adaptor.errorMarker(BaseMarker("info_marker"), "test_message") + # mock_error.assert_called_once() + + # with patch("onaplogging.markerLogAdaptor.LoggerAdapter.exception") as mock_exception: + # log_adaptor.exceptionMarker(BaseMarker("info_marker"), "test_message") + # mock_exception.assert_called_once() + + # with patch("onaplogging.markerLogAdaptor.LoggerAdapter.critical") as mock_critical: + # log_adaptor.criticalMarker(BaseMarker("info_marker"), "test_message") + # mock_critical.assert_called_once() + + # with patch("onaplogging.markerLogAdaptor.LoggerAdapter.log") as mock_log: + # log_adaptor.logMarker(BaseMarker("info_marker"), "info", "test_message") + # mock_log.assert_called_once() + + # with pytest.raises(TypeError): + # log_adaptor.infoMarker("info_marker_str", "test_message") + + # with pytest.raises(Exception): + # log_adaptor = MarkerLogAdaptor(MagicMock(), extra={"marker": "exception"}) + # log_adaptor.infoMarker(BaseMarker("info_marker"), "test_message") diff --git a/pylog/tests/test_mdc_context.py b/pylog/tests/test_mdc_context.py new file mode 100644 index 0000000..9e8b1bc --- /dev/null +++ b/pylog/tests/test_mdc_context.py @@ -0,0 +1,189 @@ +# Copyright (c) 2020 Deutsche Telekom. +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + +import logging +import sys +import unittest + +if sys.version_info[0] < 3: + from mock import MagicMock, patch +if sys.version_info[0] >= 3: + from unittest.mock import MagicMock, patch + +import pytest + +from onaplogging.mdcContext import ( + _getmdcs, + MDCContext, + info, + debug, + warning, + exception, + critical, + error, + log, + handle +) + + +class TestMDCContext(unittest.TestCase): + + def setUp(self): + super(TestMDCContext, self).setUp() + + self.TEST_KEY = "key" + self.TEST_VALUE = "value" + + self.mdc_context = MDCContext() + + def test_mdc_context(self): + + self.assertTrue(self.mdc_context.isEmpty()) + self.assertIsNone(self.mdc_context.get(self.TEST_KEY)) + self.mdc_context.remove(self.TEST_KEY) + self.mdc_context.put(self.TEST_KEY, self.TEST_VALUE) + self.assertFalse(self.mdc_context.isEmpty()) + self.assertEqual(self.mdc_context.get(self.TEST_KEY), self.TEST_VALUE) + self.assertDictEqual(self.mdc_context.result(), {self.TEST_KEY: self.TEST_VALUE}) + self.mdc_context.remove(self.TEST_KEY) + self.assertTrue(self.mdc_context.isEmpty()) + self.assertDictEqual(self.mdc_context.result(), {}) + self.mdc_context.put(self.TEST_KEY, self.TEST_VALUE) + self.assertFalse(self.mdc_context.isEmpty()) + self.assertEqual(self.mdc_context.get(self.TEST_KEY), self.TEST_VALUE) + self.assertDictEqual(self.mdc_context.result(), {self.TEST_KEY: self.TEST_VALUE}) + self.mdc_context.clear() + self.assertTrue(self.mdc_context.isEmpty()) + self.assertDictEqual(self.mdc_context.result(), {}) + + def test_getmdcs(self): + with patch("onaplogging.mdcContext.MDC", self.mdc_context): + self.assertIsNone(_getmdcs(None)) + self.mdc_context.put(self.TEST_KEY, self.TEST_VALUE) + self.assertDictEqual(_getmdcs(None), {"mdc": {self.TEST_KEY: self.TEST_VALUE}}) + self.assertDictEqual(_getmdcs({"test": "value"}), {"mdc": {self.TEST_KEY: self.TEST_VALUE}, "test": "value"}) + with pytest.raises(KeyError): + _getmdcs({self.TEST_KEY: self.TEST_VALUE}) + with pytest.raises(KeyError): + _getmdcs({"mdc": "exception"}) + + def test_fetchkeys_info(self): + with patch("onaplogging.mdcContext.MDC", self.mdc_context): + test_self = MagicMock() + test_self.isEnabledFor.return_value = False + info(test_self, "msg") + test_self._log.assert_not_called() + test_self.isEnabledFor.return_value = True + info(test_self, "msg") + test_self._log.assert_called_once_with(logging.INFO, "msg", (), extra=None) + test_self._log.reset_mock() + self.mdc_context.put(self.TEST_KEY, self.TEST_VALUE) + info(test_self, "msg") + test_self._log.assert_called_once_with(logging.INFO, "msg", (), extra={"mdc": {self.TEST_KEY: self.TEST_VALUE}}) + + def test_fetchkeys_debug(self): + with patch("onaplogging.mdcContext.MDC", self.mdc_context): + test_self = MagicMock() + test_self.isEnabledFor.return_value = False + debug(test_self, "msg") + test_self._log.assert_not_called() + test_self.isEnabledFor.return_value = True + debug(test_self, "msg") + test_self._log.assert_called_once_with(logging.DEBUG, "msg", (), extra=None) + test_self._log.reset_mock() + self.mdc_context.put(self.TEST_KEY, self.TEST_VALUE) + debug(test_self, "msg") + test_self._log.assert_called_once_with(logging.DEBUG, "msg", (), extra={"mdc": {self.TEST_KEY: self.TEST_VALUE}}) + + def test_fetchkeys_warning(self): + with patch("onaplogging.mdcContext.MDC", self.mdc_context): + test_self = MagicMock() + test_self.isEnabledFor.return_value = False + warning(test_self, "msg") + test_self._log.assert_not_called() + test_self.isEnabledFor.return_value = True + warning(test_self, "msg") + test_self._log.assert_called_once_with(logging.WARNING, "msg", (), extra=None) + test_self._log.reset_mock() + self.mdc_context.put(self.TEST_KEY, self.TEST_VALUE) + warning(test_self, "msg") + test_self._log.assert_called_once_with(logging.WARNING, "msg", (), extra={"mdc": {self.TEST_KEY: self.TEST_VALUE}}) + + def test_fetchkeys_exception(self): + with patch("onaplogging.mdcContext.MDC", self.mdc_context): + test_self = MagicMock() + test_self.isEnabledFor.return_value = False + exception(test_self, "msg") + test_self.error.assert_called_once_with("msg", exc_info=1, extra=None) + + def test_fetchkeys_critical(self): + with patch("onaplogging.mdcContext.MDC", self.mdc_context): + test_self = MagicMock() + test_self.isEnabledFor.return_value = False + critical(test_self, "msg") + test_self._log.assert_not_called() + test_self.isEnabledFor.return_value = True + critical(test_self, "msg") + test_self._log.assert_called_once_with(logging.CRITICAL, "msg", (), extra=None) + test_self._log.reset_mock() + self.mdc_context.put(self.TEST_KEY, self.TEST_VALUE) + critical(test_self, "msg") + test_self._log.assert_called_once_with(logging.CRITICAL, "msg", (), extra={"mdc": {self.TEST_KEY: self.TEST_VALUE}}) + + def test_fetchkeys_error(self): + with patch("onaplogging.mdcContext.MDC", self.mdc_context): + test_self = MagicMock() + test_self.isEnabledFor.return_value = False + error(test_self, "msg") + test_self._log.assert_not_called() + test_self.isEnabledFor.return_value = True + error(test_self, "msg") + test_self._log.assert_called_once_with(logging.ERROR, "msg", (), extra=None) + test_self._log.reset_mock() + self.mdc_context.put(self.TEST_KEY, self.TEST_VALUE) + error(test_self, "msg") + test_self._log.assert_called_once_with(logging.ERROR, "msg", (), extra={"mdc": {self.TEST_KEY: self.TEST_VALUE}}) + + def test_fetchkeys_log(self): + with patch("onaplogging.mdcContext.MDC", self.mdc_context): + test_self = MagicMock() + test_self.isEnabledFor.return_value = False + logging.raiseExceptions = False + log(test_self, "invalid_level", "msg") + logging.raiseExceptions = True + with pytest.raises(TypeError): + log(test_self, "invalid_level", "msg") + log(test_self, logging.DEBUG, "msg") + test_self._log.assert_not_called() + test_self.isEnabledFor.return_value = True + log(test_self, logging.DEBUG, "msg") + test_self._log.assert_called_once() + + def test_handle(self): + with patch("onaplogging.mdcContext.MDC", self.mdc_context): + test_self = MagicMock() + record = MagicMock() + test_self.disabled = True + test_self.filter.return_value = False + handle(test_self, record) + test_self.callHandlers.assert_not_called() + + test_self.disabled = False + test_self.filter.return_value = False + handle(test_self, record) + test_self.callHandlers.assert_not_called() + test_self.filter.assert_called_once_with(record) + + test_self.filter.reset_mock() + test_self.disabled = False + test_self.filter.return_value = True + handle(test_self, record) + test_self.callHandlers.assert_called_once() + test_self.filter.assert_called_once_with(record) diff --git a/pylog/tests/test_mdc_formatter.py b/pylog/tests/test_mdc_formatter.py new file mode 100644 index 0000000..6866b9d --- /dev/null +++ b/pylog/tests/test_mdc_formatter.py @@ -0,0 +1,108 @@ +# Copyright (c) 2020 Deutsche Telekom. +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + +import sys +import unittest + +if sys.version_info[0] < 3: + from mock import MagicMock, patch +if sys.version_info[0] >= 3: + from unittest.mock import MagicMock, patch + +import pytest + +from onaplogging.mdcformatter import MDCFormatter + + +class TestMdcFormatter(unittest.TestCase): + + def test_mdc_formatter_init(self): + mdc_formatter = MDCFormatter() + self.assertEqual(mdc_formatter.style, "%") + self.assertEqual(mdc_formatter._mdc_tag, "%(mdc)s") + self.assertEqual(mdc_formatter._mdcFmt, "{reqeustID}") + + mdc_formatter = MDCFormatter(mdcfmt="{test}") + self.assertEqual(mdc_formatter.style, "%") + self.assertEqual(mdc_formatter._mdc_tag, "%(mdc)s") + self.assertEqual(mdc_formatter._mdcFmt, "{test}") + + if sys.version_info[0] >= 3: + mdc_formatter = MDCFormatter(style="{") + self.assertEqual(mdc_formatter.style, "{") + self.assertEqual(mdc_formatter._mdc_tag, "{mdc}") + self.assertEqual(mdc_formatter._mdcFmt, "{reqeustID}") + + mdc_formatter = MDCFormatter(style="$") + self.assertEqual(mdc_formatter.style, "$") + self.assertEqual(mdc_formatter._mdc_tag, "${mdc}") + self.assertEqual(mdc_formatter._mdcFmt, "{reqeustID}") + + with pytest.raises(ValueError): + MDCFormatter(style="*") + + def test_mdc_fmt_key(self): + mdc_formatter = MDCFormatter() + brace, not_brace = mdc_formatter._mdcfmtKey() + self.assertEqual(brace, ["{reqeustID}"]) + self.assertEqual(list(not_brace), ["reqeustID"]) + + mdc_formatter = MDCFormatter(mdcfmt="{test} {value} {anything}") + brace, not_brace = mdc_formatter._mdcfmtKey() + self.assertEqual(brace, ["{test}", "{value}", "{anything}"]) + self.assertEqual(list(not_brace), ["test", "value", "anything"]) + + mdc_formatter = MDCFormatter(mdcfmt="no_braces") + brace, not_brace = mdc_formatter._mdcfmtKey() + self.assertEqual(brace, []) + self.assertIsNone(not_brace) + + mdc_formatter = MDCFormatter(mdcfmt="}what?{") + brace, not_brace = mdc_formatter._mdcfmtKey() + self.assertEqual(brace, []) + self.assertIsNone(not_brace) + + mdc_formatter = MDCFormatter(mdcfmt="}{hello}{") + brace, not_brace = mdc_formatter._mdcfmtKey() + self.assertEqual(brace, ["{hello}"]) + self.assertEqual(list(not_brace), ["hello"]) + + mdc_formatter = MDCFormatter(mdcfmt="}{}{hel{lo}{") + brace, not_brace = mdc_formatter._mdcfmtKey() + self.assertEqual(brace, []) + self.assertIsNone(not_brace) + + def test_format(self): + record = MagicMock() + with patch("onaplogging.mdcformatter.MarkerFormatter.format") as mock_marker_formatter_format: + mdc_formatter = MDCFormatter() + mdc_formatter.format(record) + mock_marker_formatter_format.assert_called_once_with(record) + self.assertEqual(mdc_formatter._fmt, "%(message)s") + + if sys.version_info[0] >= 3: + with patch("onaplogging.mdcformatter.MarkerFormatter.format") as mock_marker_formatter_format: + mdc_formatter = MDCFormatter(fmt="{mdc}", style="{", mdcfmt="{key}") + mdc_formatter.format(record) + mock_marker_formatter_format.assert_called_once_with(record) + self.assertEqual(mdc_formatter._fmt, "key=") + + record.mdc = {"key": 123} + with patch("onaplogging.mdcformatter.MarkerFormatter.format") as mock_marker_formatter_format: + mdc_formatter = MDCFormatter(fmt="{mdc}", style="{", mdcfmt="no_braces") + mdc_formatter.format(record) + mock_marker_formatter_format.assert_called_once_with(record) + self.assertEqual(mdc_formatter._fmt, "") + + with patch("onaplogging.mdcformatter.MarkerFormatter.format") as mock_marker_formatter_format: + mdc_formatter = MDCFormatter(fmt="{mdc}", style="{", mdcfmt="{key}") + mdc_formatter.format(record) + mock_marker_formatter_format.assert_called_once_with(record) + self.assertEqual(mdc_formatter._fmt, "key=123") diff --git a/pylog/tests/test_monkey.py b/pylog/tests/test_monkey.py new file mode 100644 index 0000000..4f71fe2 --- /dev/null +++ b/pylog/tests/test_monkey.py @@ -0,0 +1,47 @@ +# Copyright (c) 2020 Deutsche Telekom. +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + +import sys +import unittest + +if sys.version_info[0] < 3: + from mock import patch +if sys.version_info[0] >= 3: + from unittest.mock import patch + +from onaplogging.monkey import patch_all, patch_loggingMDC, patch_loggingYaml + + +class TestMonkey(unittest.TestCase): + + def test_patch_all(self): + with patch("onaplogging.monkey.patch_loggingMDC") as mock_mdc: + with patch("onaplogging.monkey.patch_loggingYaml") as mock_yaml: + patch_all() + mock_mdc.assert_called_once() + mock_yaml.assert_called_once() + + with patch("onaplogging.monkey.patch_loggingMDC") as mock_mdc: + with patch("onaplogging.monkey.patch_loggingYaml") as mock_yaml: + patch_all(mdc=False) + mock_mdc.assert_not_called() + mock_yaml.assert_called_once() + + with patch("onaplogging.monkey.patch_loggingMDC") as mock_mdc: + with patch("onaplogging.monkey.patch_loggingYaml") as mock_yaml: + patch_all(yaml=False) + mock_mdc.assert_called_once() + mock_yaml.assert_not_called() + + with patch("onaplogging.monkey.patch_loggingMDC") as mock_mdc: + with patch("onaplogging.monkey.patch_loggingYaml") as mock_yaml: + patch_all(mdc=False, yaml=False) + mock_mdc.assert_not_called() + mock_yaml.assert_not_called() diff --git a/pylog/tests/test_utils.py b/pylog/tests/test_utils.py new file mode 100644 index 0000000..5a64aab --- /dev/null +++ b/pylog/tests/test_utils.py @@ -0,0 +1,45 @@ +# Copyright (c) 2020 Deutsche Telekom. +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + +import sys + +import unittest +if sys.version_info[0] < 3: + from mock import patch, MagicMock +if sys.version_info[0] >= 3: + from unittest.mock import patch, MagicMock + +from onaplogging.utils import is_above_python_2_7, is_above_python_3_2 + + +class TestUtils(unittest.TestCase): + + def test_is_above_python_3_2(self): + with patch("onaplogging.utils.sys.version_info", (3, 4, 7)): + assert is_above_python_3_2() is True + + with patch("onaplogging.utils.sys.version_info", (2, 7, 5)): + assert is_above_python_3_2() is False + + with patch("onaplogging.utils.sys.version_info", (3, 2, 0, "final", 0)): + assert is_above_python_3_2() is True + + def test_is_above_python_2_7(self): + with patch("onaplogging.utils.sys.version_info", (3, 4, 7)): + assert is_above_python_2_7() is True + + with patch("onaplogging.utils.sys.version_info", (2, 7, 5)): + assert is_above_python_2_7() is True + + with patch("onaplogging.utils.sys.version_info", (2, 5, 6)): + assert is_above_python_2_7() is False + + with patch("onaplogging.utils.sys.version_info", (2, 7, 0, "final", 0)): + assert is_above_python_2_7() is True diff --git a/pylog/tox.ini b/pylog/tox.ini index 87209a1..3f92b52 100644 --- a/pylog/tox.ini +++ b/pylog/tox.ini @@ -14,8 +14,7 @@ downloadcache = ~/cache/pip [testenv] basepython = python3 -deps = -r{toxinidir}/requirements.txt - pytest +deps = -r{toxinidir}/test_requirements.txt coverage pytest-cov setenv = PYTHONPATH={toxinidir}/ @@ -25,7 +24,7 @@ commands = pytest [testenv:pep8] basepython = python3 deps=flake8 -commands=flake8 +commands=flake8 onaplogging [flake8] basepython = python3 @@ -34,4 +33,4 @@ exclude = env,venv,.venv,.git,.tox,dist,doc,*egg,build [testenv:cover] basepython = python3 -commands = py.test --cov onaplogging +commands = py.test --cov-report term-missing --cov onaplogging |