Modify tests to run on zero ports systems (#3663)

Allow systems which starts with zero ports to run platform tests without failing even if the tests skip the interface checks.

Change the port verification to relay on config_db.json instead of minigraph.xml
Change places where test fail in case of zero ports on setup

What is the motivation for this PR?
Allow systems which starts with zero ports to run platform tests without failing even if the tests skip the interface checks.

How did you do it?
Rerun pre defined list of tests where system is defined with zero ports and modify tests one by one to pass but to keep functional as possible (taking into account that interfaces are not configured)

How did you verify/test it?
Rerun tests and ensure they are passing on a system with ports and with zero ports.

Signed-off-by: Sharon Lutati <slutati@nvidia.com>

tests/common/platform/interface_utils.py

@@ -1,6 +1,7 @@
 """
 Helper script for checking status of interfaces
 
+
 This script contains re-usable functions for checking status of interfaces on SONiC.
 """
 
@@ -44,47 +45,39 @@ def check_interface_status(dut, asic_index, interfaces, xcvr_skip_list):
     asichost = dut.asic_instance(asic_index)
     namespace = asichost.get_asic_namespace()
     logging.info("Check interface status using cmd 'show interface'")
-    #TODO Remove this logic when minigraph facts supports namespace in multi_asic
-    mg_ports = dut.minigraph_facts(host=dut.hostname)["ansible_facts"]["minigraph_ports"]
-    if asic_index is not None:
-        portmap = get_port_map(dut, asic_index)
-        # Check if the interfaces of this AISC is present in mg_ports
-        interface_list = {k:v for k, v in portmap.items() if k in mg_ports}
-        mg_ports = interface_list
+    ports = get_port_map(dut, asic_index)
     output = dut.command("show interface description")
     intf_status = parse_intf_status(output["stdout_lines"][2:])
     check_intf_presence_command = 'show interface transceiver presence {}'
     for intf in interfaces:
-        expected_oper = "up" if intf in mg_ports else "down"
-        expected_admin = "up" if intf in mg_ports else "down"
+        expected_oper = "up" if intf in ports else "down"
+        expected_admin = "up" if intf in ports else "down"
         if intf not in intf_status:
             logging.info("Missing status for interface %s" % intf)
             return False
         if intf_status[intf]["oper"] != expected_oper:
-            logging.info("Oper status of interface %s is %s, expected '%s'" % (intf, intf_status[intf]["oper"],
-                                                                               expected_oper))
+            logging.info("Oper status of interface %s is %s, expected '%s'" % (intf, intf_status[intf]["oper"], expected_oper))
             return False
         if intf_status[intf]["admin"] != expected_admin:
-            logging.info("Admin status of interface %s is %s, expected '%s'" % (intf, intf_status[intf]["admin"],
-                                                                                expected_admin))
+            logging.info("Admin status of interface %s is %s, expected '%s'" % (intf, intf_status[intf]["admin"], expected_admin))
             return False
 
         # Cross check the interface SFP presence status
-        if intf not in xcvr_skip_list[dut.hostname]:
+        if xcvr_skip_list and 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)
 
-    logging.info("Check interface status using the interface_facts module")
-    intf_facts = dut.interface_facts(up_ports=mg_ports, namespace=namespace)["ansible_facts"]
+    intf_facts = dut.interface_facts(up_ports=ports, namespace=namespace)["ansible_facts"]
     down_ports = intf_facts["ansible_interface_link_down_ports"]
     if len(down_ports) != 0:
         logging.info("Some interfaces are down: %s" % str(down_ports))
         return False
 
     return True
 
+
 # This API to check the interface information actoss all front end ASIC's
 def check_all_interface_information(dut, interfaces, xcvr_skip_list):
     for asic_index in dut.get_frontend_asic_ids():
@@ -100,6 +93,7 @@ def check_all_interface_information(dut, interfaces, xcvr_skip_list):
 
     return True
 
+
 # This API to check the interface information per asic.
 def check_interface_information(dut, asic_index, interfaces, xcvr_skip_list):
     if not all_transceivers_detected(dut, asic_index, interfaces, xcvr_skip_list):
@@ -111,6 +105,7 @@ def check_interface_information(dut, asic_index, interfaces, xcvr_skip_list):
 
     return True
 
