Optimize check_interface_status function. (sonic-net#21055)

Add cache to get_ports_map.
Call transceiver presence once for all ports together.

Author: OriTrabelsi

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
 

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)