Only create CdbFwHandler if module support CDB

Author: Junchao-Mellanox

sonic_platform_base/sonic_xcvr/api/public/c_cmis.py

@@ -55,8 +55,8 @@
 })
 
 class CCmisApi(CmisApi):
-    def __init__(self, xcvr_eeprom, cdb):
-        super(CCmisApi, self).__init__(xcvr_eeprom, cdb)
+    def __init__(self, xcvr_eeprom, init_cdb=False):
+        super(CCmisApi, self).__init__(xcvr_eeprom, init_cdb)
 
     def _get_vdm_key_to_db_prefix_map(self):
         combined_map = {**CMIS_VDM_KEY_TO_DB_PREFIX_KEY_MAP, **C_CMIS_DELTA_VDM_KEY_TO_DB_PREFIX_KEY_MAP}

sonic_platform_base/sonic_xcvr/api/public/cmis.py

@@ -11,8 +11,11 @@
 
 import logging
 from ...codes.public.cmis import CmisCodes
+from ...codes.public.cdb import CdbCodes
 from ...codes.public.sff8024 import Sff8024
 from ...fields import consts
+from ...mem_maps.public.cdb import CdbMemMap
+from ...cdb.cdb_fw import CdbFwHandler as CdbFw
 from ..xcvr_api import XcvrApi
 from .cmisCDB import CmisCdbApi
 from .cmisVDM import CmisVdmApi
