Cdb fw upgrade

sonic_platform_base/sonic_xcvr/api/public/cmis.py

@@ -1175,6 +1175,10 @@ def get_module_fw_info(self):
 
         # get fw info (CMD 0100h)
         rpllen, rpl_chkcode, rpl = self.cdb.get_fw_info()
+        # Interface NACK or timeout
+        if (rpllen is None) or (rpl_chkcode is None):
+            return {'status': False, 'info': "Interface fail", 'result': 0} # Return result 0 for distinguishing CDB is maybe in busy or failure.
+
         # password issue
         if self.cdb.cdb_chkcode(rpl) != rpl_chkcode:
             string = 'Get module FW info: Need to enter password\n'

sonic_platform_base/sonic_xcvr/api/public/cmisCDB.py

@@ -121,12 +121,12 @@ def cdb1_chkstatus(self):
             30h-3Fh=Custom
         '''
         status = self.xcvr_eeprom.read(consts.CDB1_STATUS)
-        is_busy = bool((status >> 7) & 0x1)
+        is_busy = bool(((0x80 if status is None else status) >> 7) & 0x1)
         cnt = 0
         while is_busy and cnt < MAX_WAIT:
             time.sleep(0.1)
             status = self.xcvr_eeprom.read(consts.CDB1_STATUS)
-            is_busy = bool((status >> 7) & 0x1)
+            is_busy = bool(((0x80 if status is None else status) >> 7) & 0x1)
             cnt += 1
         return status
 
@@ -382,6 +382,8 @@ def block_write_epl(self, addr, data, autopaging_flag, writelength):
         cmd = bytearray(b'\x01\x04\x08\x00\x04\x00\x00\x00')
         addr_byte = struct.pack('>L',addr)
         cmd += addr_byte
+        cmd[130-INIT_OFFSET] = (epl_len >> 8) & 0xff
+        cmd[131-INIT_OFFSET] =  epl_len       & 0xff
         cmd[133-INIT_OFFSET] = self.cdb_chkcode(cmd)
         self.write_cdb(cmd)
         status = self.cdb1_chkstatus()
@@ -433,6 +435,7 @@ def run_fw_image(self, mode = 0x01):
         cmd[137-INIT_OFFSET] = mode
         cmd[138-INIT_OFFSET] = 2 # Delay to Reset 512 ms
         cmd[133-INIT_OFFSET] = self.cdb_chkcode(cmd)
+        delay = int.from_bytes(cmd[138-INIT_OFFSET:138+2-INIT_OFFSET], byteorder='big') + 50 # Add few ms on setting time.
         self.write_cdb(cmd)
         status = self.cdb1_chkstatus()
         if (status != 0x1):
@@ -444,6 +447,7 @@ def run_fw_image(self, mode = 0x01):
         else:
             txt = 'Run firmware status: Success'
         logger.info(txt)
+        time.sleep(delay/1000) # Wait "delay time" to avoid other cmd sent before "run_fw_image" start.
         return status
 
     # Commit FW image

tests/sonic_xcvr/test_cmis.py

@@ -996,6 +996,27 @@ def test_get_module_level_flag(self, mock_response, expected):
         result = self.api.get_module_level_flag()
         assert result == expected
 
+    @pytest.mark.parametrize("mock_response, expected", [
+        ((128, 1, [0] * 128),  {'status': True, 'info': "", 'result': 0}),
+        ((None, 1, [0] * 128),  {'status': False, 'info': "", 'result': 0}),
+        ((128, None, [0] * 128),  {'status': False, 'info': "", 'result': 0}),
+        ((128, 0, [0] * 128),  {'status': False, 'info': "", 'result': None}),
+    ])
+    def test_get_module_fw_info(self, mock_response, expected):
+        self.api.cdb = MagicMock()
+        self.api.cdb.cdb_chkcode = MagicMock()
+        self.api.cdb.cdb_chkcode.return_value = 1
+        self.api.get_module_active_firmware = MagicMock()
+        self.api.get_module_active_firmware.return_value = "1.0"
+        self.api.get_module_inactive_firmware = MagicMock()
+        self.api.get_module_inactive_firmware.return_value = "1.1"
+        self.api.cdb.get_fw_info = MagicMock()
+        self.api.cdb.get_fw_info.return_value = mock_response
+        result = self.api.get_module_fw_info()
+        if result['status'] == False: # Check 'result' when 'status' == False for distinguishing error type.
+            assert result['result'] == expected['result']
+        assert result['status'] == expected['status']
+
     @pytest.mark.parametrize("input_param, mock_response, expected", [
         (1, 1,  (True, 'Module FW run: Success\n')),
         (1, 64,  (False, 'Module FW run: Fail\nFW_run_status 64\n')),