From 2a8bfa61f21e2e1d39284ab966fba02048c1019f Mon Sep 17 00:00:00 2001 From: Dante Su Date: Thu, 2 Dec 2021 07:01:11 +0000 Subject: [PATCH] Rebased and squashed Signed-off-by: Dante Su --- sonic_platform_base/sfp_base.py | 14 + .../sonic_xcvr/api/public/cmis.py | 302 +++++++++++++++++- .../sonic_xcvr/api/public/cmisCDB.py | 2 +- .../sonic_xcvr/fields/consts.py | 7 + .../sonic_xcvr/fields/public/cmis.py | 59 ++++ .../sonic_xcvr/mem_maps/public/cmis.py | 14 + .../sonic_xcvr/sfp_optoe_base.py | 114 +++++++ tests/sonic_xcvr/test_cmis.py | 2 + 8 files changed, 510 insertions(+), 4 deletions(-) diff --git a/sonic_platform_base/sfp_base.py b/sonic_platform_base/sfp_base.py index d6e735487..e8bb29814 100644 --- a/sonic_platform_base/sfp_base.py +++ b/sonic_platform_base/sfp_base.py @@ -17,6 +17,12 @@ class SfpBase(device_base.DeviceBase): # Device type definition. Note, this is a constant. DEVICE_TYPE = "sfp" + # SFP port type. Please note this is the cage type rather than transceiver type. + SFP_PORT_TYPE_UNSPECIFIED = "UNSPECIFIED" + SFP_PORT_TYPE_SFP = "SFP" + SFP_PORT_TYPE_QSFP = "QSFP" + SFP_PORT_TYPE_QSFPDD = "QSFP_DD" + # Generic error types definition SFP_STATUS_INITIALIZING = 'Initializing' SFP_STATUS_OK = 'OK' @@ -59,6 +65,14 @@ def __init__(self): self._xcvr_api_factory = XcvrApiFactory(self.read_eeprom, self.write_eeprom) self._xcvr_api = None + def get_port_type(self): + """ + Retrieves the port/cage type of this SFP + Returns: + A string, the port/cage type of this SFP + """ + return self.SFP_PORT_TYPE_UNSPECIFIED + def get_num_thermals(self): """ Retrieves the number of thermals available on this SFP diff --git a/sonic_platform_base/sonic_xcvr/api/public/cmis.py b/sonic_platform_base/sonic_xcvr/api/public/cmis.py index d065f219d..8a0cbf195 100644 --- a/sonic_platform_base/sonic_xcvr/api/public/cmis.py +++ b/sonic_platform_base/sonic_xcvr/api/public/cmis.py @@ -8,7 +8,9 @@ from ...fields import consts from ..xcvr_api import XcvrApi +import ast import logging +from ...codes.public.cmis import CmisCodes from ...fields import consts from ..xcvr_api import XcvrApi from .cmisCDB import CmisCdbApi @@ -135,8 +137,7 @@ def get_transceiver_info(self): "specification_compliance": admin_info[consts.MEDIA_TYPE_FIELD], "vendor_date": admin_info[consts.VENDOR_DATE_FIELD], "vendor_oui": admin_info[consts.VENDOR_OUI_FIELD], - # TODO - "application_advertisement": "N/A", + "application_advertisement": admin_info.get(consts.APPLS_ADVT_FIELD, "N/A") } xcvr_info['host_electrical_interface'] = self.get_host_electrical_interface() xcvr_info['media_interface_code'] = self.get_module_media_interface() @@ -903,6 +904,9 @@ def set_lpmode(self, lpmode): lpmode_val = self.xcvr_eeprom.read(consts.MODULE_LEVEL_CONTROL) if lpmode_val is not None: + # Turn on software control mode + if self.xcvr_eeprom.read(consts.CMIS_MAJOR_REVISION) >= 4: + lpmode_val = lpmode_val & ~(1 << 6) # LowPwrAllowRequestHW if lpmode is True: lpmode_val = lpmode_val | (1 << 4) self.xcvr_eeprom.write(consts.MODULE_LEVEL_CONTROL, lpmode_val) @@ -914,7 +918,7 @@ def set_lpmode(self, lpmode): time.sleep(1) lpmode = self.xcvr_eeprom.read(consts.TRANS_MODULE_STATUS_FIELD) if lpmode is not None: - if lpmode.get('ModuleState') == 'ModuleReady': + if lpmode.get('ModuleState') in ['ModulePwrUp', 'ModuleReady']: return True return False return False @@ -1745,4 +1749,296 @@ def get_transceiver_loopback(self): for lane in range(1, self.NUM_CHANNELS+1): trans_loopback['host_input_loopback_lane%d' % lane] = 'N/A' return trans_loopback + + def set_datapath_init(self, host_lanemask): + """ + Put the datapath into the initialized state + + Args: + host_lanemask: Integer, a bitmask of the lanes on the system/host side + e.g. 0x5 for lane 0 and lane 2. + + Returns: + Boolean, true if success otherwise false + """ + cmis_major = self.xcvr_eeprom.read(consts.CMIS_MAJOR_REVISION) + data = self.xcvr_eeprom.read(consts.DATAPATH_DEINIT_FIELD) + for lane in range(self.NUM_CHANNELS): + if (1 << lane) & host_lanemask == 0: + continue + if cmis_major >= 4: # CMIS v4 onwards + data &= ~(1 << lane) + else: # CMIS v3 + data |= (1 << lane) + self.xcvr_eeprom.write(consts.DATAPATH_DEINIT_FIELD, data) + + def set_datapath_deinit(self, host_lanemask): + """ + Put the datapath into the de-initialized state + + Args: + host_lanemask: Integer, a bitmask of the lanes on the system/host side + e.g. 0x5 for lane 0 and lane 2. + + Returns: + Boolean, true if success otherwise false + """ + cmis_major = self.xcvr_eeprom.read(consts.CMIS_MAJOR_REVISION) + data = self.xcvr_eeprom.read(consts.DATAPATH_DEINIT_FIELD) + for lane in range(self.NUM_CHANNELS): + if (1 << lane) & host_lanemask == 0: + continue + if cmis_major >= 4: # CMIS v4 onwards + data |= (1 << lane) + else: # CMIS v3 + data &= ~(1 << lane) + self.xcvr_eeprom.write(consts.DATAPATH_DEINIT_FIELD, data) + + def get_host_speed(self, ifname): + """ + Get the port speed from the host interface name + + Args: + ifname: String, host interface name + + Returns: + Integer, the port speed if success otherwise 0 + """ + # see HOST_ELECTRICAL_INTERFACE of sff8024.py + speed = 0 + if '400G' in ifname: + speed = 400000 + elif '200G' in ifname: + speed = 200000 + elif '100G' in ifname or 'CAUI-4' in ifname: + speed = 100000 + elif '50G' in ifname or 'LAUI-2' in ifname: + speed = 50000 + elif '40G' in ifname or 'XLAUI' in ifname or 'XLPPI' in ifname: + speed = 40000 + elif '25G' in ifname: + speed = 25000 + elif '10G' in ifname or 'SFI' in ifname or 'XFI' in ifname: + speed = 10000 + elif '1000BASE' in ifname: + speed = 1000 + return speed + + def get_cmis_state(self): + """ + Get the CMIS states + + Returns: + Dictionary, the states of module, config error and datapath + """ + state = { + 'module_state': self.get_module_state(), + 'config_state': self.get_config_datapath_hostlane_status(), + 'datapath_state': self.get_datapath_state() + } + return state + + def get_cmis_application_selected(self, host_lane): + """ + Get the CMIS selected application code of a host lane + + Args: + host_lane: + Integer, the lane id on the host/system side + + Returns: + Integer, the transceiver-specific application code + """ + ap_code = 0 + if host_lane in range(self.NUM_CHANNELS) and not self.is_flat_memory(): + field = "{}_{}_{}".format(consts.STAGED_CTRL_APSEL_FIELD, 0, host_lane + 1) + ap_code = self.xcvr_eeprom.read(field) >> 4 + + return (ap_code & 0xf) + + def get_cmis_application_matched(self, host_speed, host_lanemask): + """ + Get the CMIS application code that matches the specified host side configurations + + Args: + host_speed: + Integer, the port speed of the host interface + host_lanemask: + Integer, a bitmask of the lanes on the host side + e.g. 0x5 for lane 0 and lane 2. + + Returns: + Integer, the transceiver-specific application code + """ + if host_speed == 0 or host_lanemask == 0: + return 0 + + host_lane_count = 0 + for lane in range(self.NUM_CHANNELS): + if (1 << lane) & host_lanemask == 0: + continue + host_lane_count += 1 + + appl_code = 0 + appl_dict = self.xcvr_eeprom.read(consts.APPLS_ADVT_FIELD) + if appl_dict is None or appl_dict.strip() in ["None", "{}", ""]: + return 0 + appl_dict = ast.literal_eval(appl_dict) + for c in appl_dict.keys(): + d = appl_dict[c] + if d.get('host_lane_count') != host_lane_count: + continue + if self.get_host_speed(d.get('host_electrical_interface_id')) != host_speed: + continue + appl_code = c + break + + return (appl_code & 0xf) + + def has_cmis_application_update(self, host_speed, host_lanemask): + """ + Check for CMIS application update and retrieve the new application code + + Args: + host_speed: + Integer, the port speed of the host interface + host_lanemask: + Integer, a bitmask of the lanes on the host side + e.g. 0x5 for lane 0 and lane 2. + + Returns: + (has_update, new_appl) + """ + if host_speed == 0 or host_lanemask == 0 or self.is_flat_memory(): + return (False, 1) + + app_new = self.get_cmis_application_matched(host_speed, host_lanemask) + if app_new != 1 or host_lanemask != (1 << self.NUM_CHANNELS) - 1: + logger.info("Non-default application is not supported") + return (False, 1) + + app_old = 0 + for lane in range(self.NUM_CHANNELS): + if (1 << lane) & host_lanemask == 0: + continue + if app_old == 0: + app_old = self.get_cmis_application_selected(lane) + elif app_old != self.get_cmis_application_selected(lane): + logger.info("Not all the lanes are in the same application mode") + logger.info("Forcing application update...") + return (True, app_new) + + if app_old == app_new: + skip = True + dp_state = self.get_datapath_state() + conf_state = self.get_config_datapath_hostlane_status() + for lane in range(self.NUM_CHANNELS): + if (1 << lane) & host_lanemask == 0: + continue + name = "DP{}State".format(lane + 1) + if dp_state[name] != CmisCodes.DATAPATH_STATE[4]: + skip = False + break + name = "ConfigStatusLane{}".format(lane + 1) + if conf_state[name] != CmisCodes.CONFIG_STATUS[1]: + skip = False + break + if skip: + return (False, app_old) + + return (True, app_new) + + def set_cmis_application_stop(self, host_lanemask): + """ + Deinitialize the datapath and turn off Tx power to the associated line lanes + + Args: + host_lanemask: + Integer, a bitmask of the lanes on the host side + e.g. 0x5 for lane 0 and lane 2. + + Returns: + Boolean, true if success otherwise false + """ + # D.2.2 Software Deinitialization + self.set_datapath_deinit(host_lanemask) + self.set_lpmode(True) + + # D.1.3 Software Configuration and Initialization + self.tx_disable_channel(host_lanemask, True) + self.set_lpmode(False) + return True + + def set_cmis_application_apsel(self, host_lanemask, appl_code): + """ + Update the selected application code to the specified host lanes + + Args: + host_lanemask: + Integer, a bitmask of the lanes on the host side + e.g. 0x5 for lane 0 and lane 2. + appl_code: + Integer, the desired application code + + Returns: + Boolean, true if success otherwise false + """ + # Update the application selection + for lane in range(self.NUM_CHANNELS): + if (1 << lane) & host_lanemask == 0: + continue + addr = "{}_{}_{}".format(consts.STAGED_CTRL_APSEL_FIELD, 0, lane + 1) + data = appl_code << 4 + self.xcvr_eeprom.write(addr, data) + + # Apply DataPathInit + return self.xcvr_eeprom.write("%s_%d" % (consts.STAGED_CTRL_APPLY_DPINIT_FIELD, 0), host_lanemask) + + def set_cmis_application_start(self, host_lanemask): + """ + Initialize the datapath associated with the specified host lanes, while the Tx power + state of the line side will not be updated. + + Args: + host_lanemask: + Integer, a bitmask of the lanes on the host side + e.g. 0x5 for lane 0 and lane 2. + + Returns: + Boolean, true if success otherwise false + """ + return self.set_datapath_init(host_lanemask) + + def set_cmis_application_txon(self, host_lanemask): + """ + Turn on Tx power of the lanes on the line side associated with the specified host lanes + + Args: + host_lanemask: + Integer, a bitmask of the lanes on the host side + e.g. 0x5 for lane 0 and lane 2. + + Returns: + Boolean, true if success otherwise false + """ + self.tx_disable_channel(host_lanemask, False) + + def get_error_description(self): + dp_state = self.get_datapath_state() + conf_state = self.get_config_datapath_hostlane_status() + for lane in range(self.NUM_CHANNELS): + name = "DP{}State".format(lane + 1) + if dp_state[name] != CmisCodes.DATAPATH_STATE[4]: + return dp_state[name] + + name = "ConfigStatusLane{}".format(lane + 1) + if conf_state[name] != CmisCodes.CONFIG_STATUS[1]: + return conf_state[name] + + state = self.get_module_state() + if state != CmisCodes.MODULE_STATE[3]: + return state + + return None + # TODO: other XcvrApi methods diff --git a/sonic_platform_base/sonic_xcvr/api/public/cmisCDB.py b/sonic_platform_base/sonic_xcvr/api/public/cmisCDB.py index ad321ad30..c3a9c1e15 100644 --- a/sonic_platform_base/sonic_xcvr/api/public/cmisCDB.py +++ b/sonic_platform_base/sonic_xcvr/api/public/cmisCDB.py @@ -26,7 +26,7 @@ def __init__(self, xcvr_eeprom): super(CmisCdbApi, self).__init__(xcvr_eeprom) self.cdb_instance_supported = self.xcvr_eeprom.read(consts.CDB_SUPPORT) self.failed_status_dict = self.xcvr_eeprom.mem_map.codes.CDB_FAIL_STATUS - assert self.cdb_instance_supported != 0 + #assert self.cdb_instance_supported != 0 def cdb1_chkflags(self): ''' diff --git a/sonic_platform_base/sonic_xcvr/fields/consts.py b/sonic_platform_base/sonic_xcvr/fields/consts.py index a675dddb1..bf8205d92 100644 --- a/sonic_platform_base/sonic_xcvr/fields/consts.py +++ b/sonic_platform_base/sonic_xcvr/fields/consts.py @@ -270,6 +270,7 @@ TRANS_CONFIG_FIELD = "TransceiverConfig" MODULE_LEVEL_CONTROL = "ModuleControl" +APPLS_ADVT_FIELD = "Applications Advertisement" CTRLS_ADVT_FIELD = "Supported Controls Advertisement" FLAGS_ADVT_FIELD = "Supported Flags Advertisement" PAGE_SUPPORT_ADVT_FIELD = "Supported Pages Advertisement" @@ -278,6 +279,7 @@ LANE_MON_ADVT_FIELD = "Supported Lane Monitor Advertisement" LANE_DATAPATH_CTRL_FIELD = "Lane Control and Data Path Control" LANE_DATAPATH_STATUS_FIELD = "Lane Status and Data Path Status" +DATAPATH_DEINIT_FIELD = "Data Path Deinit" LEN_MULT_FIELD = "LengthMultiplier" MAX_POWER_FIELD = "MaxPower" MGMT_CHAR_FIELD = "Management Characteristics" @@ -285,6 +287,11 @@ MODULE_CHAR_ADVT_FIELD = "Module Characteristics Advertising" +STAGED_CTRL_FIELD = "Staged Control Set" +STAGED_CTRL_APPLY_DPINIT_FIELD = "Staged Control Set Apply DataPathInit" +STAGED_CTRL_APPLY_IMMEDIATE_FIELD = "Staged Control Set Apply Immediate" +STAGED_CTRL_APSEL_FIELD = "Staged Control Set ApSel" + # C-CMIS # Module configuration support fields diff --git a/sonic_platform_base/sonic_xcvr/fields/public/cmis.py b/sonic_platform_base/sonic_xcvr/fields/public/cmis.py index 88806b1d5..271b1e00a 100644 --- a/sonic_platform_base/sonic_xcvr/fields/public/cmis.py +++ b/sonic_platform_base/sonic_xcvr/fields/public/cmis.py @@ -1,5 +1,7 @@ from ..xcvr_field import NumberRegField +from ..xcvr_field import RegField from .. import consts +from ...codes.public.sff8024 import Sff8024 class CableLenField(NumberRegField): def __init__(self, name, offset, *fields, **kwargs): @@ -11,3 +13,60 @@ def decode(self, raw_data, **decoded_deps): len_mult = decoded_deps.get(consts.LEN_MULT_FIELD) mult = 10 ** (len_mult - 1) return base_len * mult + +class ApplicationAdvertField(RegField): + """ + Interprets application advertising bytes as a string + """ + def __init__(self, name, offset, *fields, **kwargs): + super(ApplicationAdvertField, self).__init__(name, offset, *fields, **kwargs) + self.size = kwargs.get("size") + + def decode(self, raw_data, **decoded_deps): + media_dict = { + 1: Sff8024.NM_850_MEDIA_INTERFACE, + 2: Sff8024.SM_MEDIA_INTERFACE, + 3: Sff8024.PASSIVE_COPPER_MEDIA_INTERFACE, + 4: Sff8024.ACTIVE_CABLE_MEDIA_INTERFACE, + 5: Sff8024.BASE_T_MEDIA_INTERFACE + } + + # Select the media dictionary based on media type(i.e. BYTE 85) + media_if_dict = media_dict.get(raw_data[0]) + host_if_dict = Sff8024.HOST_ELECTRICAL_INTERFACE + + if media_if_dict is None: + return None + + idx = 1 + pos = 1 + dat = {} + while pos < self.size: + appl = { } + + code = raw_data[pos+0] + if code in [0x00, 0xff]: + break + if code in host_if_dict: + appl['host_electrical_interface_id'] = host_if_dict[code] + else: + appl['host_electrical_interface_id'] = 'Unknown' + + code = raw_data[pos+1] + if code in [0x00, 0xff]: + break + if code in media_if_dict: + appl['module_media_interface_id'] = media_if_dict[code] + else: + appl['module_media_interface_id'] = 'Unknown' + + appl['host_lane_count'] = raw_data[pos+2] >> 4 + appl['media_lane_count'] = raw_data[pos+2] & 0xf + appl['host_lane_assignment_options'] = raw_data[pos+3] + appl['media_lane_assignment_options'] = None + + dat[idx] = appl + idx += 1 + pos += 4 + + return str(dat) diff --git a/sonic_platform_base/sonic_xcvr/mem_maps/public/cmis.py b/sonic_platform_base/sonic_xcvr/mem_maps/public/cmis.py index 2586ee662..0f72f54e9 100644 --- a/sonic_platform_base/sonic_xcvr/mem_maps/public/cmis.py +++ b/sonic_platform_base/sonic_xcvr/mem_maps/public/cmis.py @@ -16,6 +16,7 @@ ) from ...fields import consts from ...fields.public.cmis import CableLenField +from ...fields.public.cmis import ApplicationAdvertField class CmisMemMap(XcvrMemMap): def __init__(self, codes): @@ -29,6 +30,7 @@ def __init__(self, codes): # Should contain ONLY Lower page fields self.ADMIN_INFO = RegGroupField(consts.ADMIN_INFO_FIELD, + ApplicationAdvertField(consts.APPLS_ADVT_FIELD, self.getaddr(0x0, 85), size=33), CodeRegField(consts.ID_FIELD, self.getaddr(0x0, 0), self.codes.XCVR_IDENTIFIERS), CodeRegField(consts.ID_ABBRV_FIELD, self.getaddr(0x0, 128), self.codes.XCVR_IDENTIFIER_ABBRV), StringRegField(consts.VENDOR_NAME_FIELD, self.getaddr(0x0, 129), size=16), @@ -170,6 +172,7 @@ def __init__(self, codes): ) self.LANE_DATAPATH_CTRL = RegGroupField(consts.LANE_DATAPATH_CTRL_FIELD, + NumberRegField(consts.DATAPATH_DEINIT_FIELD, self.getaddr(0x10, 128), ro=False), NumberRegField(consts.TX_DISABLE_FIELD, self.getaddr(0x10, 130), ro=False) ) @@ -365,6 +368,17 @@ def __init__(self, codes): NumberRegField(consts.CDB_RPL_LENGTH, self.getaddr(0x9f, 134), size=1, ro=False), NumberRegField(consts.CDB_RPL_CHKCODE, self.getaddr(0x9f, 135), size=1, ro=False), ) + + self.STAGED_CTRL = RegGroupField(consts.STAGED_CTRL_FIELD, + NumberRegField("%s_%d" % (consts.STAGED_CTRL_APPLY_DPINIT_FIELD, 0), + self.getaddr(0x10, 143), ro=False), + NumberRegField("%s_%d" % (consts.STAGED_CTRL_APPLY_IMMEDIATE_FIELD, 0), + self.getaddr(0x10, 144), ro=False), + *(NumberRegField("%s_%d_%d" % (consts.STAGED_CTRL_APSEL_FIELD, 0, lane), + self.getaddr(0x10, 144 + lane), ro=False) + for lane in range(1, 9)) + ) + # TODO: add remaining fields def getaddr(self, page, offset, page_size=128): diff --git a/sonic_platform_base/sonic_xcvr/sfp_optoe_base.py b/sonic_platform_base/sonic_xcvr/sfp_optoe_base.py index 829022456..e5323b045 100644 --- a/sonic_platform_base/sonic_xcvr/sfp_optoe_base.py +++ b/sonic_platform_base/sonic_xcvr/sfp_optoe_base.py @@ -187,3 +187,117 @@ def write_eeprom(self, offset, num_bytes, write_buffer): except (OSError, IOError): return False return True + + def get_lpmode(self): + api = self.get_xcvr_api() + return api.get_lpmode() if api is not None else False + + def set_lpmode(self, lpmode): + api = self.get_xcvr_api() + return api.set_lpmode(lpmode) if api is not None else False + + def get_module_state(self): + api = self.get_xcvr_api() + return api.get_module_state() if api is not None else None + + def get_error_description(self): + api = self.get_xcvr_api() + return api.get_error_description() if api is not None else None + + def is_flat_memory(self): + api = self.get_xcvr_api() + return api.is_flat_memory() if api is not None else True + + def get_cmis_state(self): + """ + Retrieve the CMIS transceiver states including ModuleState, DataPath and ConfigError + """ + api = self.get_xcvr_api() + return api.get_cmis_state() if api is not None else None + + def has_cmis_application_update(self, host_speed, host_lanes): + """ + Check for CMIS application update and the new application code + + Args: + host_speed: + Integer, the port speed of the host interface + host_lanes: + Integer List, a list of the 0-based lane indexes of the host interface + + Returns: + (updated, appl_code) + """ + api = self.get_xcvr_api() + if api is not None: + return api.has_cmis_application_update(host_speed, host_lanes) + return (False, 1) + + def set_cmis_application_stop(self, host_lanes): + """ + (Stage 1) Data Path Deinitialization Tx power disable + + Args: + host_lanes: + Integer List, a list of the 0-based lane indexes of the host interface + + Returns: + A boolean, True if successful, False if not + """ + ret = False + api = self.get_xcvr_api() + if api is not None: + ret = api.set_cmis_application_stop(host_lanes) + return ret + + def set_cmis_application_apsel(self, host_lanes, appl_code=1): + """ + (Stage 2) Update the application selection + + Args: + host_lanes: + Integer List, a list of the 0-based lane indexes of the host interface + appl_code: + Integer, the desired application code + + Returns: + A boolean, True if successful, False if not + """ + ret = False + api = self.get_xcvr_api() + if api is not None: + ret = api.set_cmis_application_apsel(host_lanes, appl_code) + return ret + + def set_cmis_application_start(self, host_lanes): + """ + (Stage 3) Initialize the new data path + + Args: + host_lanes: + Integer List, a list of the 0-based lane indexes of the host interface + + Returns: + A boolean, True if successful, False if not + """ + ret = False + api = self.get_xcvr_api() + if api is not None: + ret = api.set_cmis_application_start(host_lanes) + return ret + + def set_cmis_application_txon(self, host_lanes): + """ + (Stage 4) Initialize the new data path + Args: + host_lanes: + Integer List, a list of the 0-based lane indexes of the host interface + + Returns: + A boolean, True if successful, False if not + """ + ret = False + api = self.get_xcvr_api() + if api is not None: + ret = api.set_cmis_application_txon(host_lanes) + return ret diff --git a/tests/sonic_xcvr/test_cmis.py b/tests/sonic_xcvr/test_cmis.py index 2c0d730d3..881035154 100644 --- a/tests/sonic_xcvr/test_cmis.py +++ b/tests/sonic_xcvr/test_cmis.py @@ -1132,6 +1132,8 @@ def test_get_transceiver_info(self, mock_response, expected): self.api.get_module_media_type.return_value = mock_response[13] self.api.get_module_hardware_revision = MagicMock() self.api.get_module_hardware_revision.return_value = '0.0' + self.api.is_flat_memory = MagicMock() + self.api.is_flat_memory.return_value = False result = self.api.get_transceiver_info() assert result == expected