diff --git a/tests/common/platform/interface_utils.py b/tests/common/platform/interface_utils.py index 1036c87f93..43ca304ffd 100644 --- a/tests/common/platform/interface_utils.py +++ b/tests/common/platform/interface_utils.py @@ -7,6 +7,7 @@ import re import logging import json +import functools from collections import defaultdict from natsort import natsorted from .transceiver_utils import all_transceivers_detected @@ -101,9 +102,13 @@ def check_interface_status(dut, asic_index, interfaces, xcvr_skip_list): output = dut.command("show interface description") intf_status = parse_intf_status(output["stdout_lines"][2:]) if dut.is_multi_asic: - check_intf_presence_command = 'show interface transceiver presence -n {} {}'.format(namespace, {}) + check_intf_presence_command = 'show interface transceiver presence -n {}'.format(namespace) else: - check_intf_presence_command = 'show interface transceiver presence {}' + check_intf_presence_command = 'show interface transceiver presence' + check_inerfaces_presence_output = dut.command(check_intf_presence_command)["stdout_lines"][2:] + check_inerfaces_presence_output = ( + {ports_presence.split()[0]: ports_presence.split()[1] for ports_presence in check_inerfaces_presence_output} + ) for intf in interfaces: expected_oper = "up" if intf in mg_ports else "down" expected_admin = "up" if intf in mg_ports else "down" @@ -121,10 +126,10 @@ def check_interface_status(dut, asic_index, interfaces, xcvr_skip_list): # Cross check the interface SFP presence status if intf not in xcvr_skip_list[dut.hostname]: - check_presence_output = dut.command(check_intf_presence_command.format(intf)) - presence_list = check_presence_output["stdout_lines"][2].split() - assert intf in presence_list, "Wrong interface name in the output: %s" % str(presence_list) - assert 'Present' in presence_list, "Status is not expected, presence status: %s" % str(presence_list) + assert intf in check_inerfaces_presence_output, "Wrong interface name in the output for: %s" % str(intf) + interface_presence = check_inerfaces_presence_output.get(intf, '') + assert 'Present' in interface_presence, \ + "Status is not expected, presence status: %s" % str({intf: interface_presence}) logging.info("Check interface status using the interface_facts module") intf_facts = dut.interface_facts(up_ports=mg_ports, namespace=namespace)["ansible_facts"] @@ -164,6 +169,7 @@ def check_interface_information(dut, asic_index, interfaces, xcvr_skip_list): return True +@functools.lru_cache(maxsize=1) def get_port_map(dut, asic_index=None): """ @summary: Get the port mapping info from the DUT @@ -216,7 +222,7 @@ def get_physical_port_indices(duthost, logical_intfs=None): asic_subcommand = f'-n asic{asic_index}' if asic_index is not None else '' cmd_keys = f'sonic-db-cli {asic_subcommand} CONFIG_DB KEYS "PORT|Ethernet*"' cmd_hget = f'sonic-db-cli {asic_subcommand} CONFIG_DB HGET $key index' - cmd = f'for key in $({cmd_keys}); do echo "$key : $({cmd_hget})" ; done' + cmd = f'for key in $({cmd_keys}); do echo "$key : $({cmd_hget})" ; done' # noqa: E702,E203 cmd_out = duthost.command(cmd, _uses_shell=True)["stdout_lines"] cmd_out_dict = {} for line in cmd_out: @@ -282,7 +288,7 @@ def get_fec_eligible_interfaces(duthost, supported_speeds): if oper == "up" and speed in supported_speeds: interfaces.append(intf_name) else: - logging.info(f"Skip for {intf_name}: oper_state:{oper} speed:{speed}") + logging.info(f"Skip for {intf_name}: oper_state: {oper} speed: {speed}") return interfaces diff --git a/tests/platform_tests/test_sequential_restart.py b/tests/platform_tests/test_sequential_restart.py index 32cfa4cbe8..a68b536eb0 100644 --- a/tests/platform_tests/test_sequential_restart.py +++ b/tests/platform_tests/test_sequential_restart.py @@ -76,10 +76,10 @@ def restart_service_and_check(localhost, dut, enum_frontend_asic_index, service, interface_wait_time = 300 if dut.facts["platform"] == "x86_64-cel_e1031-r0": interface_wait_time = 900 - pytest_assert(wait_until(interface_wait_time, 20, 0, check_interface_information, dut, - enum_frontend_asic_index, interfaces, xcvr_skip_list), - "Not all interface information are detected within {} seconds".format(interface_wait_time)) - + if interfaces: + pytest_assert(wait_until(interface_wait_time, 20, 0, check_interface_information, dut, + enum_frontend_asic_index, interfaces, xcvr_skip_list), + "Not all interface information are detected within {} seconds".format(interface_wait_time)) logging.info("Check transceiver status on asic %s" % enum_frontend_asic_index) check_transceiver_basic(dut, enum_frontend_asic_index, interfaces, xcvr_skip_list)