[ModuleBase] Add get_midplane_down_reason() API

sonic_platform_base/module_base.py

@@ -713,30 +713,40 @@ def get_module_state_transition(self, module_name):
         Args:
             module_name: The name of the module.
         Returns:
-            bool: Returns True if the flag is set, False otherwise.
+            bool: True when the flag, transition type, and non-expired start
+                time describe an active transition; False otherwise.
         """
         module_name = module_name.upper()
         module_key = "CHASSIS_MODULE_TABLE|" + module_name
         state_db = self.get_state_db()
         with self._transition_operation_lock():
             try:
                 current_flag = state_db.hget(module_key, "transition_in_progress")
+                if current_flag != "True":
+                    return False
 
-                # Clear the flag if it's set but has exceeded the timeout period
-                if current_flag == "True":
-                    start_time_str = state_db.hget(module_key, "transition_start_time")
-                    transition_type = state_db.hget(module_key, "transition_type")
-                    if start_time_str is not None and transition_type is not None:
-                        start_time = int(start_time_str)
-                        current_time = int(time.time())
-                        timeout = self._load_transition_timeouts().get(transition_type, 0)
-                        if current_time - start_time > timeout:
-                            # Timeout occurred, clear the flag (lock already held)
-                            self._clear_transition_fields(module_name)
-                            return False
-
-                return current_flag == "True"
+                start_time_str = state_db.hget(module_key, "transition_start_time")
+                transition_type = state_db.hget(module_key, "transition_type")
+
+                if start_time_str is None or transition_type not in self._TRANSITION_TIMEOUT_DEFAULTS:
+                    self._clear_transition_fields(module_name)
+                    return False
+
+                try:
+                    start_time = int(start_time_str)
+                except (TypeError, ValueError):
+                    self._clear_transition_fields(module_name)
+                    return False
+
+                current_time = int(time.time())
+                timeout = self._load_transition_timeouts().get(transition_type, 0)
+                if current_time - start_time > timeout:
+                    self._clear_transition_fields(module_name)
+                    return False
+
+                return True
             except Exception as e:
+                sys.stderr.write("Error getting transition flag for module {}: {}\n".format(module_name, str(e)))
                 return False
 
     ##############################################
@@ -1098,6 +1108,19 @@ def get_midplane_ip(self):
         """
         raise NotImplementedError
 
+    def get_midplane_down_reason(self):
+        """
+        Retrieves the midplane down reason.
+
+        Returns:
+            A tuple (string, string) where the first element is a string
+            containing the midplane down reason. This string must be one of
+            the REBOOT_CAUSE_* strings predefined in ChassisBase. If the first
+            string is "REBOOT_CAUSE_HARDWARE_OTHER", the second string can be
+            used to pass a description of the midplane down reason.
+        """
+        raise NotImplementedError
+
     def is_midplane_reachable(self):
         """
         Retrieves the reachability status of the module from the Supervisor or

tests/module_base_test.py

@@ -39,7 +39,7 @@ def teardown_method(self):
     @pytest.mark.parametrize(
         "method_name",
         ["get_dpu_id", "get_reboot_cause", "get_state_info", "get_pci_bus_info", "pci_detach", "pci_reattach",
-         "do_power_cycle"],
+         "do_power_cycle", "get_midplane_down_reason"],
     )
     def test_not_implemented_methods_raise(self, method_name):
         with pytest.raises(NotImplementedError):
@@ -958,9 +958,11 @@ def test_clear_module_state_transition_various_modules(self, mod):
     def test_get_module_state_transition(self, ret, expected):
         db = MagicMock()
         self.module.state_db = db
-        db.hget.side_effect = [ret, None, None] if ret == "True" else [ret]
+        db.hget.side_effect = [ret, "1000", "startup"] if ret == "True" else [ret]
         with patch.object(self.module, "get_name", return_value="DPU0"), \
-             patch.object(self.module, "_transition_operation_lock"):
+             patch.object(self.module, "_transition_operation_lock"), \
+             patch.object(self.module, "_load_transition_timeouts", return_value={"startup": 300}), \
+             patch("time.time", return_value=1100):
             assert self.module.get_module_state_transition("dpu0") is expected
         db.hget.assert_any_call(self._key("DPU0"), "transition_in_progress")
 
@@ -976,9 +978,11 @@ def test_get_module_state_transition_db_error(self, capsys):
     def test_get_module_state_transition_various_modules(self, mod):
         db = MagicMock()
         self.module.state_db = db
-        db.hget.side_effect = ["True", None, None]
+        db.hget.side_effect = ["True", "1000", "startup"]
         with patch.object(self.module, "get_name", return_value=mod), \
-             patch.object(self.module, "_transition_operation_lock"):
+             patch.object(self.module, "_transition_operation_lock"), \
+             patch.object(self.module, "_load_transition_timeouts", return_value={"startup": 300}), \
+             patch("time.time", return_value=1100):
             assert self.module.get_module_state_transition(mod.lower()) is True
         db.hget.assert_any_call(self._key(mod), "transition_in_progress")
 
@@ -1014,31 +1018,23 @@ def test_get_module_state_transition_within_timeout_returns_true(self):
             assert self.module.get_module_state_transition("dpu0") is True
             mock_clear.assert_not_called()
 
-    def test_get_module_state_transition_missing_start_time(self):
-        """Test that get_module_state_transition returns True when start_time is missing"""
-        db = MagicMock()
-        self.module.state_db = db
-        # Flag is set but start_time is None
-        db.hget.side_effect = ["True", None, "startup"]
-
-        with patch.object(self.module, "get_name", return_value="DPU0"), \
-             patch.object(self.module, "_transition_operation_lock"), \
-             patch.object(self.module, "clear_module_state_transition") as mock_clear:
-            assert self.module.get_module_state_transition("dpu0") is True
-            mock_clear.assert_not_called()
-
-    def test_get_module_state_transition_missing_transition_type(self):
-        """Test that get_module_state_transition returns True when transition_type is missing"""
+    @pytest.mark.parametrize("start_time,transition_type", [
+        (None, "startup"),
+        ("1000", None),
+        ("not-a-timestamp", "startup"),
+        ("1000", "invalid"),
+    ])
+    def test_get_module_state_transition_invalid_metadata(self, start_time, transition_type):
+        """Incomplete or invalid transition metadata is cleared and treated as inactive."""
         db = MagicMock()
         self.module.state_db = db
-        # Flag is set, start_time exists but transition_type is None
-        db.hget.side_effect = ["True", "1000", None]
+        db.hget.side_effect = ["True", start_time, transition_type]
 
         with patch.object(self.module, "get_name", return_value="DPU0"), \
              patch.object(self.module, "_transition_operation_lock"), \
-             patch.object(self.module, "clear_module_state_transition") as mock_clear:
-            assert self.module.get_module_state_transition("dpu0") is True
-            mock_clear.assert_not_called()
+             patch.object(self.module, "_clear_transition_fields") as mock_clear:
+            assert self.module.get_module_state_transition("dpu0") is False
+            mock_clear.assert_called_once_with("DPU0")
 
     @pytest.mark.parametrize("transition_type", ["startup", "shutdown", "reboot"])
     def test_get_module_state_transition_timeout_for_different_types(self, transition_type):