Modify stonith action retrieval and improve test coverage for _get_stonith_action method (#12)

Modify stonith action retrieval to use local cluster commands instead of cibadmin query; as that throws an error when the stonith-action is not explicitly set.

Author: devanshjainms

src/module_utils/get_cluster_status.py

@@ -48,16 +48,20 @@ def _get_stonith_action(self) -> None:
         """
         Retrieves the stonith action from the system.
         """
+        self.result["stonith_action"] = "unknown"
         try:
             stonith_action = self.execute_command_subprocess(STONITH_ACTION[self.ansible_os_family])
-            stonith_action = (
-                stonith_action.split("stonith-action:")[-1]
-                if self.ansible_os_family == "REDHAT"
-                else stonith_action
-            )
-            self.result["stonith_action"] = stonith_action.strip()
-        except Exception:
-            self.result["stonith_action"] = "reboot"
+            actions = [
+                "reboot",
+                "poweroff",
+                "off",
+            ]
+            for action in actions:
+                if action in stonith_action:
+                    self.result["stonith_action"] = action
+                    break
+        except Exception as ex:
+            self.log(logging.WARNING, f"Failed to get stonith action: {str(ex)}")
 
     def _validate_cluster_basic_status(self, cluster_status_xml: ET.Element):
         """

tests/module_utils/get_cluster_status_test.py

@@ -66,7 +66,7 @@ def base_checker(self):
         """
         return TestableBaseClusterChecker(ansible_os_family="REDHAT")
 
-    def test_get_stonith_action(self, mocker, base_checker):
+    def test_get_stonith_action_rhel94(self, mocker, base_checker):
         """
         Test the _get_stonith_action method when the command executes successfully.
 
@@ -75,14 +75,40 @@ def test_get_stonith_action(self, mocker, base_checker):
         :param base_checker: Instance of TestableBaseClusterChecker.
         :type base_checker: TestableBaseClusterChecker
         """
-        mock_execute = mocker.patch.object(
-            base_checker, "execute_command_subprocess", return_value="stonith-action:reboot"
-        )
+        return_values = ["reboot", "poweroff", "off"]
+        for return_value in return_values:
+            mock_execute = mocker.patch.object(
+                base_checker,
+                "execute_command_subprocess",
+                return_value="Cluster Properties: cib-bootstrap-options\n"
+                + f" stonith-action={return_value}",
+            )
 
-        base_checker._get_stonith_action()
+            base_checker._get_stonith_action()
+            mock_execute.assert_called_once()
+            assert base_checker.result["stonith_action"] == return_value
 
-        mock_execute.assert_called_once()
-        assert base_checker.result["stonith_action"] == "reboot"
+    def test_get_stonith_action(self, mocker, base_checker):
+        """
+        Test the _get_stonith_action method when the command executes successfully.
+
+        :param mocker: Mocking library to patch methods.
+        :type mocker: mocker.MockerFixture
+        :param base_checker: Instance of TestableBaseClusterChecker.
+        :type base_checker: TestableBaseClusterChecker
+        """
+        return_values = ["reboot", "poweroff", "off"]
+        for return_value in return_values:
+            mock_execute = mocker.patch.object(
+                base_checker,
+                "execute_command_subprocess",
+                return_value="Cluster Properties: cib-bootstrap-options\n"
+                + f" stonith-action: {return_value}",
+            )
+
+            base_checker._get_stonith_action()
+            mock_execute.assert_called_once()
+            assert base_checker.result["stonith_action"] == return_value
 
     def test_get_stonith_action_exception(self, mocker, base_checker):
         """
@@ -100,7 +126,7 @@ def test_get_stonith_action_exception(self, mocker, base_checker):
         base_checker._get_stonith_action()
 
         mock_execute.assert_called_once()
-        assert base_checker.result["stonith_action"] == "reboot"
+        assert base_checker.result["stonith_action"] == "unknown"
 
     def test_validate_cluster_basic_status_success(self, mocker, base_checker):
         """