1 files changed, 311 insertions, 0 deletions
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")