[CMIS] Catch Exception to avoid CMIS code crash (#299)

* catch Exception to avoid CMIS code crash

Signed-off-by: Kebo Liu <[email protected]>

* fix review comments, add more UT test

Signed-off-by: Kebo Liu <[email protected]>

Signed-off-by: Kebo Liu <[email protected]>

Author: keboliu

sonic_platform_base/sonic_xcvr/api/public/cmis.py

@@ -203,9 +203,9 @@ def get_transceiver_bulk_status(self):
             bulk_status["tx%dpower" % i] = self.mw_to_dbm(tx_power[i - 1]) if self.get_tx_power_support() else 'N/A'
 
         laser_temp_dict = self.get_laser_temperature()
-        bulk_status['laser_temperature'] = laser_temp_dict['monitor value']
         self.vdm_dict = self.get_vdm()
         try:
+            bulk_status['laser_temperature'] = laser_temp_dict['monitor value']
             bulk_status['prefec_ber'] = self.vdm_dict['Pre-FEC BER Average Media Input'][1][0]
             bulk_status['postfec_ber'] = self.vdm_dict['Errored Frames Average Media Input'][1][0]
         except (KeyError, TypeError):
@@ -1842,7 +1842,11 @@ def get_application_advertisement(self):
 
         if not self.is_flat_memory():
             # Read the application advertisement in page01
-            dic.update(self.xcvr_eeprom.read(consts.APPLS_ADVT_FIELD_PAGE01))
+            try:
+                dic.update(self.xcvr_eeprom.read(consts.APPLS_ADVT_FIELD_PAGE01))
+            except TypeError as e:
+                logger.error('Failed to read APPLS_ADVT_FIELD_PAGE01: ' + str(e))
+                return ret
 
         for app in range(1, 16):
             buf = {}

sonic_platform_base/sonic_xcvr/api/public/cmisVDM.py

@@ -50,7 +50,9 @@ def get_vdm_page(self, page, VDM_flag_page):
         if page not in [0x20, 0x21, 0x22, 0x23]:
             raise ValueError('Page not in VDM Descriptor range!')
         vdm_descriptor = self.xcvr_eeprom.read_raw(page * PAGE_SIZE + PAGE_OFFSET, PAGE_SIZE)
-        
+        if not vdm_descriptor:
+            return {}
+
         # Odd Adress VDM observable type ID, real-time monitored value in Page + 4
         vdm_typeID = vdm_descriptor[1::2]
         # Even Address
@@ -83,6 +85,9 @@ def get_vdm_page(self, page, VDM_flag_page):
                 vdm_thrsh_low_alarm_raw = self.xcvr_eeprom.read_raw(vdm_low_alarm_offset, VDM_SIZE, True)
                 vdm_thrsh_high_warn_raw = self.xcvr_eeprom.read_raw(vdm_high_warn_offset, VDM_SIZE, True)
                 vdm_thrsh_low_warn_raw = self.xcvr_eeprom.read_raw(vdm_low_warn_offset, VDM_SIZE, True)
+                if not vdm_value_raw or not vdm_thrsh_high_alarm_raw or not vdm_thrsh_low_alarm_raw \
+                   or not vdm_high_warn_offset or not vdm_thrsh_low_warn_raw:
+                    return {}
                 if vdm_format == 'S16':
                     vdm_value = struct.unpack('>h',vdm_value_raw)[0] * scale
                     vdm_thrsh_high_alarm = struct.unpack('>h', vdm_thrsh_high_alarm_raw)[0] * scale

tests/sonic_xcvr/test_cmisVDM.py

@@ -64,6 +64,54 @@ def test_get_vdm_page(self, input_param, mock_response, expected):
         result = self.api.get_vdm_page(*input_param)
         assert result == expected
 
+    @pytest.mark.parametrize("input_param, mock_response, expected", [
+        (
+            [0x20, [0]*128],    # input_param
+            [                   # mock_response
+                (
+                    16, 9, 16, 11, 16, 13, 16, 15, 32, 10, 33, 10,  0,  0,  0,  0,
+                    80,128, 0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,
+                    160,143,0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,
+                    0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,
+                    0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,
+                    0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,
+                    0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,
+                    0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,  0,
+                ),
+
+                None,
+                None,
+                None,
+                None,
+                None,
+                None,
+                None,
+                None,
+            ],
+            {}
+        )
+    ])
+    def test_get_vdm_page_none_vdm_value_raw(self, input_param, mock_response, expected):
+        self.api.xcvr_eeprom.read_raw = MagicMock()
+        self.api.xcvr_eeprom.read_raw.side_effect = mock_response
+        result = self.api.get_vdm_page(*input_param)
+        assert result == expected
+
+    @pytest.mark.parametrize("input_param, mock_response, expected", [
+        (
+            [0x20, [0]*128],    # input_param
+            [                   # mock_response
+                None,
+            ],
+            {}
+        )
+    ])
+    def test_get_vdm_page_none_vdm_descriptor(self, input_param, mock_response, expected):
+        self.api.xcvr_eeprom.read_raw = MagicMock()
+        self.api.xcvr_eeprom.read_raw.side_effect = mock_response
+        result = self.api.get_vdm_page(*input_param)
+        assert result == expected
+
     @pytest.mark.parametrize("mock_response, expected", [
         (
             [   # mock_response