[Mellanox] Update spf test related to error status when sw control is enabled

tests/common/platform/transceiver_utils.py

@@ -387,3 +387,69 @@ def is_passive_cable(sfp_eeprom_info):
                         "CR" in spec_compliance.get("Extended Specification Compliance", " "):
                     return True
     return False
+
+
+def get_passive_cable_port_list(dut):
+    passive_cable_port_list = []
+    cmd_show_eeprom = "sudo sfputil show eeprom -d"
+    eeprom_infos = dut.command(cmd_show_eeprom)['stdout']
+    eeprom_infos = parse_sfp_eeprom_infos(eeprom_infos)
+    for port_name, eeprom_info in eeprom_infos.items():
+        if is_passive_cable(eeprom_info):
+            logging.info(f"{port_name} is passive cable")
+            passive_cable_port_list.append(port_name)
+    logging.info(f"Ports with passive cable are: {passive_cable_port_list}")
+    return passive_cable_port_list
+
+
+def get_passive_cable_port_list(dut):
+    passive_cable_port_list = []
+    cmd_show_eeprom = "sudo sfputil show eeprom -d"
+    eeprom_infos = dut.command(cmd_show_eeprom)['stdout']
+    eeprom_infos = parse_sfp_eeprom_infos(eeprom_infos)
+    for port_name, eeprom_info in eeprom_infos.items():
+        if is_passive_cable(eeprom_info):
+            logging.info(f"{port_name} is passive cable")
+            passive_cable_port_list.append(port_name)
+    logging.info(f"Ports with passive cable are: {passive_cable_port_list}")
+    return passive_cable_port_list
+
+
+def get_cmis_cable_port_list(dut):
+    cmis_cable_port_list = []
+    cmd_show_eeprom = "sudo sfputil show eeprom -d"
+    eeprom_infos = dut.command(cmd_show_eeprom)['stdout']
+    eeprom_infos = parse_sfp_eeprom_infos(eeprom_infos)
+    for port_name, eeprom_info in eeprom_infos.items():
+        if 'CMIS Revision' in eeprom_info:
+            logging.info(f"{port_name} is cmis cable")
+            cmis_cable_port_list.append(port_name)
+    logging.info(f"CMIS cables are: {cmis_cable_port_list}")
+    return cmis_cable_port_list
+
+
+def get_port_expected_error_state_for_mellanox_device_on_sw_control_enabled(intf, passive_cable_ports, cmis_cable_ports):
+    expected_state = 'OK'
+    if intf in passive_cable_ports:
+        # for active module, the expected state is OK
+        # for cmis passive module, the expected state is ModuleLowPwr
+        # for non cmis passive module, the expected state is Not supported
+        expected_state = 'ModuleLowPwr' if intf in cmis_cable_ports else 'Not supported'
+    logging.info(f"port {intf}, expected error state:{expected_state}")
+    return expected_state
+
+
+def is_sw_control_enabled(duthost, port_index):
+    """
+    @summary: This method is for checking if software control SAI attribute set to 1 in sai.profile
+    @param: duthosts: duthosts fixture
+    """
+    sw_control_enabled = False
+    module_path = f'/sys/module/sx_core/asic0/module{int(port_index) - 1}'
+    cmd_check_if_exist_conctrol_file = f'ls {module_path} | grep control'
+    if duthost.shell(cmd_check_if_exist_conctrol_file, module_ignore_errors=True)['stdout']:
+        cmd_get_control_value = f"sudo cat {module_path}/control"
+        if duthost.shell(cmd_get_control_value)['stdout'] == "1":
+            sw_control_enabled = True
+    logging.info(f'The sw control enable of port index {port_index} is {sw_control_enabled}')
+    return sw_control_enabled

tests/platform_tests/api/test_sfp.py

@@ -11,6 +11,11 @@
 from tests.common.fixtures.conn_graph_facts import conn_graph_facts     # noqa F401
 from tests.common.fixtures.duthost_utils import shutdown_ebgp           # noqa F401
 from tests.common.platform.device_utils import platform_api_conn    # noqa F401
+from tests.common.platform.transceiver_utils import is_sw_control_enabled,\
+    get_port_expected_error_state_for_mellanox_device_on_sw_control_enabled
+from tests.common.mellanox_data import is_mellanox_device
+from collections import defaultdict
+
 
 from .platform_api_test_base import PlatformApiTestBase
 
