Align test case for new t0 topo (sonic-net#16852)

Align acl test for t0-isolated-d16u16s1

tests/acl/custom_acl_table/test_custom_acl_table.py

@@ -11,6 +11,8 @@
 from tests.common.helpers.assertions import pytest_assert
 from tests.common.plugins.loganalyzer.loganalyzer import LogAnalyzer, LogAnalyzerError
 from tests.common.dualtor.mux_simulator_control import toggle_all_simulator_ports_to_rand_selected_tor  # noqa F401
+from tests.common.utilities import get_upstream_neigh_type
+from tests.common.utilities import get_neighbor_ptf_port_list
 
 logger = logging.getLogger(__name__)
 
@@ -276,11 +278,16 @@ def test_custom_acl(rand_selected_dut, rand_unselected_dut, tbinfo, ptfadapter,
     src_port_indice = mg_facts['minigraph_ptf_indices'][src_port]
     # Put all portchannel members into dst_ports
     dst_port_indices = []
-    for _, v in mg_facts['minigraph_portchannels'].items():
-        for member in v['members']:
-            dst_port_indices.append(mg_facts['minigraph_ptf_indices'][member])
-            if "dualtor-aa" in tbinfo["topo"]["name"]:
-                dst_port_indices.append(mg_facts_unselected_dut['minigraph_ptf_indices'][member])
+    if len(mg_facts['minigraph_portchannels']):
+        for _, v in mg_facts['minigraph_portchannels'].items():
+            for member in v['members']:
+                dst_port_indices.append(mg_facts['minigraph_ptf_indices'][member])
+                if "dualtor-aa" in tbinfo["topo"]["name"]:
+                    dst_port_indices.append(mg_facts_unselected_dut['minigraph_ptf_indices'][member])
+    else:
+        topo = tbinfo["topo"]["type"]
+        upstream_neigh_type = get_upstream_neigh_type(topo)
+        dst_port_indices = get_neighbor_ptf_port_list(rand_selected_dut, upstream_neigh_type, tbinfo)
 
     test_pkts = build_testing_pkts(router_mac)
     for rule, pkt in list(test_pkts.items()):

tests/acl/null_route/test_null_route_helper.py

@@ -13,6 +13,9 @@
 import ptf.testutils as testutils
 from tests.common.helpers.assertions import pytest_require
 from tests.common.plugins.loganalyzer.loganalyzer import LogAnalyzer, LogAnalyzerError
+from tests.common.utilities import get_upstream_neigh_type
+from tests.common.utilities import get_neighbor_ptf_port_list
+from tests.common.utilities import get_neighbor_port_list
 
 logger = logging.getLogger(__name__)
 
@@ -113,22 +116,16 @@ def remove_acl_table(duthost):
     duthost.shell_cmds(cmds=cmds)
 
 
-def get_neighbor_ports(mg_facts, neighbor_name):
-    neighbor_ports = []
-    for key, value in list(mg_facts["minigraph_neighbors"].items()):
-        if neighbor_name in value["name"]:
-            neighbor_ports.append(key)
-    return neighbor_ports
-
-
 @pytest.fixture(scope="module")
 def create_acl_table(rand_selected_dut, tbinfo):
     """
     Create two ACL tables on DUT for testing.
     """
     mg_facts = rand_selected_dut.get_extended_minigraph_facts(tbinfo)
-    if tbinfo["topo"]["type"] == "mx":
-        neighbor_ports = get_neighbor_ports(mg_facts, "M0")
+    topo = tbinfo["topo"]["type"]
+    if topo == "mx" or len(mg_facts["minigraph_portchannels"]) == 0:
+        upstream_neigh_type = get_upstream_neigh_type(topo)
+        neighbor_ports = get_neighbor_port_list(rand_selected_dut, upstream_neigh_type)
         ports = ",".join(neighbor_ports)
     else:
         # Get the list of LAGs
@@ -252,9 +249,10 @@ def test_null_route_helper(rand_selected_dut, tbinfo, ptfadapter,
     rx_port = ptf_port_info['port']
     router_mac = rand_selected_dut.facts["router_mac"]
     mg_facts = rand_selected_dut.get_extended_minigraph_facts(tbinfo)
-    if tbinfo["topo"]["type"] == "mx":
-        neighbor_ports = get_neighbor_ports(mg_facts, "M0")
-        ptf_interfaces = [mg_facts['minigraph_ptf_indices'][port] for port in neighbor_ports]
+    topo = tbinfo["topo"]["type"]
+    if topo == "mx" or len(mg_facts["minigraph_portchannels"]) == 0:
+        upstream_neigh_type = get_upstream_neigh_type(topo)
+        ptf_interfaces = get_neighbor_ptf_port_list(rand_selected_dut, upstream_neigh_type, tbinfo)
     else:
         portchannel_members = []
         for _, v in list(mg_facts["minigraph_portchannels"].items()):

tests/acl/test_acl.py

@@ -396,9 +396,12 @@ def setup(duthosts, ptfhost, rand_selected_dut, rand_unselected_dut, tbinfo, ptf
             # In multi-asic we need config both in host and namespace.
             if namespace:
                 acl_table_ports[''] += port
+    if len(port_channels) and topo in ["t0", "m0_vlan", "m0_l3"] or tbinfo["topo"]["name"] in ("t1-lag", "t1-64-lag",
+                                                                                               "t1-64-lag-clet",
+                                                                                               "t1-56-lag",
+                                                                                               "t1-28-lag",
+                                                                                               "t1-32-lag"):
 
-    if topo in ["t0", "m0_vlan", "m0_l3"] or tbinfo["topo"]["name"] in ("t1-lag", "t1-64-lag", "t1-64-lag-clet",
-                                                                        "t1-56-lag", "t1-28-lag", "t1-32-lag"):
         for k, v in list(port_channels.items()):
             acl_table_ports[v['namespace']].append(k)
             # In multi-asic we need config both in host and namespace.