--- /dev/null
+# Copyright (c) 2019 AT&T Intellectual Property.
+# Copyright (c) 2018-2019 Nokia.
+#
+# 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.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# This source code is part of the near-RT RIC (RAN Intelligent Controller)
+# platform project (RICP).
+#
+"""Unit tests for Logger.py"""
+import unittest
+from unittest.mock import patch
+
+from ricxappframe.logger.mdclogger import Level, MDCLogger
+from .mdclogtestutils import TestMdcLogUtils
+
+
+class TestMdcLog(unittest.TestCase):
+ """Unit tests for mdclog.py"""
+
+ def setUp(self):
+ self.logger = MDCLogger()
+
+ def tearDown(self):
+ pass
+
+ def test_that_get_level_returns_the_current_log_level(self):
+
+ # default level is ERROR
+ self.assertEqual(self.logger.get_level(), Level.ERROR)
+ self.logger.set_level(Level.INFO)
+ self.assertEqual(self.logger.get_level(), Level.INFO)
+ self.logger.set_level(Level.WARNING)
+ self.assertEqual(self.logger.get_level(), Level.WARNING)
+ self.logger.set_level(Level.ERROR)
+ self.assertEqual(self.logger.get_level(), Level.ERROR)
+ self.logger.set_level(Level.DEBUG)
+ self.assertEqual(self.logger.get_level(), Level.DEBUG)
+
+ def test_that_set_level_does_not_accept_incorrect_level(self):
+
+ self.logger.set_level(Level.INFO)
+ self.logger.set_level(55)
+ self.assertEqual(self.logger.get_level(), Level.INFO)
+
+ @patch('ricxappframe.logger.mdclogger.MDCLogger._output_log')
+ def test_that_logs_with_lower_than_current_level_(self, output_mock):
+
+ self.logger.set_level(Level.WARNING)
+ self.logger.log(Level.DEBUG, "DEBUG")
+ self.logger.log(Level.INFO, "INFO")
+ self.logger.log(Level.WARNING, "WARNING")
+ self.logger.log(Level.ERROR, "ERROR")
+
+ self.assertEqual(2, output_mock.call_count)
+ logs = TestMdcLogUtils.get_logs_as_json(output_mock.call_args_list)
+ self.assertEqual(logs[0]["msg"], "WARNING")
+ self.assertEqual(logs[1]["msg"], "ERROR")
+
+ @patch('ricxappframe.logger.mdclogger.MDCLogger._output_log')
+ def test_that_logs_with_lower_than_current_level_are_not_logged(self, output_mock):
+
+ self.logger.set_level(Level.WARNING)
+ self.logger.log(Level.DEBUG, "DEBUG")
+ self.logger.log(Level.INFO, "INFO")
+ self.logger.log(Level.WARNING, "WARNING")
+ self.logger.log(Level.ERROR, "ERROR")
+
+ self.assertEqual(2, output_mock.call_count)
+ logs = TestMdcLogUtils.get_logs_as_json(output_mock.call_args_list)
+ self.assertEqual(logs[0]["msg"], "WARNING")
+ self.assertEqual(logs[1]["msg"], "ERROR")
+
+ @patch('ricxappframe.logger.mdclogger.MDCLogger._output_log')
+ def test_that_log_contains_correct_criticality(self, output_mock):
+
+ self.logger.set_level(Level.DEBUG)
+
+ self.logger.log(Level.DEBUG, "debug test log")
+ self.logger.log(Level.INFO, "info test log")
+ self.logger.log(Level.WARNING, "warning test log")
+ self.logger.log(Level.ERROR, "error test log")
+
+ self.logger.debug("another debug test log")
+ self.logger.info("another info test log")
+ self.logger.warning("another warning test log")
+ self.logger.error("another error test log")
+
+ self.assertEqual(8, output_mock.call_count)
+ logs = TestMdcLogUtils.get_logs_as_json(output_mock.call_args_list)
+ self.assertEqual(logs[0]["crit"], "DEBUG")
+ self.assertEqual(logs[1]["crit"], "INFO")
+ self.assertEqual(logs[2]["crit"], "WARNING")
+ self.assertEqual(logs[3]["crit"], "ERROR")
+ self.assertEqual(logs[4]["crit"], "DEBUG")
+ self.assertEqual(logs[5]["crit"], "INFO")
+ self.assertEqual(logs[6]["crit"], "WARNING")
+ self.assertEqual(logs[7]["crit"], "ERROR")
+
+ @patch('time.time')
+ @patch('ricxappframe.logger.mdclogger.MDCLogger._output_log')
+ def test_that_log_contains_correct_timestamp(self, output_mock, mock_time):
+
+ mock_time.return_value = 1554806251.4388545
+ self.logger.error("timestamp test")
+
+ logs = TestMdcLogUtils.get_logs_as_json(output_mock.call_args_list)
+ self.assertEqual(logs[0]["ts"], 1554806251439)
+
+ @patch('ricxappframe.logger.mdclogger.MDCLogger._output_log')
+ def test_that_log_contains_correct_message(self, output_mock):
+
+ self.logger.error("message test")
+ logs = TestMdcLogUtils.get_logs_as_json(output_mock.call_args_list)
+ print(logs)
+ self.assertEqual(logs[0]["msg"], "message test")
+
+ @patch('ricxappframe.logger.mdclogger.MDCLogger._output_log')
+ def test_that_log_message_is_escaped_to_valid_json_string(self, output_mock):
+
+ self.logger.set_level(Level.DEBUG)
+
+ self.logger.info(r'\ and "')
+
+ logs = TestMdcLogUtils.get_logs(output_mock.call_args_list)
+ self.assertTrue(r'\\ and \"' in logs[0])
+ logs = TestMdcLogUtils.get_logs_as_json(output_mock.call_args_list)
+ self.assertEqual(logs[0]["msg"], r'\ and "')
+
+ @patch('ricxappframe.logger.mdclogger.MDCLogger._output_log')
+ def test_that_empty_mdc_is_logged_correctly(self, output_mock):
+ self.logger.mdclog_format_init(configmap_monitor=True)
+ self.logger.error("empty mdc test")
+ self.logger.error(output_mock.call_args_list)
+ logs = TestMdcLogUtils.get_logs_as_json(output_mock.call_args_list)
+ self.assertEqual(logs[0]["msg"], 'empty mdc test')
+
+ @patch('ricxappframe.logger.mdclogger.MDCLogger._output_log')
+ def test_that_config_map_is_monitored_correctly(self, output_mock):
+ src = open("//tmp//log", "w")
+ src.write("log-level: debug\n")
+ src.close()
+ self.logger.filename = "/tmp/log"
+ self.logger.dirname = "/tmp/"
+ self.logger.mdc = {"PID": "", "SYSTEM_NAME": "", "HOST_NAME": "", "SERVICE_NAME": "", "CONTAINER_NAME": "", "POD_NAME": ""}
+ self.logger.get_env_params_values()
+ self.logger.parse_file()
+ self.logger.error("Hello")
+ self.assertEqual(self.logger.get_level(), Level.DEBUG)
+
+ @patch('ricxappframe.logger.mdclogger.MDCLogger._output_log')
+ def test_that_mdc_values_are_logged_correctly(self, output_mock):
+
+ self.logger.add_mdc("key1", "value1")
+ self.logger.add_mdc("key2", "value2")
+ self.logger.error("mdc test")
+
+ logs = TestMdcLogUtils.get_logs_as_json(output_mock.call_args_list)
+ self.assertEqual(logs[0]["mdc"]["key1"], "value1")
+ self.assertEqual(logs[0]["mdc"]["key2"], "value2")
+
+ @patch('ricxappframe.logger.mdclogger.MDCLogger._output_log')
+ def test_that_mdc_pid_logged_correctly(self, output_mock):
+ self.logger.mdclog_format_init(configmap_monitor=True)
+ self.logger.error("mdc test")
+ logs = TestMdcLogUtils.get_logs_as_json(output_mock.call_args_list)
+ self.assertTrue(logs[0]["mdc"]["PID"])
+
+ def test_that_mdc_values_can_be_added_and_removed(self):
+
+ self.logger.add_mdc("key1", "value1")
+ self.logger.add_mdc("key2", "value2")
+ self.assertEqual(self.logger.get_mdc("key2"), "value2")
+ self.assertEqual(self.logger.get_mdc("key1"), "value1")
+ self.assertEqual(self.logger.get_mdc("non_existent"), None)
+ self.logger.remove_mdc("key1")
+ self.assertEqual(self.logger.get_mdc("key1"), None)
+ self.logger.remove_mdc("non_existent")
+ self.logger.clean_mdc()
+ self.assertEqual(self.logger.get_mdc("key2"), None)
+
+ @patch('ricxappframe.logger.mdclogger.MDCLogger._output_log')
+ def test_multiple_logger_instances(self, output_mock):
+
+ logger1 = MDCLogger("logger1")
+ logger2 = MDCLogger("logger2")
+ logger1.add_mdc("logger1_key1", "logger1_value1")
+ logger1.add_mdc("logger1_key2", "logger1_value2")
+ logger2.add_mdc("logger2_key1", "logger2_value1")
+ logger2.add_mdc("logger2_key2", "logger2_value2")
+
+ logger1.error("error msg")
+ logger2.error("warning msg")
+
+ logs = TestMdcLogUtils.get_logs_as_json(output_mock.call_args_list)
+ self.assertEqual(2, output_mock.call_count)
+
+ self.assertEqual(logs[0]["id"], "logger1")
+ self.assertEqual(logs[0]["crit"], "ERROR")
+ self.assertEqual(logs[0]["msg"], "error msg")
+ self.assertEqual(logs[0]["mdc"]["logger1_key1"], "logger1_value1")
+ self.assertEqual(logs[0]["mdc"]["logger1_key2"], "logger1_value2")
+ self.assertEqual(len(logs[0]["mdc"]), 2)
+
+ self.assertEqual(logs[1]["id"], "logger2")
+ self.assertEqual(logs[1]["crit"], "ERROR")
+ self.assertEqual(logs[1]["msg"], "warning msg")
+ self.assertEqual(logs[1]["mdc"]["logger2_key1"], "logger2_value1")
+ self.assertEqual(logs[1]["mdc"]["logger2_key2"], "logger2_value2")
+ self.assertEqual(len(logs[1]["mdc"]), 2)
+
+
+if __name__ == '__main__':
+ unittest.main()
Code Review / ric-plt / xapp-frame-py.git / blobdiff