#!/usr/bin/env python # Copyright 2016 NEC Corporation # 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 imp import os import sys from unittest import mock from oslotest import base sys.modules['dbus'] = mock.MagicMock() this_dir = os.path.dirname(sys.modules[__name__].__file__) ansible_dir = os.path.join(this_dir, '..', '..', 'ansible') systemd_worker_file = os.path.join(ansible_dir, 'module_utils', 'kolla_systemd_worker.py') swm = imp.load_source('kolla_systemd_worker', systemd_worker_file) class TestSystemd(base.BaseTestCase): def setUp(self) -> None: super(TestSystemd, self).setUp() self.params_dict = dict( name='test', restart_policy='no', client_timeout=120, restart_retries=10, graceful_timeout=15 ) swm.sleep = mock.Mock() self.sw = swm.SystemdWorker(self.params_dict) def test_manager(self): self.assertIsNotNone(self.sw) self.assertIsNotNone(self.sw.manager) def test_start(self): self.sw.perform_action = mock.Mock(return_value=True) self.sw.wait_for_unit = mock.Mock(return_value=True) self.sw.start() self.sw.perform_action.assert_called_once_with( 'StartUnit', 'kolla-test-container.service', 'replace' ) def test_restart(self): self.sw.perform_action = mock.Mock(return_value=True) self.sw.wait_for_unit = mock.Mock(return_value=True) self.sw.restart() self.sw.perform_action.assert_called_once_with( 'RestartUnit', 'kolla-test-container.service', 'replace' ) def test_stop(self): self.sw.perform_action = mock.Mock(return_value=True) self.sw.stop() self.sw.perform_action.assert_called_once_with( 'StopUnit', 'kolla-test-container.service', 'replace' ) def test_reload(self): self.sw.perform_action = mock.Mock(return_value=True) self.sw.reload() self.sw.perform_action.assert_called_once_with( 'Reload', 'kolla-test-container.service', 'replace' ) def test_enable(self): self.sw.perform_action = mock.Mock(return_value=True) self.sw.enable() self.sw.perform_action.assert_called_once_with( 'EnableUnitFiles', ['kolla-test-container.service'], False, True ) def test_check_unit_change(self): self.sw.generate_unit_file = mock.Mock() self.sw.check_unit_file = mock.Mock(return_value=True) open_mock = mock.mock_open(read_data='test data') return_val = None with mock.patch('builtins.open', open_mock, create=True): return_val = self.sw.check_unit_change('test data') self.assertFalse(return_val) self.sw.generate_unit_file.assert_not_called() open_mock.assert_called_with( '/etc/systemd/system/kolla-test-container.service', 'r' ) open_mock.return_value.read.assert_called_once() def test_check_unit_change_diff(self): self.sw.generate_unit_file = mock.Mock() self.sw.check_unit_file = mock.Mock(return_value=True) open_mock = mock.mock_open(read_data='new data') return_val = None with mock.patch('builtins.open', open_mock, create=True): return_val = self.sw.check_unit_change('old data') self.assertTrue(return_val) self.sw.generate_unit_file.assert_not_called() open_mock.assert_called_with( '/etc/systemd/system/kolla-test-container.service', 'r' ) open_mock.return_value.read.assert_called_once() @mock.patch( 'kolla_systemd_worker.TEMPLATE', """${name}, ${restart_policy}, ${graceful_timeout}, ${restart_timeout}, ${restart_retries}""" ) def test_generate_unit_file(self): self.sw = swm.SystemdWorker(self.params_dict) p = self.params_dict ref_string = f"""{p.get('name')}, {p.get('restart_policy')}, {p.get('graceful_timeout')}, {p.get('client_timeout')}, {p.get('restart_retries')}""" ret_string = self.sw.generate_unit_file() self.assertEqual(ref_string, ret_string) def test_create_unit_file(self): self.sw.generate_unit_file = mock.Mock(return_value='test data') self.sw.check_unit_change = mock.Mock(return_value=True) self.sw.reload = mock.Mock() self.sw.enable = mock.Mock() open_mock = mock.mock_open() return_val = None with mock.patch('builtins.open', open_mock, create=True): return_val = self.sw.create_unit_file() self.assertTrue(return_val) open_mock.assert_called_with( '/etc/systemd/system/kolla-test-container.service', 'w' ) open_mock.return_value.write.assert_called_once_with('test data') self.sw.reload.assert_called_once() self.sw.enable.assert_called_once() def test_create_unit_file_no_change(self): self.sw.generate_unit_file = mock.Mock() self.sw.check_unit_change = mock.Mock(return_value=False) self.sw.reload = mock.Mock() self.sw.enable = mock.Mock() open_mock = mock.mock_open() return_val = self.sw.create_unit_file() self.assertFalse(return_val) open_mock.assert_not_called() self.sw.reload.assert_not_called() self.sw.enable.assert_not_called() def test_remove_unit_file(self): self.sw.check_unit_file = mock.Mock(return_value=True) os.remove = mock.Mock() self.sw.reload = mock.Mock() return_val = self.sw.remove_unit_file() self.assertTrue(return_val) os.remove.assert_called_once_with( '/etc/systemd/system/kolla-test-container.service' ) self.sw.reload.assert_called_once() def test_get_unit_state(self): unit_list = [ ('foo.service', '', 'loaded', 'active', 'exited'), ('kolla-test-container.service', '', 'loaded', 'active', 'running') ] self.sw.manager.ListUnits = mock.Mock(return_value=unit_list) state = self.sw.get_unit_state() self.sw.manager.ListUnits.assert_called_once() self.assertEqual('running', state) def test_get_unit_state_not_exist(self): unit_list = [ ('foo.service', '', 'loaded', 'active', 'exited'), ('bar.service', '', 'loaded', 'active', 'running') ] self.sw.manager.ListUnits = mock.Mock(return_value=unit_list) state = self.sw.get_unit_state() self.sw.manager.ListUnits.assert_called_once() self.assertIsNone(state) def test_wait_for_unit(self): self.sw.get_unit_state = mock.Mock() self.sw.get_unit_state.side_effect = ['starting', 'running'] result = self.sw.wait_for_unit(10) self.assertTrue(result) def test_wait_for_unit_timeout(self): self.sw.get_unit_state = mock.Mock() self.sw.get_unit_state.side_effect = [ 'starting', 'starting', 'failed', 'failed'] result = self.sw.wait_for_unit(10) self.assertFalse(result)