+
 def get_port_map(dut, asic_index=None):
     """
     @summary: Get the port mapping info from the DUT

tests/common/plugins/sanity_check/checks.py

@@ -135,10 +135,12 @@ def _check_interfaces_on_dut(*args, **kwargs):
         check_result = {"failed": True, "check_item": "interfaces", "host": dut.hostname}
         for asic in dut.asics:
             ip_interfaces = []
+            phy_interfaces = []
             cfg_facts = asic.config_facts(host=dut.hostname,
                                           source="persistent", verbose=False)['ansible_facts']
-            phy_interfaces = [k for k, v in cfg_facts["PORT"].items() if
-                              "admin_status" in v and v["admin_status"] == "up"]
+            if "PORT" in cfg_facts:
+                phy_interfaces = [k for k, v in cfg_facts["PORT"].items() if
+                                  "admin_status" in v and v["admin_status"] == "up"]
             if "PORTCHANNEL_INTERFACE" in cfg_facts:
                 ip_interfaces = cfg_facts["PORTCHANNEL_INTERFACE"].keys()
             if "VLAN_INTERFACE" in cfg_facts:

tests/platform_tests/mellanox/test_check_sfp_presence.py

@@ -3,7 +3,6 @@
 """
 import logging
 import os
-import json
 import pytest
 
 from tests.common.fixtures.conn_graph_facts import conn_graph_facts
@@ -13,11 +12,12 @@
     pytest.mark.topology('any')
 ]
 
+
 def test_check_sfp_presence(duthosts, rand_one_dut_hostname, conn_graph_facts):
     """This test case is to check SFP presence status with CLI and sysfs.
     """
     duthost = duthosts[rand_one_dut_hostname]
-    ports_config = json.loads(duthost.command("sudo sonic-cfggen -d --var-json PORT")["stdout"])
+    duthost.command("sudo sonic-cfggen -d --var-json PORT")
     check_intf_presence_command = 'show interface transceiver presence {}'
 
     logging.info("Use show interface status information")

tests/platform_tests/mellanox/test_check_sfp_using_ethtool.py

