Wait for specified duration during LPmode on/off (#503)

* wait for specified duration during LPmode on/off * Add Macro and remove redundant sleep * Update test_cmis.py * Address review comments * Indentation issue
sonic-net · Oct 18, 2024 · 4dcbaa7 · 4dcbaa7
1 parent 1abd824
commit 4dcbaa7
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 6 deletions.
diff --git a/sonic_platform_base/sonic_xcvr/api/public/cmis.py b/sonic_platform_base/sonic_xcvr/api/public/cmis.py
@@ -1066,6 +1066,21 @@ def get_lpmode(self):
                 return True
         return False
 
+    def wait_time_condition(self, condition_func, expected_state, duration_ms, delay_retry):
+        '''
+        This function will wait and retry based on
+        condition function state and delay provided
+        '''
+        start_time = time.time()
+        duration = duration_ms / 1000
+        # Loop until the duration has elapsed
+        while (time.time() - start_time) < duration:
+            if condition_func() == expected_state:
+                return True
+            # Sleep for a delay_retry interval before the next check
+            time.sleep(delay_retry)
+        return condition_func() == expected_state
+
     def set_lpmode(self, lpmode):
         '''
         This function sets the module to low power state.
@@ -1077,23 +1092,21 @@ def set_lpmode(self, lpmode):
         if self.is_flat_memory() or not self.get_lpmode_support():
             return False
 
+        DELAY_RETRY = 0.1
         lpmode_val = self.xcvr_eeprom.read(consts.MODULE_LEVEL_CONTROL)
         if lpmode_val is not None:
             if lpmode is True:
                 # Force module transition to LowPwr under SW control
                 lpmode_val = lpmode_val | (1 << CmisApi.LowPwrRequestSW)
                 self.xcvr_eeprom.write(consts.MODULE_LEVEL_CONTROL, lpmode_val)
-                time.sleep(0.1)
-                return self.get_lpmode()
+                return self.wait_time_condition(self.get_lpmode, True, self.get_module_pwr_down_duration(), DELAY_RETRY)
             else:
                 # Force transition from LowPwr to HighPower state under SW control.
                 # This will transition LowPwrS signal to False. (see Table 6-12 CMIS v5.0)
                 lpmode_val = lpmode_val & ~(1 << CmisApi.LowPwrRequestSW)
                 lpmode_val = lpmode_val & ~(1 << CmisApi.LowPwrAllowRequestHW)
                 self.xcvr_eeprom.write(consts.MODULE_LEVEL_CONTROL, lpmode_val)
-                time.sleep(1)
-                mstate = self.get_module_state()
-                return True if mstate == 'ModuleReady' else False
+                return self.wait_time_condition(self.get_module_state, 'ModuleReady', self.get_module_pwr_up_duration(), DELAY_RETRY)
         return False
 
     def get_loopback_capability(self):

diff --git a/tests/sonic_xcvr/test_cmis.py b/tests/sonic_xcvr/test_cmis.py
@@ -1012,7 +1012,7 @@ def test_reset(self):
         assert kall[0] == (consts.MODULE_LEVEL_CONTROL, 0x8)
 
     @pytest.mark.parametrize("lpmode", [
-        ([True], [False])
+        True, False
     ])
     def test_set_low_power(self, lpmode):
         self.api.xcvr_eeprom.read = MagicMock()
@@ -1024,6 +1024,8 @@ def test_set_low_power(self, lpmode):
         self.api.get_lpmode_support = MagicMock()
         self.api.get_lpmode_support.return_value = True
         self.api.get_module_state = MagicMock()
+        self.api.get_lpmode = MagicMock()
+        self.api.get_lpmode.return_value = True
         self.api.get_module_state.return_value = "ModuleReady"
         self.api.set_lpmode(lpmode)