[Mellanox] Update the test_check_sfp_eeprom due to design change

tests/common/platform/interface_utils.py

@@ -11,6 +11,8 @@
 from collections import defaultdict
 from natsort import natsorted
 from .transceiver_utils import all_transceivers_detected
+import ast
+from tests.common.mellanox_data import is_mellanox_device
 
 
 def parse_intf_status(lines):
@@ -318,3 +320,44 @@ def get_lport_to_first_subport_mapping(duthost, logical_intfs=None):
     first_subport_dict = {k: pport_to_lport_mapping[v][0] for k, v in physical_port_indices.items()}
     logging.debug("First subports mapping: {}".format(first_subport_dict))
     return first_subport_dict
+
+
+def get_ports_with_flat_memory(dut, conn_graph_facts):
+    """
+    This method is to get ports with flat memory
+    """
+    port_indexes_with_flat_memory = []
+    if is_mellanox_device(dut):
+        port_indexes_with_flat_memory = get_port_indexes_with_flat_memory(dut)
+
+    physical_intfs = conn_graph_facts["device_conn"][dut.hostname]
+    physical_port_index_map = get_physical_port_indices(dut, physical_intfs)
+    ports_with_flat_memory = []
+    for port, index in physical_port_index_map.items():
+        if index in port_indexes_with_flat_memory:
+            ports_with_flat_memory.append(port)
+    logging.info(f"Ports with flat memory: {ports_with_flat_memory}")
+    return ports_with_flat_memory
+
+
+def get_port_indexes_with_flat_memory(dut):
+    """
+    This method is to get port indexes with flat memory
+    """
+    cmd = """
+cat << EOF > get_port_indexes_with_flat_memory.py
+import sonic_platform.platform as P
+num_sfps = P.Platform().get_chassis().get_num_sfps()
+port_indexes_with_flat_memory = []
+for i in range(1, num_sfps + 1):
+    is_flat_memory = P.Platform().get_chassis().get_sfp(i).get_xcvr_api().is_flat_memory()
+    if is_flat_memory:
+        port_indexes_with_flat_memory.append(i)
+print(port_indexes_with_flat_memory)
+EOF
+"""
+    dut.shell(cmd)
+    port_indexes_with_flat_memory = dut.shell("python3 get_port_indexes_with_flat_memory.py")["stdout"]
+    port_indexes_with_flat_memory = ast.literal_eval(port_indexes_with_flat_memory)
+    logging.info(f"Port indexes with flat memory: {port_indexes_with_flat_memory}")
+    return port_indexes_with_flat_memory

tests/common/platform/transceiver_utils.py

@@ -207,19 +207,6 @@ def get_sfp_eeprom_map_per_port(eeprom_infos):
     return sfp_eeprom_map_per_port
 
 