@@ -5,10 +5,11 @@
 https://github.com/Azure/SONiC/blob/master/doc/pmon/sonic_platform_test_plan.md
 """
 import logging
-import os
 import json
 import pytest
 from tests.common.fixtures.conn_graph_facts import conn_graph_facts
+from tests.platform_tests.sfp.util import get_dev_conn
+from tests.common.platform.interface_utils import get_port_map, check_interface_status
 from tests.common.mellanox_data import SPC3_HWSKUS
 from check_hw_mgmt_service import check_hw_management_service
 
@@ -17,18 +18,23 @@
     pytest.mark.topology('any')
 ]
 
-def test_check_sfp_using_ethtool(duthosts, rand_one_dut_hostname, conn_graph_facts, tbinfo):
+
+def test_check_sfp_using_ethtool(duthosts, rand_one_dut_hostname, conn_graph_facts, tbinfo, enum_frontend_asic_index):
     """This test case is to check SFP using the ethtool.
     """
     duthost = duthosts[rand_one_dut_hostname]
-    ports_config = json.loads(duthost.command("sudo sonic-cfggen -d --var-json PORT")["stdout"])
-
+    ports_config = {}
+    ports_config_output = duthost.command("sudo sonic-cfggen -d --var-json PORT")["stdout"]
+    portmap, dev_conn = get_dev_conn(duthost, conn_graph_facts, enum_frontend_asic_index)
+    if ports_config_output:
+        # in the case of zero ports the json output can be none
+        ports_config = json.loads(ports_config_output)
     logging.info("Use the ethtool to check SFP information")
     if duthost.facts["hwsku"] in SPC3_HWSKUS:
         lanes_divider = 8
     else:
         lanes_divider = 4
-    for intf in conn_graph_facts["device_conn"][duthost.hostname]:
+    for intf in dev_conn:
         intf_lanes = ports_config[intf]["lanes"]
         sfp_id = int(intf_lanes.split(",")[0])/lanes_divider + 1
 
@@ -46,9 +52,8 @@ def test_check_sfp_using_ethtool(duthosts, rand_one_dut_hostname, conn_graph_fac
                     "Unexpected line %s in %s" % (line, str(ethtool_sfp_output["stdout_lines"]))
 
     logging.info("Check interface status")
-    mg_facts = duthost.get_extended_minigraph_facts(tbinfo)
-    intf_facts = duthost.interface_facts(up_ports=mg_facts["minigraph_ports"])["ansible_facts"]
-    assert len(intf_facts["ansible_interface_link_down_ports"]) == 0, \
-        "Some interfaces are down: %s" % str(intf_facts["ansible_interface_link_down_ports"])
+    for asic_index in duthost.get_frontend_asic_ids():
+        interface_list = get_port_map(duthost, asic_index)
+        assert check_interface_status(duthost, asic_index, interface_list, []), "Not all interfaces are up"
 
     check_hw_management_service(duthost)

tests/platform_tests/sfp/test_sfputil.py

@@ -14,6 +14,7 @@
 from util import parse_eeprom
 from util import parse_output
 from util import get_dev_conn
+from tests.common.platform.interface_utils import get_port_map, check_interface_status
 from tests.common.utilities import skip_version
 
 cmd_sfp_presence = "sudo sfputil show presence"
@@ -119,10 +120,9 @@ def test_check_sfputil_reset(duthosts, enum_rand_one_per_hwsku_frontend_hostname
             assert parsed_presence[intf] == "Present", "Interface presence is not 'Present'"
 
     logging.info("Check interface status")
-    mg_facts = duthost.get_extended_minigraph_facts(tbinfo)
-    intf_facts = duthost.interface_facts(up_ports=mg_facts["minigraph_ports"])["ansible_facts"]
-    assert len(intf_facts["ansible_interface_link_down_ports"]) == 0, \
-        "Some interfaces are down: {}".format(intf_facts["ansible_interface_link_down_ports"])
+    for asic_index in duthost.get_frontend_asic_ids():
+        interface_list = get_port_map(duthost, asic_index)
+        assert check_interface_status(duthost, asic_index, interface_list, []), "Not all interfaces are up"
 
 
 def test_check_sfputil_low_power_mode(duthosts, enum_rand_one_per_hwsku_frontend_hostname, enum_frontend_asic_index, conn_graph_facts, tbinfo, xcvr_skip_list):

tests/platform_tests/sfp/util.py

@@ -33,15 +33,12 @@ def parse_eeprom(output_lines):
 
 
 def get_dev_conn(duthost, conn_graph_facts, asic_index):
-    dev_conn = conn_graph_facts["device_conn"][duthost.hostname]
-
     # Get the interface pertaining to that asic
     portmap = get_port_map(duthost, asic_index)
     logging.info("Got portmap {}".format(portmap))
 
-    if asic_index is not None:
-        # Check if the interfaces of this AISC is present in conn_graph_facts
-        dev_conn = {k: v for k, v in portmap.items() if k in conn_graph_facts["device_conn"][duthost.hostname]}
-        logging.info("ASIC {} interface_list {}".format(asic_index, dev_conn))
+    # Check if the interfaces of this AISC is present in conn_graph_facts
+    dev_conn = {k: v for k, v in portmap.items() if k in conn_graph_facts["device_conn"][duthost.hostname]}
+    logging.info("ASIC {} interface_list {}".format(asic_index, dev_conn))
 
     return portmap, dev_conn

tests/platform_tests/test_sequential_restart.py

@@ -93,15 +93,11 @@ def test_restart_swss(duthosts, rand_one_dut_hostname, enum_frontend_asic_index,
     """
     duthost = duthosts[rand_one_dut_hostname]
     all_interfaces = conn_graph_facts["device_conn"][duthost.hostname]
-
-    if enum_frontend_asic_index is not None:
-        # Get the interface pertaining to that asic
-        interface_list = get_port_map(duthost, enum_frontend_asic_index)
-
-        # Check if the interfaces of this AISC is present in conn_graph_facts
-        new_intf_dict = {k:v for k, v in interface_list.items() if k in all_interfaces}
-        all_interfaces = new_intf_dict
-        logging.info("ASIC {} interface_list {}".format(enum_frontend_asic_index, all_interfaces))
+    interface_list = get_port_map(duthost, enum_frontend_asic_index)
+    # Check if the interfaces of this AISC is present in conn_graph_facts
+    new_intf_dict = {k:v for k, v in interface_list.items() if k in all_interfaces}
+    all_interfaces = new_intf_dict
+    logging.info("ASIC {} interface_list {}".format(enum_frontend_asic_index, all_interfaces))
 
     restart_service_and_check(localhost, duthost, enum_frontend_asic_index, "swss", all_interfaces, xcvr_skip_list)
 

tests/snmp/conftest.py

@@ -1,4 +1,5 @@
 import pytest
+from tests.common.platform.interface_utils import get_port_map
 from tests.common.utilities import wait_until
 
 @pytest.fixture(scope="module", autouse=True)
@@ -16,3 +17,15 @@ def pytest_addoption(parser):
                     default=False,
                     help="Set percentage difference for snmp test",
                     type=int)