@@ -47,6 +52,10 @@ def setup(request, duthosts, enum_rand_one_per_hwsku_hostname,
     physical_port_index_map = get_physical_port_indices(duthost, physical_intfs)
     sfp_setup["physical_port_index_map"] = physical_port_index_map
 
+    sfp_setup["index_physical_port_map"] = defaultdict(list)
+    for port, index in physical_port_index_map.items():
+        sfp_setup["index_physical_port_map"][index].append(port)
+
     sfp_port_indices = set([physical_port_index_map[intf] for intf in list(physical_port_index_map.keys())])
     sfp_setup["sfp_port_indices"] = sorted(sfp_port_indices)
 
@@ -852,23 +861,31 @@ def test_power_override(self, duthosts, enum_rand_one_per_hwsku_hostname, localh
         self.assert_expectations()
 
     def test_get_error_description(self, duthosts, enum_rand_one_per_hwsku_hostname, localhost,
-                                   platform_api_conn):      # noqa F811
+                                   platform_api_conn, passive_cable_ports, cmis_cable_ports):  # noqa F811
         """This function tests get_error_description() API (supported on 202106 and above)"""
-        skip_release(duthosts[enum_rand_one_per_hwsku_hostname], ["201811", "201911", "202012"])
+        duthost = duthosts[enum_rand_one_per_hwsku_hostname]
+        skip_release(duthost, ["201811", "201911", "202012"])
 
         for i in self.sfp_setup["sfp_test_port_indices"]:
             error_description = sfp.get_error_description(platform_api_conn, i)
             if self.expect(error_description is not None,
                            "Unable to retrieve transceiver {} error description".format(i)):
                 if "Not implemented" in error_description:
                     pytest.skip("get_error_description isn't implemented. Skip the test")
-                if "Not supported" in error_description:
+
+                expected_state = 'OK'
+                if is_mellanox_device(duthost) and is_sw_control_enabled(duthost, i):
+                    intf = self.sfp_setup["index_physical_port_map"][i][0]
+                    expected_state = get_port_expected_error_state_for_mellanox_device_on_sw_control_enabled(
+                        intf, passive_cable_ports[duthost.hostname], cmis_cable_ports[duthost.hostname])
+                elif "Not supported" in error_description:
                     logger.warning("test_get_error_description: Skipping transceiver {} as error description not "
                                    "supported on this port)".format(i))
                     continue
                 if self.expect(isinstance(error_description, str) or isinstance(error_description, str),
                                "Transceiver {} error description appears incorrect".format(i)):
-                    self.expect(error_description == "OK", "Transceiver {} is not present".format(i))
+                    self.expect(error_description == expected_state,
+                                f"Transceiver {i} is not {expected_state}, actual state is:{error_description}.")
         self.assert_expectations()
 
     def test_thermals(self, platform_api_conn):     # noqa F811

tests/platform_tests/conftest.py

@@ -5,7 +5,8 @@
 from tests.common.mellanox_data import is_mellanox_device
 from .args.counterpoll_cpu_usage_args import add_counterpoll_cpu_usage_args
 from tests.common.helpers.mellanox_thermal_control_test_helper import suspend_hw_tc_service, resume_hw_tc_service
-from tests.common.platform.transceiver_utils import get_ports_with_flat_memory
+from tests.common.platform.transceiver_utils import get_ports_with_flat_memory, \
+    get_passive_cable_port_list, get_cmis_cable_port_list
 
 
 @pytest.fixture(autouse=True, scope="module")
@@ -208,3 +209,21 @@ def port_list_with_flat_memory(duthosts):
         ports_with_flat_memory.update({dut.hostname: get_ports_with_flat_memory(dut)})
     logging.info(f"port list with flat memory: {ports_with_flat_memory}")
     return ports_with_flat_memory
+
+
+@pytest.fixture(scope="module")
+def passive_cable_ports(duthosts):
+    passive_cable_ports = {}
+    for dut in duthosts:
+        passive_cable_ports.update({dut.hostname: get_passive_cable_port_list(dut)})
+    logging.info(f"passive_cable_ports: {passive_cable_ports}")
+    return passive_cable_ports
+
+
+@pytest.fixture(scope="module")
+def cmis_cable_ports(duthosts):
+    cmis_cable_ports = {}
+    for dut in duthosts:
+        cmis_cable_ports.update({dut.hostname: get_cmis_cable_port_list(dut)})
+    logging.info(f"cmis_cable_ports: {cmis_cable_ports}")
+    return cmis_cable_ports

tests/platform_tests/sfp/test_sfputil.py

@@ -17,6 +17,10 @@
 from tests.common.fixtures.duthost_utils import shutdown_ebgp   # noqa F401
 from tests.common.port_toggle import default_port_toggle_wait_time
 from tests.common.platform.interface_utils import get_physical_port_indices
+from tests.common.mellanox_data import is_mellanox_device
+from tests.common.platform.transceiver_utils import is_sw_control_enabled,\
+    get_port_expected_error_state_for_mellanox_device_on_sw_control_enabled
+
 
 cmd_sfp_presence = "sudo sfputil show presence"
 cmd_sfp_eeprom = "sudo sfputil show eeprom"
@@ -338,7 +342,8 @@ def test_check_sfputil_presence(duthosts, enum_rand_one_per_hwsku_frontend_hostn
 @pytest.mark.parametrize("cmd_sfp_error_status",
                          ["sudo sfputil show error-status", "sudo sfputil show error-status --fetch-from-hardware"])
 def test_check_sfputil_error_status(duthosts, enum_rand_one_per_hwsku_frontend_hostname,
-                                    enum_frontend_asic_index, conn_graph_facts, cmd_sfp_error_status, xcvr_skip_list):
+                                    enum_frontend_asic_index, conn_graph_facts, cmd_sfp_error_status, xcvr_skip_list,
+                                    passive_cable_ports, cmis_cable_ports):
     """
     @summary: Check SFP error status using 'sfputil show error-status'
               and 'sfputil show error-status --fetch-from-hardware'
@@ -355,14 +360,22 @@ def test_check_sfputil_error_status(duthosts, enum_rand_one_per_hwsku_frontend_h
     if "NOT implemented" in sfp_error_status['stdout']:
         pytest.skip("Skip test as error status isn't supported")
     parsed_presence = parse_output(sfp_error_status["stdout_lines"][2:])
+    physical_port_index_map = get_physical_port_indices(duthost, conn_graph_facts["device_conn"][duthost.hostname])
     for intf in dev_conn:
         if intf not in xcvr_skip_list[duthost.hostname]:
-            if "Not supported" in sfp_error_status['stdout']:
-                logger.warning("test_check_sfputil_error_status: Skipping transceiver {} as error status not "
-                               "supported on this port)".format(intf))
+            expected_state = 'OK'
+            intf_index = physical_port_index_map[intf]
+            if cmd_sfp_error_status == "sudo sfputil show error-status --fetch-from-hardware"\
+                    and is_mellanox_device(duthost) and is_sw_control_enabled(duthost, intf_index):
+                expected_state = get_port_expected_error_state_for_mellanox_device_on_sw_control_enabled(
+                    intf, passive_cable_ports[duthost.hostname], cmis_cable_ports[duthost.hostname])
+            elif "Not supported" in sfp_error_status['stdout']:
+                logger.warning("test_check_sfputil_error_status: Skipping transceiver {} as error status "
+                               "not supported on this port)".format(intf))
                 continue
-            assert intf in parsed_presence, "Interface is not in output of '{}'".format(cmd_sfp_presence)
-            assert parsed_presence[intf] == "OK", "Interface error status is not 'OK'"
+            assert intf in parsed_presence, "Interface is not in output of '{}'".format(cmd_sfp_error_status)
+            assert parsed_presence[intf] == expected_state, \
+                f"Interface {intf}'s error status is not {expected_state}, actual state is:{parsed_presence[intf]}."
 
 
 def test_check_sfputil_eeprom(duthosts, enum_rand_one_per_hwsku_frontend_hostname,

tests/platform_tests/sfp/util.py

@@ -12,9 +12,9 @@ def parse_output(output_lines):
     res = {}
     for line in output_lines:
         fields = line.split()
-        if len(fields) != 2:
+        if len(fields) < 2:
             continue
-        res[fields[0]] = fields[1]
+        res[fields[0]] = line.replace(fields[0], '').strip()
     return res