Added API to decommison all datapaths of a CMIS compliant module (#471)

sonic-net · May 16, 2024 · 862a67c · 862a67c
1 parent 0362460
commit 862a67c
Show file tree

Hide file tree

Showing 2 changed files with 58 additions and 0 deletions.
diff --git a/sonic_platform_base/sonic_xcvr/api/public/cmis.py b/sonic_platform_base/sonic_xcvr/api/public/cmis.py
@@ -2123,6 +2123,31 @@ def scs_apply_datapath_init(self, channel):
         '''
         return self.xcvr_eeprom.write("%s_%d" % (consts.STAGED_CTRL_APPLY_DPINIT_FIELD, 0), channel)
 
+    def decommission_all_datapaths(self):
+        '''
+            Return True if all datapaths are successfully de-commissioned, False otherwise
+        '''
+        # De-init all datpaths
+        self.set_datapath_deinit((1 << self.NUM_CHANNELS) - 1)
+        # Decommision all lanes by apply AppSel=0
+        self.set_application(((1 << self.NUM_CHANNELS) - 1), 0, 0)
+        # Start with AppSel=0 i.e undo any default AppSel
+        self.scs_apply_datapath_init((1 << self.NUM_CHANNELS) - 1)
+
+        dp_state = self.get_datapath_state()
+        config_state = self.get_config_datapath_hostlane_status()
+
+        for lane in range(self.NUM_CHANNELS):
+            name = "DP{}State".format(lane + 1)
+            if dp_state[name] != 'DataPathDeactivated':
+                return False
+
+            name = "ConfigStatusLane{}".format(lane + 1)
+            if config_state[name] != 'ConfigSuccess':
+                return False
+
+        return True
+
     def get_rx_output_amp_max_val(self):
         '''
         This function returns the supported RX output amp val

diff --git a/tests/sonic_xcvr/test_cmis.py b/tests/sonic_xcvr/test_cmis.py
@@ -2311,6 +2311,39 @@ def test_set_application(self):
         self.api.set_application(0x7fffffff, 1, 1)
         assert self.api.xcvr_eeprom.write.call_count == self.api.NUM_CHANNELS
 
+    @pytest.mark.parametrize("datapath_state,config_state", [
+        ( {
+            'DP1State': 'DataPathDeactivated',
+            'DP2State': 'DataPathDeactivated',
+            'DP3State': 'DataPathDeactivated',
+            'DP4State': 'DataPathDeactivated',
+            'DP5State': 'DataPathDeactivated',
+            'DP6State': 'DataPathDeactivated',
+            'DP7State': 'DataPathDeactivated',
+            'DP8State': 'DataPathDeactivated',
+          },
+          {
+            'ConfigStatusLane1': 'ConfigSuccess',
+            'ConfigStatusLane2': 'ConfigSuccess',
+            'ConfigStatusLane3': 'ConfigSuccess',
+            'ConfigStatusLane4': 'ConfigSuccess',
+            'ConfigStatusLane5': 'ConfigSuccess',
+            'ConfigStatusLane6': 'ConfigSuccess',
+            'ConfigStatusLane7': 'ConfigSuccess',
+            'ConfigStatusLane8': 'ConfigSuccess' 
+          } )
+    ])
+    def test_decommission_all_datapaths(self, datapath_state, config_state):
+        self.api.xcvr_eeprom.read = MagicMock()
+        self.api.xcvr_eeprom.write = MagicMock()
+        self.api.set_datapath_deinit = MagicMock(return_value = True)
+
+        self.api.get_datapath_state = MagicMock()
+        self.api.get_datapath_state.return_value = datapath_state
+        self.api.get_config_datapath_hostlane_status = MagicMock()
+        self.api.get_config_datapath_hostlane_status.return_value = config_state
+        assert True == self.api.decommission_all_datapaths()
+
     def test_set_module_si_eq_pre_settings(self):
         optics_si_dict = { "OutputEqPreCursorTargetRx":{
                              "OutputEqPreCursorTargetRx1":2, "OutputEqPreCursorTargetRx2":2, "OutputEqPreCursorTargetRx3":2, "OutputEqPreCursorTargetRx4":2,