@@ -135,11 +138,17 @@ def set_cache_enabled(cls, enabled: bool):
         """
         cls.cache_enabled = bool(enabled)
 
-    def __init__(self, xcvr_eeprom, cdb_fw_hdlr=None):
+    def __init__(self, xcvr_eeprom, init_cdb=False):
         super(CmisApi, self).__init__(xcvr_eeprom)
         self.vdm = CmisVdmApi(xcvr_eeprom) if not self.is_flat_memory() else None
         self.cdb = CmisCdbApi(xcvr_eeprom) if self.is_cdb_supported() else None
-        self.cdb_fw_hdlr = cdb_fw_hdlr if self.is_cdb_supported() else None
+        self.cdb_fw_hdlr = self._create_cdb_fw_handler() if init_cdb else None
+
+    def _create_cdb_fw_handler(self):
+        if not self.is_cdb_supported():
+            return None
+        cdb_mem_map = CdbMemMap(CdbCodes)
+        return CdbFw(self.xcvr_eeprom.reader, self.xcvr_eeprom.writer, cdb_mem_map)
 
     def get_cdb_fw_handler(self):
         return self.cdb_fw_hdlr

sonic_platform_base/sonic_xcvr/xcvr_api_factory.py

@@ -12,9 +12,6 @@
 from .api.public.c_cmis import CCmisApi
 from .mem_maps.public.cmis import CmisMemMap
 from .mem_maps.public.c_cmis import CCmisMemMap
-from .mem_maps.public.cdb import CdbMemMap
-from .cdb.cdb_fw import CdbFwHandler as CdbFw
-from .codes.public.cdb import CdbCodes
 
 from .codes.credo.aec_800g import CmisAec800gCodes
 from .api.credo.aec_800g import CmisAec800gApi
@@ -89,13 +86,11 @@ def _create_cmis_api(self):
              ('EOPTOLINK' in vendor_name and vendor_pn in EOP_800G_VENDOR_PN_LIST):
             api = self._create_api(CmisCodes, CmisMemMap, CmisFr800gApi)
         else:
-            cdb_mem_map = CdbMemMap(CdbCodes)
-            cdb_fw = CdbFw(self.reader, self.writer, cdb_mem_map)
             xcvr_eeprom = XcvrEeprom(self.reader, self.writer, CmisMemMap(CmisCodes))
-            api = CmisApi(xcvr_eeprom, cdb_fw)
+            api = CmisApi(xcvr_eeprom, init_cdb=True)
             if api.is_coherent_module():
                 xcvr_eeprom = XcvrEeprom(self.reader, self.writer, CCmisMemMap(CmisCodes))
-                api = CCmisApi(xcvr_eeprom, cdb_fw)
+                api = CCmisApi(xcvr_eeprom, init_cdb=True)
         return api
 
     def _create_qsfp_api(self):

tests/sonic_xcvr/test_ccmis.py

@@ -3,24 +3,18 @@
 import pytest
 from sonic_platform_base.sonic_xcvr.api.public.c_cmis import CCmisApi, C_CMIS_XCVR_INFO_DEFAULT_DICT
 from sonic_platform_base.sonic_xcvr.mem_maps.public.c_cmis import CCmisMemMap
-from sonic_platform_base.sonic_xcvr.mem_maps.public.cdb import CdbMemMap
 from sonic_platform_base.sonic_xcvr.xcvr_eeprom import XcvrEeprom
 from sonic_platform_base.sonic_xcvr.codes.public.cmis import CmisCodes
-from sonic_platform_base.sonic_xcvr.codes.public.cdb import CdbCodes
-from sonic_platform_base.sonic_xcvr.cdb.cdb_fw import CdbFwHandler as CdbFw
 
 
-CdbFw.initFwHandler = MagicMock(return_value=True)
-
 class TestCCmis(object):
     codes = CmisCodes
     mem_map = CCmisMemMap(codes)
     reader = MagicMock(return_value=None)
     writer = MagicMock()
     eeprom = XcvrEeprom(reader, writer, mem_map)
 
-    cdb = CdbFw(reader, writer, CdbMemMap(CdbCodes))
-    api = CCmisApi(eeprom, cdb)
+    api = CCmisApi(eeprom, init_cdb=False)
 
     @pytest.mark.parametrize("mock_response, expected", [
         (8, 150),

tests/sonic_xcvr/test_cdb_fw.py

@@ -59,7 +59,7 @@ def test_initFwHandler_send_cmd_failure(self):
 
         result = handler.initFwHandler()
 
-        assert result == True
+        assert result == False
 
     def test_initFwHandler_read_reply_none(self):
         """Test initFwHandler when read_reply returns None"""
@@ -75,7 +75,7 @@ def test_initFwHandler_read_reply_none(self):
 
         result = handler.initFwHandler()
 
-        assert result == True
+        assert result == False
 
     def test_initFwHandler_lpl_only(self):
         """Test initFwHandler with LPL only mechanism"""

tests/sonic_xcvr/test_cmis.py

@@ -2997,3 +2997,11 @@ def test_get_tx_adaptive_eq_fail_flag(self, mock_response, expected):
         self.api.xcvr_eeprom.read.return_value = mock_response[1]
         result = self.api.get_tx_adaptive_eq_fail_flag()
         assert result == expected
+
+    @patch('sonic_platform_base.sonic_xcvr.cdb.cdb_fw.CdbFwHandler.initFwHandler', MagicMock(return_value=True))
+    @patch('sonic_platform_base.sonic_xcvr.api.public.cmis.CmisApi.is_cdb_supported')
+    def test_create_cdb_fw_handler(self, mock_cdb_support):
+        mock_cdb_support.return_value = False
+        assert self.api._create_cdb_fw_handler() is None
+        mock_cdb_support.return_value = True
+        assert self.api._create_cdb_fw_handler()

tests/sonic_xcvr/test_sfp_optoe_base.py

@@ -9,11 +9,8 @@
 from sonic_platform_base.sonic_xcvr.xcvr_eeprom import XcvrEeprom 
 from sonic_platform_base.sonic_xcvr.codes.public.cmis import CmisCodes 
 from sonic_platform_base.sonic_xcvr.api.public.sff8472 import Sff8472Api
-from sonic_platform_base.sonic_xcvr.mem_maps.public.cdb import CdbMemMap
-from sonic_platform_base.sonic_xcvr.codes.public.cdb import CdbCodes
-from sonic_platform_base.sonic_xcvr.cdb.cdb_fw import CdbFwHandler as CdbFw
 
-CdbFw.initFwHandler = MagicMock(return_value=True)
+
 class TestSfpOptoeBase(object): 
 
     codes = CmisCodes 
@@ -22,9 +19,8 @@ class TestSfpOptoeBase(object):
     writer = MagicMock() 
     eeprom = XcvrEeprom(reader, writer, mem_map) 
     sfp_optoe_api = SfpOptoeBase() 
-    cdb = CdbFw(reader, writer, CdbMemMap(CdbCodes))
-    ccmis_api = CCmisApi(eeprom, cdb) 
-    cmis_api = CmisApi(eeprom, cdb) 
+    ccmis_api = CCmisApi(eeprom, init_cdb=False) 
+    cmis_api = CmisApi(eeprom, init_cdb=False) 
     sff8472_api = Sff8472Api(eeprom)
 
     def test_is_transceiver_vdm_supported_non_cmis(self):