From d835b2cae8abc71958b69362162e6a70c3d7ef63 Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Wed, 17 Apr 2024 08:48:59 +0200 Subject: Adding upstream version 4.6.0. Signed-off-by: Daniel Baumann --- test/unittests/test_watchdog.py | 311 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 311 insertions(+) create mode 100644 test/unittests/test_watchdog.py (limited to 'test/unittests/test_watchdog.py') diff --git a/test/unittests/test_watchdog.py b/test/unittests/test_watchdog.py new file mode 100644 index 0000000..957f21f --- /dev/null +++ b/test/unittests/test_watchdog.py @@ -0,0 +1,311 @@ +import unittest + +try: + from unittest import mock +except ImportError: + import mock + +from crmsh import watchdog +from crmsh import bootstrap +from crmsh import constants + + +class TestWatchdog(unittest.TestCase): + """ + Unitary tests for crmsh.watchdog.Watchdog + """ + + @classmethod + def setUpClass(cls): + """ + Global setUp. + """ + + def setUp(self): + """ + Test setUp. + """ + self.watchdog_inst = watchdog.Watchdog() + self.watchdog_join_inst = watchdog.Watchdog(remote_user="alice", peer_host="node1") + + def tearDown(self): + """ + Test tearDown. + """ + + @classmethod + def tearDownClass(cls): + """ + Global tearDown. + """ + + def test_watchdog_device_name(self): + res = self.watchdog_inst.watchdog_device_name + assert res is None + + @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr') + def test_verify_watchdog_device_ignore_error(self, mock_run): + mock_run.return_value = (1, None, "error") + res = self.watchdog_inst._verify_watchdog_device("/dev/watchdog", True) + self.assertEqual(res, False) + mock_run.assert_called_once_with("wdctl /dev/watchdog") + + @mock.patch('crmsh.utils.fatal') + @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr') + def test_verify_watchdog_device_error(self, mock_run, mock_error): + mock_run.return_value = (1, None, "error") + mock_error.side_effect = ValueError + with self.assertRaises(ValueError) as err: + self.watchdog_inst._verify_watchdog_device("/dev/watchdog") + mock_error.assert_called_once_with("Invalid watchdog device /dev/watchdog: error") + mock_run.assert_called_once_with("wdctl /dev/watchdog") + + @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr') + def test_verify_watchdog_device(self, mock_run): + mock_run.return_value = (0, None, None) + res = self.watchdog_inst._verify_watchdog_device("/dev/watchdog") + self.assertEqual(res, True) + + @mock.patch('crmsh.watchdog.invoke') + def test_load_watchdog_driver(self, mock_invoke): + self.watchdog_inst._load_watchdog_driver("softdog") + mock_invoke.assert_has_calls([ + mock.call("echo softdog > /etc/modules-load.d/watchdog.conf"), + mock.call("systemctl restart systemd-modules-load") + ]) + + @mock.patch('crmsh.utils.parse_sysconfig') + def test_get_watchdog_device_from_sbd_config(self, mock_parse): + mock_parse_inst = mock.Mock() + mock_parse.return_value = mock_parse_inst + mock_parse_inst.get.return_value = "/dev/watchdog" + res = self.watchdog_inst._get_watchdog_device_from_sbd_config() + self.assertEqual(res, "/dev/watchdog") + mock_parse.assert_called_once_with(bootstrap.SYSCONFIG_SBD) + + @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr') + def test_driver_is_loaded(self, mock_run): + output = """ +button 24576 0 +softdog 16384 2 +btrfs 1474560 1 + """ + mock_run.return_value = (0, output, None) + res = self.watchdog_inst._driver_is_loaded("softdog") + assert res is not None + mock_run.assert_called_once_with("lsmod") + + @mock.patch('crmsh.utils.fatal') + @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr') + def test_set_watchdog_info_error(self, mock_run, mock_error): + mock_run.return_value = (1, None, "error") + mock_error.side_effect = ValueError + with self.assertRaises(ValueError): + self.watchdog_inst._set_watchdog_info() + mock_run.assert_called_once_with(watchdog.Watchdog.QUERY_CMD) + mock_error.assert_called_once_with("Failed to run {}: error".format(watchdog.Watchdog.QUERY_CMD)) + + @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr') + def test_set_watchdog_info(self, mock_run): + output = """ +Discovered 3 watchdog devices: + +[1] /dev/watchdog +Identity: Software Watchdog +Driver: softdog +CAUTION: Not recommended for use with sbd. + +[2] /dev/watchdog0 +Identity: Software Watchdog +Driver: softdog +CAUTION: Not recommended for use with sbd. + +[3] /dev/watchdog1 +Identity: iTCO_wdt +Driver: iTCO_wdt + """ + mock_run.return_value = (0, output, None) + self.watchdog_inst._set_watchdog_info() + self.assertEqual(self.watchdog_inst._watchdog_info_dict, {'/dev/watchdog': 'softdog', '/dev/watchdog0': 'softdog', '/dev/watchdog1': 'iTCO_wdt'}) + + @mock.patch('crmsh.watchdog.Watchdog._verify_watchdog_device') + def test_get_device_through_driver_none(self, mock_verify): + self.watchdog_inst._watchdog_info_dict = {'/dev/watchdog': 'softdog', '/dev/watchdog0': 'softdog', '/dev/watchdog1': 'iTCO_wdt'} + mock_verify.return_value = False + res = self.watchdog_inst._get_device_through_driver("iTCO_wdt") + self.assertEqual(res, None) + mock_verify.assert_called_once_with("/dev/watchdog1") + + @mock.patch('crmsh.watchdog.Watchdog._verify_watchdog_device') + def test_get_device_through_driver(self, mock_verify): + self.watchdog_inst._watchdog_info_dict = {'/dev/watchdog': 'softdog', '/dev/watchdog0': 'softdog', '/dev/watchdog1': 'iTCO_wdt'} + mock_verify.return_value = True + res = self.watchdog_inst._get_device_through_driver("iTCO_wdt") + self.assertEqual(res, "/dev/watchdog1") + mock_verify.assert_called_once_with("/dev/watchdog1") + + @mock.patch('crmsh.utils.fatal') + @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr') + def test_get_driver_through_device_remotely_error(self, mock_run, mock_error): + mock_run.return_value = (1, None, "error") + self.watchdog_join_inst._get_driver_through_device_remotely("test") + mock_run.assert_called_once_with("ssh {} alice@node1 sudo sbd query-watchdog".format(constants.SSH_OPTION)) + mock_error.assert_called_once_with("Failed to run sudo sbd query-watchdog remotely: error") + + @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr') + def test_get_driver_through_device_remotely_none(self, mock_run): + mock_run.return_value = (0, "data", None) + res = self.watchdog_join_inst._get_driver_through_device_remotely("/dev/watchdog") + self.assertEqual(res, None) + mock_run.assert_called_once_with("ssh {} alice@node1 sudo sbd query-watchdog".format(constants.SSH_OPTION)) + + @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr') + def test_get_driver_through_device_remotely(self, mock_run): + output = """ +Discovered 3 watchdog devices: + +[1] /dev/watchdog +Identity: Software Watchdog +Driver: softdog +CAUTION: Not recommended for use with sbd. + +[2] /dev/watchdog0 +Identity: Software Watchdog +Driver: softdog +CAUTION: Not recommended for use with sbd. + +[3] /dev/watchdog1 +Identity: iTCO_wdt +Driver: iTCO_wdt + """ + mock_run.return_value = (0, output, None) + res = self.watchdog_join_inst._get_driver_through_device_remotely("/dev/watchdog") + self.assertEqual(res, "softdog") + mock_run.assert_called_once_with("ssh {} alice@node1 sudo sbd query-watchdog".format(constants.SSH_OPTION)) + + def test_get_first_unused_device_none(self): + res = self.watchdog_inst._get_first_unused_device() + self.assertEqual(res, None) + + @mock.patch('crmsh.watchdog.Watchdog._verify_watchdog_device') + def test_get_first_unused_device(self, mock_verify): + mock_verify.return_value = True + self.watchdog_inst._watchdog_info_dict = {'/dev/watchdog': 'softdog', '/dev/watchdog0': 'softdog', '/dev/watchdog1': 'iTCO_wdt'} + res = self.watchdog_inst._get_first_unused_device() + self.assertEqual(res, "/dev/watchdog") + mock_verify.assert_called_once_with("/dev/watchdog", ignore_error=True) + + @mock.patch('crmsh.watchdog.Watchdog._get_first_unused_device') + @mock.patch('crmsh.watchdog.Watchdog._verify_watchdog_device') + @mock.patch('crmsh.watchdog.Watchdog._get_watchdog_device_from_sbd_config') + def test_set_input_from_config(self, mock_from_config, mock_verify, mock_first): + mock_from_config.return_value = "/dev/watchdog" + mock_verify.return_value = True + self.watchdog_inst._set_input() + mock_first.assert_not_called() + mock_from_config.assert_called_once_with() + + @mock.patch('crmsh.watchdog.Watchdog._get_first_unused_device') + @mock.patch('crmsh.watchdog.Watchdog._verify_watchdog_device') + @mock.patch('crmsh.watchdog.Watchdog._get_watchdog_device_from_sbd_config') + def test_set_input(self, mock_from_config, mock_verify, mock_first): + mock_from_config.return_value = None + mock_first.return_value = None + self.watchdog_inst._set_input() + self.assertEqual(self.watchdog_inst._input, "softdog") + mock_from_config.assert_called_once_with() + mock_verify.assert_not_called() + mock_first.assert_called_once_with() + + def test_valid_device_false(self): + res = self.watchdog_inst._valid_device("test") + self.assertEqual(res, False) + + @mock.patch('crmsh.watchdog.Watchdog._verify_watchdog_device') + def test_valid_device(self, mock_verify): + mock_verify.return_value = True + self.watchdog_inst._watchdog_info_dict = {'/dev/watchdog': 'softdog', '/dev/watchdog0': 'softdog', '/dev/watchdog1': 'iTCO_wdt'} + res = self.watchdog_inst._valid_device("/dev/watchdog") + self.assertEqual(res, True) + + @mock.patch('crmsh.utils.fatal') + @mock.patch('crmsh.watchdog.Watchdog._get_watchdog_device_from_sbd_config') + @mock.patch('crmsh.watchdog.Watchdog._set_watchdog_info') + def test_join_watchdog_error(self, mock_set_info, mock_from_config, mock_error): + mock_from_config.return_value = None + mock_error.side_effect = SystemExit + with self.assertRaises(SystemExit): + self.watchdog_join_inst.join_watchdog() + mock_set_info.assert_called_once_with() + mock_from_config.assert_called_once_with() + mock_error.assert_called_once_with("Failed to get watchdog device from {}".format(bootstrap.SYSCONFIG_SBD)) + + @mock.patch('crmsh.watchdog.Watchdog._load_watchdog_driver') + @mock.patch('crmsh.watchdog.Watchdog._get_driver_through_device_remotely') + @mock.patch('crmsh.watchdog.Watchdog._valid_device') + @mock.patch('crmsh.watchdog.Watchdog._get_watchdog_device_from_sbd_config') + @mock.patch('crmsh.watchdog.Watchdog._set_watchdog_info') + def test_join_watchdog(self, mock_set_info, mock_from_config, mock_valid, mock_get_driver_remotely, mock_load): + mock_from_config.return_value = "/dev/watchdog" + mock_valid.return_value = False + mock_get_driver_remotely.return_value = "softdog" + + self.watchdog_join_inst.join_watchdog() + + mock_set_info.assert_called_once_with() + mock_from_config.assert_called_once_with() + mock_valid.assert_called_once_with("/dev/watchdog") + mock_get_driver_remotely.assert_called_once_with("/dev/watchdog") + mock_load.assert_called_once_with("softdog") + + @mock.patch('crmsh.watchdog.invokerc') + @mock.patch('crmsh.watchdog.Watchdog._valid_device') + @mock.patch('crmsh.watchdog.Watchdog._set_input') + @mock.patch('crmsh.watchdog.Watchdog._set_watchdog_info') + def test_init_watchdog_valid(self, mock_set_info, mock_set_input, mock_valid, mock_invokerc): + mock_valid.return_value = True + self.watchdog_inst._input = "/dev/watchdog" + self.watchdog_inst.init_watchdog() + mock_invokerc.assert_not_called() + mock_valid.assert_called_once_with("/dev/watchdog") + + @mock.patch('crmsh.utils.fatal') + @mock.patch('crmsh.watchdog.invokerc') + @mock.patch('crmsh.watchdog.Watchdog._valid_device') + @mock.patch('crmsh.watchdog.Watchdog._set_input') + @mock.patch('crmsh.watchdog.Watchdog._set_watchdog_info') + def test_init_watchdog_error(self, mock_set_info, mock_set_input, mock_valid, mock_invokerc, mock_error): + mock_valid.return_value = False + mock_invokerc.return_value = False + self.watchdog_inst._input = "test" + mock_error.side_effect = SystemExit + + with self.assertRaises(SystemExit): + self.watchdog_inst.init_watchdog() + + mock_valid.assert_called_once_with("test") + mock_invokerc.assert_called_once_with("modinfo test") + mock_error.assert_called_once_with("Should provide valid watchdog device or driver name by -w option") + + @mock.patch('crmsh.watchdog.Watchdog._get_device_through_driver') + @mock.patch('crmsh.watchdog.Watchdog._load_watchdog_driver') + @mock.patch('crmsh.watchdog.Watchdog._driver_is_loaded') + @mock.patch('crmsh.watchdog.invokerc') + @mock.patch('crmsh.watchdog.Watchdog._valid_device') + @mock.patch('crmsh.watchdog.Watchdog._set_input') + @mock.patch('crmsh.watchdog.Watchdog._set_watchdog_info') + def test_init_watchdog(self, mock_set_info, mock_set_input, mock_valid, mock_invokerc, mock_is_loaded, mock_load, mock_get_device): + mock_valid.return_value = False + self.watchdog_inst._input = "softdog" + mock_invokerc.return_value = True + mock_is_loaded.return_value = False + mock_get_device.return_value = "/dev/watchdog" + + self.watchdog_inst.init_watchdog() + + mock_valid.assert_called_once_with("softdog") + mock_invokerc.assert_called_once_with("modinfo softdog") + mock_is_loaded.assert_called_once_with("softdog") + mock_load.assert_called_once_with("softdog") + mock_set_info.assert_has_calls([mock.call(), mock.call()]) + mock_get_device.assert_called_once_with("softdog") -- cgit v1.2.3