Cdb fw upgrade (#308)

* [Cloudlight] QSFP-DD FW upgrade doesn't work (#257)

- Description
cdb1_chkstatus will crash when i2c NACK or timeout.

- Motivation and Context
I2C of transceiver might NACK or stretching when FW upgrade, assuming "None" means "CdbIsBusy" until timeout.

* [Cloudlight] QSFP-DD FW upgrade doesn't work (#257)

- Description
Waiting a delay in "run_fw_image" to ensure it is really executed.
Return a special package when get none in "get_module_fw_info".

- Motivation and Context
"run_fw_image" will be executed after a delay which according to run cmd, waiting the delay in "run_fw_image" to avoid aother cmd sent before it really executing.
CDB cmds will maybe cause several seconds NACK or stretching on i2c bus depend on implementation of module vendor, handling this situation for compatible with different implementation.

* [Cloudlight] QSFP-DD FW upgrade doesn't work (#257)

- Description
Using real length to replace fixed number in "block_write_epl" function.

- Motivation and Context
To avoid a wrong epl length used in module.

* Update unit tests for cmis.

Test : Creating "get_module_fw_info" test.

Author: CliveNi

sonic_platform_base/sonic_xcvr/api/public/cmis.py

@@ -1184,6 +1184,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')),