Update spf platform test related to error status, due to sonic-net/so…

…nic-buildimage#20964

When software control is enabled, the port error status is as follows:
1. For active module, the expected state is OK
2. For cmis passive module, the expected state is ModuleLowPwr
3. For non cmis passive module, the expected state is 'Not supported'

when software control is disabled, the port error status keep the original behaviour

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