diff --git a/sonic_platform_base/module_base.py b/sonic_platform_base/module_base.py index 53f862725..3928c61ca 100644 --- a/sonic_platform_base/module_base.py +++ b/sonic_platform_base/module_base.py @@ -713,7 +713,8 @@ 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 @@ -721,22 +722,31 @@ def get_module_state_transition(self, module_name): 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 diff --git a/tests/module_base_test.py b/tests/module_base_test.py index a57d4b90a..5ed5b4eae 100644 --- a/tests/module_base_test.py +++ b/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):