-def get_ports_with_flat_memory(dut):
-    ports_with_flat_memory = []
-    cmd_show_eeprom = "sudo sfputil show eeprom -d"
-
-    eeprom_infos = dut.command(cmd_show_eeprom, module_ignore_errors=True)['stdout']
-    sfp_eerpom_map_per_port = get_sfp_eeprom_map_per_port(eeprom_infos)
-    for port_name, sfp_eeprom in sfp_eerpom_map_per_port.items():
-        if "DOM values not supported for flat memory module" in " ".join(sfp_eeprom):
-            ports_with_flat_memory.append(port_name)
-    logging.info(f"Ports with flat memory: {ports_with_flat_memory}")
-    return ports_with_flat_memory
-
-
 def parse_one_sfp_eeprom_info(sfp_eeprom_info):
     """
     Parse the one sfp eeprom info, return top_key, sfp_eeprom_info_dict

tests/platform_tests/conftest.py

@@ -8,9 +8,9 @@
 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, \
-    get_passive_cable_port_list, get_cmis_cable_ports_and_ver
+from tests.common.platform.transceiver_utils import get_passive_cable_port_list, get_cmis_cable_ports_and_ver
 from tests.common.helpers.firmware_helper import PLATFORM_COMP_PATH_TEMPLATE
+from tests.common.platform.interface_utils import get_ports_with_flat_memory
 
 logger = logging.getLogger(__name__)
 
@@ -238,10 +238,10 @@ def dpu_npu_port_list(duthosts):
 
 
 @pytest.fixture(scope="module")
-def port_list_with_flat_memory(duthosts):
+def port_list_with_flat_memory(duthosts, conn_graph_facts):
     ports_with_flat_memory = {}
     for dut in duthosts:
-        ports_with_flat_memory.update({dut.hostname: get_ports_with_flat_memory(dut)})
+        ports_with_flat_memory.update({dut.hostname: get_ports_with_flat_memory(dut, conn_graph_facts)})
     logging.info(f"port list with flat memory: {ports_with_flat_memory}")
     return ports_with_flat_memory
 

tests/platform_tests/mellanox/test_check_sfp_eeprom.py

@@ -2,11 +2,12 @@
 import allure
 
 from tests.common.fixtures.conn_graph_facts import conn_graph_facts  # noqa: F401
-from .util import check_sfp_eeprom_info, is_support_dom, get_pci_cr0_path, get_pciconf0_path
+from .util import check_sfp_eeprom_info
 from tests.common.platform.transceiver_utils import parse_sfp_eeprom_infos
 from tests.common.utilities import wait_until
 from tests.common.helpers.assertions import pytest_assert
 from tests.common.platform.processes_utils import check_pmon_uptime_minutes
+import logging
 
 pytestmark = [
     pytest.mark.asic('mellanox', 'nvidia-bluefield'),
@@ -19,46 +20,26 @@
 
 @pytest.fixture(scope="module", autouse=True)
 def sfp_test_intfs_to_dom_map(duthosts, rand_one_dut_hostname, conn_graph_facts, xcvr_skip_list,  # noqa: F811
-                              get_sw_control_ports):
+                              get_sw_control_ports, port_list_with_flat_memory):  # noqa: F811
     '''
     This fixture is to get map sfp test intfs to dom
     '''
     duthost = duthosts[rand_one_dut_hostname]
 
-    ports_map = duthost.config_facts(host=duthost.hostname, source="running")[
-        'ansible_facts']["PORT"]
-    port_name_to_index_map = dict(
-        [(port, value["index"]) for port, value in list(ports_map.items())])
-
     sfp_test_intf_list = list(
         conn_graph_facts["device_conn"][duthost.hostname].keys())
 
     if get_sw_control_ports:
         # Exclude get_sw_control_ports from sfp_test_intf_list
         sfp_test_intf_list = [port for port in sfp_test_intf_list if port not in get_sw_control_ports]
 
-    intf_with_dom_dict = {}
-
     sfp_test_intfs_to_dom_map_dict = {}
-    platform = duthost.facts['platform']
-    dpu_platform_list = ["arm64-nvda_bf-9009d3b600cvaa", "arm64-nvda_bf-9009d3b600svaa"]
-    pci_path = get_pciconf0_path(duthost) if platform in dpu_platform_list else get_pci_cr0_path(duthost)
 
     for intf in sfp_test_intf_list:
         if intf not in xcvr_skip_list[duthost.hostname]:
-            port_index = port_name_to_index_map[intf]
-            original_port_index = port_index
-            if port_index in intf_with_dom_dict:
-                inft_support_dom = intf_with_dom_dict[port_index]
-            else:
-                if platform in dpu_platform_list and port_index == '2':
-                    pci_path = "{}.1".format(pci_path)
-                    port_index = '1'
-                inft_support_dom = is_support_dom(
-                    duthost, port_index, pci_path)
-                intf_with_dom_dict[original_port_index] = inft_support_dom
+            inft_support_dom = False if intf in port_list_with_flat_memory[duthost.hostname] else True
             sfp_test_intfs_to_dom_map_dict[intf] = inft_support_dom
-
+    logging.info(f"sfp_test_intfs_to_dom_map_dict: {sfp_test_intfs_to_dom_map_dict}")
     return sfp_test_intfs_to_dom_map_dict
 
 

tests/platform_tests/mellanox/util.py

@@ -1,6 +1,5 @@
 import re
 import logging
-import ast
 
 
 def check_sfp_eeprom_info(duthost, sfp_eeprom_info, is_support_dom, show_eeprom_cmd, is_flat_memory):
@@ -142,48 +141,6 @@ def check_dom_monitor_key_and_data_format(expected_keys_and_pattern_dict, dom_mo
                 key, dom_monitor_data[key], pattern)
 
 
-def is_support_dom(duthost, port_index, pic_cr0_path):
-    """
-    This method is to check if cable support dom
-    1. For 202012 branch(It not support mlxlink tool, so use get_transceiver_bulk_status to judge if it support dom)
-      1) Get get transceiver bulk status
-      2) Return True, When any one value for all parameters including power, bias,temperature and voltage is not in
-         ['N/A', '0.0', 0.0, '0.0000mA', '-inf'], else False.
-    2. For other branches apart from 202012
-      1) Get the pci_cro info by mlxlink tool
-      2) Return false, when all values of 5 fields
-       (Temperature|Voltage|Bias Current|Rx Power Current|Tx Power Current) are N/A, else True
-    """
-    if duthost.sonic_release in ["202012"]:
-        bulk_status_str = get_transceiver_bulk_status(duthost, port_index)
-        bulk_status_str = bulk_status_str.replace('-inf', '\'-inf\'')
-        bulk_status_dict = ast.literal_eval(bulk_status_str)
-        for k, v in list(bulk_status_dict.items()):
-            if "power" in k or "bias" in k or "temperature" in k or "voltage" in k:
-                if v not in ['N/A', '0.0', 0.0, '0.0000mA', '-inf']:
-                    logging.info("Port {} support dom".format(port_index))
-                    return True
-        logging.info("Port {} doesn't support dom".format(port_index))
-        return False
-    else:
-        pattern_for_dom_check = r'^(Temperature|Voltage|Bias Current|Rx Power Current|Tx Power Current).*: N\/A.*'
-        pci_cr0 = get_mlxlink_pci_cr0(duthost, pic_cr0_path, port_index)
-
-        check_support_dom_filed_number = 5
-        not_support_dom_field_counter = 0
-        for line in pci_cr0.split("\n"):
-            res = re.match(pattern_for_dom_check, line)
-            if res:
-                not_support_dom_field_counter += 1
-                logging.info(
-                    "Find {} Value is N/A: {}".format(not_support_dom_field_counter, line))
-            if not_support_dom_field_counter >= check_support_dom_filed_number:
-                logging.info("Port {} doesn't support dom".format(port_index))
-                return False
-        logging.info("Port {} support dom".format(port_index))
-        return True
-
-
 def get_transceiver_bulk_status(duthost, port_index):
     """
     This method is to get transceiver bulk status
@@ -197,25 +154,3 @@ def get_transceiver_bulk_status(duthost, port_index):
 """.format(port_index)
     duthost.shell(cmd)
     return duthost.command("python3 get_transceiver_bulk_status.py")["stdout"]
-
-
-def get_mlxlink_pci_cr0(duthost, pci_cr0_path, port_index):
-    """
-    This method is to get the info of /dev/mst/*_pci_cr0
-    """
-    cmd = "sudo mlxlink -d {} -p {} -m".format(pci_cr0_path, port_index)
-    return duthost.command(cmd)["stdout"]
-
-
-def get_pci_cr0_path(duthost):
-    """
-    This method is to get path for /dev/mst/*_pci_cr0
-    """
-    return duthost.shell('ls /dev/mst/*_pci_cr0')['stdout'].strip()
-
-
-def get_pciconf0_path(duthost):
-    """
-    This method is to get path for /dev/mst/*_pciconf0
-    """
-    return duthost.shell('ls /dev/mst/*_pciconf0')['stdout'].strip()