def run()

in src/module_utils/get_cluster_status.py [0:0]

• 19 lines of code
• 4 McCabe index (conditional complexity)

    def run(self) -> Dict[str, str]:
        """
        Run the cluster status check.

        :return: Result of the cluster status check.
        :rtype: Dict[str, str]
        """
        self.log(logging.INFO, "Starting cluster status check")
        self._get_stonith_action()

        try:
            while not self._is_cluster_ready():
                self.result["cluster_status"] = self.execute_command_subprocess(CLUSTER_STATUS)
                cluster_status_xml = ET.fromstring(self.result["cluster_status"])
                self.log(logging.INFO, "Cluster status retrieved")

                self._validate_cluster_basic_status(cluster_status_xml)
                self._process_node_attributes(cluster_status_xml=cluster_status_xml)

            if not self._is_cluster_stable():
                self.result["message"] = "Pacemaker cluster isn't stable"
                self.log(logging.WARNING, self.result["message"])

        except Exception as ex:
            self.handle_error(ex)

        self.result["end"] = datetime.now()
        self.result["status"] = TestStatus.SUCCESS.value
        self.log(logging.INFO, "Cluster status check completed")
        return self.result