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.
14 from collections import namedtuple
15 from tempfile import NamedTemporaryFile
17 if sys.version_info[0] < 3:
18 from mock import patch
19 if sys.version_info[0] >= 3:
20 from unittest.mock import patch
25 from onaplogging.logWatchDog import FileEventHandlers, _yamlConfig
26 from onaplogging.utils.tools import yaml_to_dict, _yaml2Dict
29 TestEvent = namedtuple("TestEvent", ["src_path"])
32 class TestLogWatchdog(unittest.TestCase):
41 super(TestLogWatchdog, self).setUp()
43 self.temp_file = NamedTemporaryFile(mode="w+t", delete=False)
44 self.temp_file.write(yaml.dump(self.TEST_DICT))
45 self.temp_file.close()
48 super(TestLogWatchdog, self).tearDown()
50 os.unlink(self.temp_file.name)
52 def test_yaml2dict(self):
53 with pytest.raises(TypeError):
56 self.assertDictEqual(self.TEST_DICT, _yaml2Dict(self.temp_file.name))
58 def test_file_event_handler(self):
60 with patch("onaplogging.logWatchDog.config.dictConfig") as mock_config:
61 mock_config.side_effect = Exception
63 feh = FileEventHandlers(self.temp_file.name)
64 self.assertIsNone(feh.currentConfig)
65 feh.on_modified(TestEvent(src_path=self.temp_file.name))
66 self.assertIsNone(feh.currentConfig)
68 with patch("onaplogging.logWatchDog.config"):
70 feh = FileEventHandlers(self.temp_file.name)
71 self.assertIsNone(feh.currentConfig)
72 feh.on_modified(TestEvent(src_path=self.temp_file.name))
73 self.assertIsNotNone(feh.currentConfig)
75 def test_patch_yaml_config(self):
77 with pytest.raises(TypeError):
78 _yamlConfig(filepath=None)
80 with pytest.raises(OSError):
81 _yamlConfig(filepath="invalid path")
83 with patch("onaplogging.logWatchDog.config.dictConfig") as mock_config:
84 _yamlConfig(filepath=self.temp_file.name)
85 mock_config.assert_called_once_with(self.TEST_DICT)
87 with patch("onaplogging.logWatchDog.config.dictConfig") as mock_config:
88 with patch("onaplogging.logWatchDog.Observer.start") as mock_observer_start:
89 _yamlConfig(filepath=self.temp_file.name, watchDog=True)
90 mock_config.assert_called_once_with(self.TEST_DICT)
91 mock_observer_start.assert_called_once()