Dev: unittests: Adjust unit test for previous commit

Author: liangxin1300

test/unittests/test_ui_sbd.py

@@ -90,7 +90,7 @@ def test_sbd_configure_completer_success(self, mock_ServiceManager, mock_is_usin
         mock_ServiceManager.return_value.service_is_active.side_effect = [True, True]
         mock_is_using_disk_based_sbd.return_value = False
         mock_is_using_diskless_sbd.return_value = True
-        self.assertEqual(ui_sbd.sbd_configure_completer(["configure", ""]), ["show", "watchdog-timeout=", "watchdog-device="])
+        self.assertEqual(ui_sbd.sbd_configure_completer(["configure", ""]), ["show", "watchdog-timeout=", "crashdump-watchdog-timeout=", "watchdog-device="])
         mock_ServiceManager.return_value.service_is_active.assert_has_calls([
             mock.call(constants.PCMK_SERVICE),
             mock.call(constants.SBD_SERVICE)
@@ -115,7 +115,7 @@ def setUp(self, mock_get_sbd_device_from_config, mock_get_sbd_device_metadata, m
         mock_get_sbd_device_from_config.return_value = ["/dev/sda1"]
         mock_get_watchdog_device_from_sbd_config.return_value = "/dev/watchdog0"
         mock_get_sbd_watchdog_timeout.return_value = 10
-        mock_get_sbd_device_metadata.return_value = {"watchdog": 10, "msgwait": 20}
+        mock_get_sbd_device_metadata.return_value = {"watchdog": 10, "allocate": 5, "loop": 5, "msgwait": 20}
         self.sbd_instance_diskbased = ui_sbd.SBD()
         self.sbd_instance_diskbased._load_attributes()
 
@@ -273,32 +273,13 @@ def test_parse_args_invalid_timeout_value(self):
             self.sbd_instance_diskbased._parse_args(["watchdog-timeout=xxx"])
         self.assertEqual(str(e.exception), "Invalid timeout value: xxx")
 
-    def test_parse_args_unknown_arg(self):
-        with self.assertRaises(ui_sbd.SBD.SyntaxError) as e:
-            self.sbd_instance_diskbased._parse_args(["name=xin"])
-        self.assertEqual(str(e.exception), "Unknown argument: name=xin")
-
     @mock.patch('logging.Logger.debug')
     @mock.patch('crmsh.watchdog.Watchdog.get_watchdog_device')
     def test_parse_args(self, mock_get_watchdog_device, mock_logger_debug):
         mock_get_watchdog_device.return_value = "/dev/watchdog0"
         args = self.sbd_instance_diskbased._parse_args(["watchdog-timeout=10", "watchdog-device=/dev/watchdog0"])
         self.assertEqual(args, {"watchdog": 10, "watchdog-device": "/dev/watchdog0"})
 
-    @mock.patch('logging.Logger.warning')
-    @mock.patch('logging.Logger.info')
-    def test_adjust_timeout_dict(self, mock_logger_info, mock_logger_warning):
-        timeout_dict = {"watchdog": 10, "msgwait": 10}
-        res = ui_sbd.SBD._adjust_timeout_dict(timeout_dict)
-        self.assertEqual(res, timeout_dict)
-        mock_logger_warning.assert_called_once_with("It's recommended to set msgwait timeout >= 2*watchdog timeout")
-        timeout_dict = {"watchdog": 10}
-        res = ui_sbd.SBD._adjust_timeout_dict(timeout_dict)
-        self.assertEqual(res, {"watchdog": 10, "msgwait": 20})
-        timeout_dict = {"msgwait": 10}
-        res = ui_sbd.SBD._adjust_timeout_dict(timeout_dict)
-        self.assertEqual(res, {"watchdog": 5, "msgwait": 10})
-
     @mock.patch("crmsh.ui_sbd.SBD.configure_usage", new_callable=mock.PropertyMock)
     @mock.patch('builtins.print')
     @mock.patch('logging.Logger.error')
@@ -310,41 +291,38 @@ def test_do_configure_no_args(self, mock_logger_error, mock_print, mock_configur
         mock_logger_error.assert_called_once_with('%s', "No argument")
         mock_print.assert_called_once_with("usage data")
 
+    @mock.patch('logging.Logger.info')
     @mock.patch('crmsh.sbd.SBDManager')
-    def test_configure_diskbase(self, mock_SBDManager):
-        parameter_dict = {"watchdog": 12, "watchdog-device": "/dev/watchdog100"}
-        self.sbd_instance_diskbased._adjust_timeout_dict = mock.Mock(return_value=parameter_dict)
+    def test_configure_diskbase(self, mock_SBDManager, mock_logger_info):
+        parameter_dict = {"watchdog": 12, "watchdog-device": "/dev/watchdog100", "crashdump-watchdog": 12}
+        self.sbd_instance_diskbased.is_kdump_service_active = mock.Mock(return_value=True)
+        self.sbd_instance_diskbased.set_crashdump_option = mock.Mock()
         mock_SBDManager.return_value.init_and_deploy_sbd = mock.Mock()
         self.sbd_instance_diskbased._configure_diskbase(parameter_dict)
         mock_SBDManager.assert_called_once_with(
             device_list_to_init=self.sbd_instance_diskbased.device_list_from_config,
-            timeout_dict={"watchdog": 12, "msgwait": 20, "watchdog-device": "/dev/watchdog100"},
-            update_dict={
-                "SBD_WATCHDOG_DEV": "/dev/watchdog100"
-            }
+            timeout_dict={'watchdog': 12, 'allocate': 5, 'loop': 5, 'msgwait': 36},
+            update_dict={'SBD_TIMEOUT_ACTION': 'flush,crashdump', 'SBD_OPTS': '-C 12', 'SBD_WATCHDOG_DEV': '/dev/watchdog100'}
         )
         mock_SBDManager.return_value.init_and_deploy_sbd.assert_called_once()
 
     @mock.patch('logging.Logger.info')
-    @mock.patch('crmsh.utils.is_subdict')
     @mock.patch('crmsh.sbd.SBDManager')
-    def test_configure_diskbase_no_change(self, mock_SBDManager, mock_is_subdict, mock_logger_info):
+    def test_configure_diskbase_no_change(self, mock_SBDManager, mock_logger_info):
         parameter_dict = {"watchdog": 10, "watchdog-device": "/dev/watchdog0"}
-        mock_is_subdict.return_value = True
         self.sbd_instance_diskbased._configure_diskbase(parameter_dict)
         mock_logger_info.assert_called_once_with("No change in SBD configuration")
 
+    @mock.patch('logging.Logger.info')
     @mock.patch('crmsh.sbd.SBDManager')
-    def test_configure_diskless(self, mock_SBDManager):
-        parameter_dict = {"watchdog": 12, "watchdog-device": "/dev/watchdog100"}
-        self.sbd_instance_diskless._adjust_timeout_dict = mock.Mock(return_value=parameter_dict)
+    def test_configure_diskless(self, mock_SBDManager, mock_logger_info):
+        parameter_dict = {"watchdog": 12, "watchdog-device": "/dev/watchdog100", "crashdump-watchdog": 12}
+        self.sbd_instance_diskless.is_kdump_service_active = mock.Mock(return_value=True)
         mock_SBDManager.return_value.init_and_deploy_sbd = mock.Mock()
         self.sbd_instance_diskless._configure_diskless(parameter_dict)
         mock_SBDManager.assert_called_once_with(
-            update_dict={
-                "SBD_WATCHDOG_DEV": "/dev/watchdog100",
-                "SBD_WATCHDOG_TIMEOUT": "12"
-            },
+            timeout_dict={'stonith-watchdog': 24},
+            update_dict={'SBD_WATCHDOG_TIMEOUT': '12', 'SBD_WATCHDOG_DEV': '/dev/watchdog100', 'SBD_TIMEOUT_ACTION': 'flush,crashdump', 'SBD_OPTS': '-C 12 -Z'},
             diskless_sbd=True
         )
         mock_SBDManager.return_value.init_and_deploy_sbd.assert_called_once()

test/unittests/test_utils.py

@@ -1340,9 +1340,3 @@ def test_check_user_access_cluster(mock_user, mock_in, mock_sudo, mock_error):
     with pytest.raises(utils.TerminateSubCommand) as err:
         utils.check_user_access('cluster')
     mock_error.assert_called_once_with('Operation is denied. The current user lacks the necessary privilege.')
-
-
-def test_is_subdict():
-    d1 = {"a": 1, "b": 2}
-    d2 = {"a": 1}
-    assert utils.is_subdict(d2, d1) is True