+
+@pytest.fixture(scope='function')
+def skip_if_no_ports(duthosts, enum_rand_one_per_hwsku_frontend_hostname):
+    """
+    Fixture that skips test execution in case dut doesn't have data ports
+    """
+    duthost = duthosts[enum_rand_one_per_hwsku_frontend_hostname]
+    for asic_index in duthost.get_frontend_asic_ids():
+        interface_list = get_port_map(duthost, asic_index)
+        if not interface_list:
+            pytest.skip("This test is not supported as there are no data ports in dut")
+

tests/snmp/test_snmp_default_route.py

@@ -10,7 +10,8 @@
 
 
 @pytest.mark.bsl
-def test_snmp_default_route(duthosts, enum_rand_one_per_hwsku_frontend_hostname, localhost, creds_all_duts, tbinfo):
+def test_snmp_default_route(duthosts, enum_rand_one_per_hwsku_frontend_hostname, localhost, 
+                            creds_all_duts, tbinfo, skip_if_no_ports):
     """compare the snmp facts between observed states and target state"""
 
     duthost = duthosts[enum_rand_one_per_hwsku_frontend_hostname]

tests/snmp/test_snmp_lldp.py

@@ -1,6 +1,5 @@
 import pytest
 import re
-from tests.common.helpers.snmp_helpers import get_snmp_facts
 
 pytestmark = [
     pytest.mark.topology('any'),
@@ -38,12 +37,15 @@ def test_snmp_lldp(duthosts, enum_rand_one_per_hwsku_hostname, localhost, creds_
         pytest.skip("LLDP not supported on supervisor node")
     hostip = duthost.host.options['inventory_manager'].get_host(duthost.hostname).vars['ansible_host']
 
-    snmp_facts = get_snmp_facts(localhost, host=hostip, version="v2c", community=creds_all_duts[duthost]["snmp_rocommunity"], wait=True)['ansible_facts']
-    mg_facts = {}
-    for asic_id in duthost.get_asic_ids():
-        mg_facts_ns   = duthost.asic_instance(asic_id).get_extended_minigraph_facts(tbinfo)['minigraph_neighbors']
-        if mg_facts_ns is not None:
-            mg_facts.update(mg_facts_ns)
+    snmp_facts = localhost.snmp_facts(host=hostip, version="v2c", community=creds_all_duts[duthost]["snmp_rocommunity"])['ansible_facts']
+    for asic in duthost.asics:
+        lldp_nei = []
+        cfg_facts = asic.config_facts(host=duthost.hostname,
+                                      source="persistent", verbose=False)['ansible_facts']
+        if "PORT" in cfg_facts:
+            for port, port_info_dict in cfg_facts["PORT"].items():
+                if re.search('ARISTA', port_info_dict['description']):
+                    lldp_nei.append(port)
 
     print snmp_facts['snmp_lldp']
     for k in ['lldpLocChassisIdSubtype', 'lldpLocChassisId', 'lldpLocSysName', 'lldpLocSysDesc']:
@@ -65,12 +67,6 @@ def test_snmp_lldp(duthosts, enum_rand_one_per_hwsku_hostname, localhost, creds_
         assert snmp_facts['snmp_lldp'][k]
         assert "No Such Object currently exists" not in snmp_facts['snmp_lldp'][k]
 
-    minigraph_lldp_nei = []
-    for k, v in mg_facts.items():
-        if "server" not in v['name'].lower():
-            minigraph_lldp_nei.append(k)
-    print minigraph_lldp_nei
-
     # Check if lldpRemTable is present
     active_intf = []
     for k, v in snmp_facts['snmp_interfaces'].items():
@@ -86,7 +82,7 @@ def test_snmp_lldp(duthosts, enum_rand_one_per_hwsku_hostname, localhost, creds_
             active_intf.append(k)
     print "lldpRemTable: ", active_intf
 
-    assert len(active_intf) >= len(minigraph_lldp_nei) * 0.8
+    assert len(active_intf) >= len(lldp_nei) * 0.8
 
     # skip neighbors that do not send chassis information via lldp
     lldp_facts= {}

tests/snmp/test_snmp_loopback.py

@@ -34,7 +34,7 @@ def get_snmp_output(ip, duthost, nbr, creds_all_duts):
 
 
 @pytest.mark.bsl
-def test_snmp_loopback(duthosts, enum_rand_one_per_hwsku_frontend_hostname, nbrhosts, tbinfo, localhost, creds_all_duts):
+def test_snmp_loopback(duthosts, enum_rand_one_per_hwsku_frontend_hostname, skip_if_no_ports, nbrhosts, tbinfo, localhost, creds_all_duts):
     """
     Test SNMP query to DUT over loopback IP
       - Send SNMP query over loopback IP from one of the BGP Neighbors