Prepare test suite for onaplogging library.
Add utils to check Python version.
Issue-ID: LOG-1229
Signed-off-by: Michal Jagiello <michal.jagiello@t-mobile.pl>
Change-Id: I12c431c10f61fceaf70d5ae36581dfd5a876ef72
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,
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):
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
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):
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)
def remove(self, key):
- if key in self.localDict:
+ if key in self._localDict:
del self._localDict[key]
def clear(self):
# 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):
: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)
"""
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)
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)
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)
--- /dev/null
+# 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)
--- /dev/null
+-r requirements.txt
+pytest
+mock
\ No newline at end of file
--- /dev/null
+# 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
+ )
+++ /dev/null
-# 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
--- /dev/null
+# 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()
--- /dev/null
+# 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))
--- /dev/null
+# 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")
--- /dev/null
+# 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")
--- /dev/null
+# 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)
--- /dev/null
+# 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")
--- /dev/null
+# 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()
--- /dev/null
+# 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
[testenv]
basepython = python3
-deps = -r{toxinidir}/requirements.txt
- pytest
+deps = -r{toxinidir}/test_requirements.txt
coverage
pytest-cov
setenv = PYTHONPATH={toxinidir}/
[testenv:pep8]
basepython = python3
deps=flake8
-commands=flake8
+commands=flake8 onaplogging
[flake8]
basepython = python3
[testenv:cover]
basepython = python3
-commands = py.test --cov onaplogging
+commands = py.test --cov-report term-missing --cov onaplogging
Code Review / logging-analytics.git / commitdiff