X-Git-Url: https://gerrit.akraino.org/r/gitweb?a=blobdiff_plain;f=cmframework%2Ftest%2Fcmupdateimpl_test.py;fp=cmframework%2Ftest%2Fcmupdateimpl_test.py;h=40763576b3a83f28a53d6311de1a1a5289e5b0f9;hb=c389bdee7b3845b55f443dbf04c0ce4083a55886;hp=0000000000000000000000000000000000000000;hpb=5030f0c004701dd422c78c71c014ef60f48139fc;p=ta%2Fconfig-manager.git diff --git a/cmframework/test/cmupdateimpl_test.py b/cmframework/test/cmupdateimpl_test.py new file mode 100644 index 0000000..4076357 --- /dev/null +++ b/cmframework/test/cmupdateimpl_test.py @@ -0,0 +1,318 @@ +# Copyright 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. + +import unittest +import mock +from mock import call +import json + +from cmframework.apis.cmupdate import CMUpdate +from cmframework.lib.cmupdateimpl import CMUpdateImpl +from cmframework.apis.cmerror import CMError + + +class CMUpdateImplTest(unittest.TestCase): + @mock.patch('cmframework.lib.cmupdateimpl.CMPluginLoader') + @mock.patch('cmframework.lib.cmupdateimpl.CMManage') + @mock.patch('cmframework.lib.cmupdateimpl.logging') + def test_init(self, mock_logging, mock_client, mock_pluginloader): + mock_pluginloader.return_value.load.return_value = ({}, None) + + updater = CMUpdate('test_plugin_path', 'test_server_ip', 'test_server_port', + 'test_client_lib_impl_module', 'test_verbose_logger') + + mock_pluginloader.assert_called_once_with('test_plugin_path') + mock_client.assert_called_once_with('test_server_ip', + 'test_server_port', + 'test_client_lib_impl_module', + 'test_verbose_logger') + + @staticmethod + def _test__read_dependency_file_incorrect(file_name): + if file_name == 'test_plugin_path/test_handler_a.deps': + return ([], ['x']) + if file_name == 'test_plugin_path/test_handler_b.deps': + return (['y'], []) + + @mock.patch.object(CMUpdateImpl, '_read_dependency_file') + @mock.patch('cmframework.lib.cmupdateimpl.CMPluginLoader') + @mock.patch('cmframework.lib.cmupdateimpl.CMManage') + @mock.patch('cmframework.lib.cmupdateimpl.logging') + def test_init_missing_handler_in_dependencies(self, + mock_logging, + mock_client, + mock_pluginloader, + mock__read_dependency_file): + mock_pluginloader.return_value.load.return_value = ({}, None) + mock__read_dependency_file.side_effect = \ + CMUpdateImplTest._test__read_dependency_file_incorrect + + test_handler_a_module = mock.MagicMock() + test_handler_a_class = mock.MagicMock() + test_handler_a_class.return_value.__str__.return_value = 'test_handler_a' + setattr(test_handler_a_module, 'test_handler_a', test_handler_a_class) + + mock_pluginloader.return_value.load.return_value = \ + ({'test_handler_a': test_handler_a_module}, None) + + with self.assertRaises(CMError) as context: + updater = CMUpdate('test_plugin_path', 'test_server_ip', 'test_server_port', + 'test_client_lib_impl_module', 'test_verbose_logger') + + test_handler_b_module = mock.MagicMock() + test_handler_b_class = mock.MagicMock() + test_handler_b_class.return_value.__str__.return_value = 'test_handler_b' + setattr(test_handler_b_module, 'test_handler_b', test_handler_b_class) + + mock_pluginloader.return_value.load.return_value = \ + ({'test_handler_b': test_handler_b_module}, None) + + with self.assertRaises(CMError) as context: + updater = CMUpdate('test_plugin_path', 'test_server_ip', 'test_server_port', + 'test_client_lib_impl_module', 'test_verbose_logger') + + mock_client.assert_not_called() + + @staticmethod + def _test__read_dependency_file(file_name): + if file_name == 'test_plugin_path/test_handler_a.deps': + return ([], ['test_handler_b']) + if file_name == 'test_plugin_path/test_handler_b.deps': + return (['test_handler_a'], []) + if file_name == 'test_plugin_path/test_handler_c.deps': + return (['test_handler_b'], ['test_handler_a']) + + @staticmethod + def _test_update_func_a(confman): + CMUpdateImplTest._test_update_func_calls.append('test_handler_a') + if str(confman) == 'raise exception': + raise Exception('test_update_exception') + + @staticmethod + def _test_update_func_b(confman): + CMUpdateImplTest._test_update_func_calls.append('test_handler_b') + + @staticmethod + def _test_update_func_c(confman): + confman.get_config.return_value = {'test_properties': '_test_update_func_c properties'} + CMUpdateImplTest._test_update_func_calls.append('test_handler_c') + + def _setup_test_handlers(self, mock_pluginloader, mock__read_dependency_file, mock_sorter): + CMUpdateImplTest._test_update_func_calls = [] + + test_handler_a_module = mock.MagicMock() + test_handler_a_class = mock.MagicMock() + test_handler_a_class.return_value.__str__.return_value = 'test_handler_a' + test_handler_a_class.return_value.update.side_effect = CMUpdateImplTest._test_update_func_a + setattr(test_handler_a_module, 'test_handler_a', test_handler_a_class) + + test_handler_b_module = mock.MagicMock() + test_handler_b_class = mock.MagicMock() + test_handler_b_class.return_value.__str__.return_value = 'test_handler_b' + test_handler_b_class.return_value.update.side_effect = CMUpdateImplTest._test_update_func_b + setattr(test_handler_b_module, 'test_handler_b', test_handler_b_class) + + test_handler_c_module = mock.MagicMock() + test_handler_c_class = mock.MagicMock() + test_handler_c_class.return_value.__str__.return_value = 'test_handler_c' + test_handler_c_class.return_value.update.side_effect = CMUpdateImplTest._test_update_func_c + setattr(test_handler_c_module, 'test_handler_c', test_handler_c_class) + + mock_pluginloader.return_value.load.return_value = \ + ({'test_handler_a': test_handler_a_module, + 'test_handler_b': test_handler_b_module, + 'test_handler_c': test_handler_c_module}, None) + + mock__read_dependency_file.side_effect = CMUpdateImplTest._test__read_dependency_file + + mock_sorter.return_value.sort.return_value = ['test_handler_c', + 'test_handler_a', + 'test_handler_b'] + + return (test_handler_a_class, test_handler_b_class, test_handler_c_class) + + @mock.patch.object(CMUpdateImpl, '_read_dependency_file') + @mock.patch('cmframework.lib.cmupdateimpl.CMDependencySort') + @mock.patch('cmframework.lib.cmupdateimpl.CMPluginLoader') + @mock.patch('cmframework.lib.cmupdateimpl.CMManage') + @mock.patch('cmframework.lib.cmupdateimpl.logging') + def test_update(self, + mock_logging, + mock_client, + mock_pluginloader, + mock_sorter, + mock__read_dependency_file): + test_handler_a_class, test_handler_b_class, test_handler_c_class = \ + self._setup_test_handlers(mock_pluginloader, mock__read_dependency_file, mock_sorter) + + updater = CMUpdate('test_plugin_path', 'test_server_ip', 'test_server_port', + 'test_client_lib_impl_module', 'test_verbose_logger') + + mock_confman = mock.MagicMock() + mock_confman.__str__.return_value = 'confman' + mock_confman.get_config.return_value = {'test_properties': 'some properties'} + + updater.update(mock_confman) + + sorter_after_graph = mock_sorter.call_args[0][0] + assert len(sorter_after_graph) == 3 + assert 'test_handler_a' in sorter_after_graph + assert 'test_handler_b' in sorter_after_graph + assert 'test_handler_c' in sorter_after_graph + assert sorter_after_graph['test_handler_a'] == ['test_handler_b'] + assert sorter_after_graph['test_handler_b'] == [] + assert sorter_after_graph['test_handler_c'] == ['test_handler_a'] + + sorter_before_graph = mock_sorter.call_args[0][1] + assert len(sorter_before_graph) == 3 + assert 'test_handler_a' in sorter_before_graph + assert 'test_handler_b' in sorter_before_graph + assert 'test_handler_c' in sorter_before_graph + assert sorter_before_graph['test_handler_a'] == [] + assert sorter_before_graph['test_handler_b'] == ['test_handler_a'] + assert sorter_before_graph['test_handler_c'] == ['test_handler_b'] + + mock_sorter.return_value.sort.assert_called_once() + + test_handler_a_class.return_value.update.assert_called_once_with(mock_confman) + test_handler_b_class.return_value.update.assert_called_once_with(mock_confman) + test_handler_c_class.return_value.update.assert_called_once_with(mock_confman) + + assert CMUpdateImplTest._test_update_func_calls == \ + mock_sorter.return_value.sort.return_value + + mock_client.return_value.set_properties.assert_called_once_with( + {'test_properties': '_test_update_func_c properties'}, True) + + @mock.patch.object(CMUpdateImpl, '_read_dependency_file') + @mock.patch('cmframework.lib.cmupdateimpl.ConfigManager') + @mock.patch('cmframework.lib.cmupdateimpl.CMDependencySort') + @mock.patch('cmframework.lib.cmupdateimpl.CMPluginLoader') + @mock.patch('cmframework.lib.cmupdateimpl.CMManage') + @mock.patch('cmframework.lib.cmupdateimpl.logging') + def test_update_no_confman(self, + mock_logging, + mock_client, + mock_pluginloader, + mock_sorter, + mock_configmanager, + mock__read_dependency_file): + test_handler_a_class, test_handler_b_class, test_handler_c_class = \ + self._setup_test_handlers(mock_pluginloader, mock__read_dependency_file, mock_sorter) + + updater = CMUpdate('test_plugin_path', 'test_server_ip', 'test_server_port', + 'test_client_lib_impl_module', 'test_verbose_logger') + + updater.update() + + test_handler_a_class.return_value.update.assert_called_once_with( + mock_configmanager.return_value) + test_handler_b_class.return_value.update.assert_called_once_with( + mock_configmanager.return_value) + test_handler_c_class.return_value.update.assert_called_once_with( + mock_configmanager.return_value) + + ''' + mock_configmanager.assert_called_once_with( + mock_client.return_value.get_properties.return_value) + ''' + + assert CMUpdateImplTest._test_update_func_calls == \ + mock_sorter.return_value.sort.return_value + + @mock.patch.object(CMUpdateImpl, '_read_dependency_file') + @mock.patch('cmframework.lib.cmupdateimpl.ConfigManager') + @mock.patch('cmframework.lib.cmupdateimpl.CMDependencySort') + @mock.patch('cmframework.lib.cmupdateimpl.CMPluginLoader') + @mock.patch('cmframework.lib.cmupdateimpl.CMManage') + @mock.patch('cmframework.lib.cmupdateimpl.logging') + def test_update_exception(self, + mock_logging, + mock_client, + mock_pluginloader, + mock_sorter, + mock_configmanager, + mock__read_dependency_file): + test_handler_a_class, test_handler_b_class, test_handler_c_class = \ + self._setup_test_handlers(mock_pluginloader, mock__read_dependency_file, mock_sorter) + + updater = CMUpdate('test_plugin_path', 'test_server_ip', 'test_server_port', + 'test_client_lib_impl_module', 'test_verbose_logger') + + mock_confman = mock.MagicMock() + mock_confman.__str__.return_value = 'raise exception' + + with self.assertRaises(Exception) as context: + updater.update(mock_confman) + + # TODO:verify plugin(s) before exception are called. + mock_logging.warning.assert_called_with('Update handler %s failed: %s', + 'test_handler_a', + 'test_update_exception') + + @mock.patch('cmframework.lib.cmupdateimpl.logging') + def test_dependency_files(self, mock_logging): + with mock.patch('cmframework.lib.cmupdateimpl.open', create=True) as mock_open: + mock_open.return_value = mock.MagicMock(spec=file) + file_handle = mock_open.return_value.__enter__.return_value + + file_handle.readline.side_effect = IOError('File not found') + before, after = CMUpdateImpl._read_dependency_file('testexception: not existing') + assert before == [] + assert after == [] + mock_logging.debug.assert_called_with('Dependency file %s not found.', + 'testexception: not existing') + + file_handle.readline.side_effect = [None] + before, after = CMUpdateImpl._read_dependency_file('./test_deps/1.deps') + assert before == [] + assert after == [] + + file_handle.readline.side_effect = ['foo', 'bar', None] + before, after = CMUpdateImpl._read_dependency_file('./test_deps/1.deps') + assert before == [] + assert after == [] + + file_handle.readline.side_effect = ['Before: a, b', 'After: c, d', None] + before, after = CMUpdateImpl._read_dependency_file('./test_deps/1.deps') + assert before == ['a', 'b'] + assert after == ['c', 'd'] + + file_handle.readline.side_effect = ['foo', + 'Before: a, b', + 'bar', + 'After: c, d', + 'something', + None] + before, after = CMUpdateImpl._read_dependency_file('./test_deps/1.deps') + assert before == ['a', 'b'] + assert after == ['c', 'd'] + + file_handle.readline.side_effect = ['After: c, d', 'Before: a, b', None] + before, after = CMUpdateImpl._read_dependency_file('./test_deps/1.deps') + assert before == ['a', 'b'] + assert after == ['c', 'd'] + + file_handle.readline.side_effect = ['Before:a,b', 'After:c,d', None] + before, after = CMUpdateImpl._read_dependency_file('./test_deps/1.deps') + assert before == ['a', 'b'] + assert after == ['c', 'd'] + + file_handle.readline.side_effect = ['Before: a, b ', 'After: c, d ', None] + before, after = CMUpdateImpl._read_dependency_file('./test_deps/1.deps') + assert before == ['a', 'b'] + assert after == ['c', 'd'] + +if __name__ == '__main__': + unittest.main()