Add dualtor support for directed broadcast test

ansible/roles/test/files/ptftests/py3/dir_bcast_test.py

@@ -11,6 +11,7 @@
 #---------------------------------------------------------------------
 import logging
 import random
+import json
 import ptf
 import ptf.packet as scapy
 import ptf.dataplane as dataplane
@@ -53,55 +54,28 @@ def __init__(self):
     def setUp(self):
         self.dataplane = ptf.dataplane_instance
         self.router_mac = self.test_params['router_mac']
-        self.setUpVlan(self.test_params['vlan_info'])
-        if self.test_params['testbed_type'] == 't0':
-            self.src_ports = list(range(1, 25)) + list(range(28, 32))
-        if self.test_params['testbed_type'] == 't0-52':
-            self.src_ports = list(range(0, 52))
-        if self.test_params['testbed_type'] == 't0-56':
-            self.src_ports = list(range(0, 2)) + list(range(4, 6)) + list(range(8, 10)) + list(range(12, 18)) + list(range(20, 22)) + \
-                             list(range(24, 26)) + list(range(28, 30)) + list(range(32, 34)) + list(range(36, 38)) + list(range(40, 46)) + \
-                             list(range(48, 50)) + list(range(52, 54))
-        if self.test_params['testbed_type'] == 't0-56-po2vlan':
-            self.src_ports = [8, 10, 16, 18, 20, 22, 24, 26, 28, 30, 32, 34, 36, 38, 44, 46, 48, 50, 52, 54] + \
-                             [12, 13, 14, 15, 40, 41, 42, 43]
-        if self.test_params['testbed_type'] == 't0-64':
-            self.src_ports = list(range(0, 2)) + list(range(4, 18)) + list(range(20, 33)) + list(range(36, 43)) + list(range(48, 49)) + list(range(52, 59))
-        if self.test_params['testbed_type'] == 't0-116':
-            self.src_ports = list(range(24, 32))
-        if self.test_params['testbed_type'] == 't0-120':
-            self.src_ports = [48, 49, 54, 55, 60, 61, 66, 67]
-
-    #---------------------------------------------------------------------
-
-    def setUpVlan(self, file_path):
-        '''
-        @summary: Populate the VLAN dictionary with IP/Prefix and member port list
-        '''
-        self._vlan_dict = {}
-        with open(file_path, 'r') as f:
-            for line in f.readlines():
-                entry = line.split(' ', 1)
-                prefix = ip_network(str(entry[0]))
-                if prefix.version != 4:
-                    continue
-                self._vlan_dict[prefix] = [int(i) for i in entry[1].split()]
+        ptf_test_port_map = self.test_params['ptf_test_port_map']
+        with open(ptf_test_port_map) as f:
+            self.ptf_test_port_map = json.load(f)
+        self.src_ports = self.ptf_test_port_map['ptf_src_ports']
+        self._vlan_dict = self.ptf_test_port_map['vlan_ip_port_pair']
 
     #---------------------------------------------------------------------
 
     def check_all_dir_bcast(self):
         '''
         @summary: Loop through all the VLANs and send directed broadcast packets
         '''
-        for vlan_pfx in self._vlan_dict:
-            bcast_ip = str(ip_network(vlan_pfx).broadcast_address)
-            dst_port_list = self._vlan_dict[vlan_pfx]
-            self.check_ip_dir_bcast(bcast_ip, dst_port_list)
-            self.check_bootp_dir_bcast(bcast_ip, dst_port_list)
+        for vlan_pfx, dst_ports in self._vlan_dict.items():
+            if ip_network(vlan_pfx).version == 4:
+                bcast_ip = str(ip_network(vlan_pfx).broadcast_address)
+                logging.info("bcast_ip: {}, vlan_pfx: {}, dst_ports: {}".format(bcast_ip, vlan_pfx, dst_ports))
+                self.check_ip_dir_bcast(bcast_ip, dst_ports)
+                self.check_bootp_dir_bcast(bcast_ip, dst_ports)
 
     #---------------------------------------------------------------------
 
-    def check_ip_dir_bcast(self, dst_bcast_ip, dst_port_list):
+    def check_ip_dir_bcast(self, dst_bcast_ip, dst_ports):
         '''
         @summary: Check directed broadcast IP forwarding and receiving on all member ports.
         '''
@@ -124,22 +98,22 @@ def check_ip_dir_bcast(self, dst_bcast_ip, dst_port_list):
         masked_exp_pkt.set_do_not_care_scapy(scapy.IP, "chksum")
         masked_exp_pkt.set_do_not_care_scapy(scapy.IP, "ttl")
 
-        src_port = random.choice([port for port in self.src_ports if port not in dst_port_list])
+        src_port = random.choice([port for port in self.src_ports if port not in dst_ports])
         send_packet(self, src_port, pkt)
         logging.info("Sending packet from port " + str(src_port) + " to " + ip_dst)
 
-        pkt_count = count_matched_packets_all_ports(self, masked_exp_pkt, dst_port_list)
+        pkt_count = count_matched_packets_all_ports(self, masked_exp_pkt, dst_ports)
         '''
         Check if broadcast packet is received on all member ports of vlan
         '''
-        logging.info("Received " + str(pkt_count) + " broadcast packets, expecting " + str(len(dst_port_list)))
-        assert (pkt_count == len(dst_port_list)), "received {} expected {}".format(pkt_count, len(dst_port_list))
+        logging.info("Received " + str(pkt_count) + " broadcast packets, expecting " + str(len(dst_ports)))
+        assert (pkt_count == len(dst_ports)), "received {} expected {}".format(pkt_count, len(dst_ports))
 
         return
 
     #---------------------------------------------------------------------
 
-    def check_bootp_dir_bcast(self, dst_bcast_ip, dst_port_list):
+    def check_bootp_dir_bcast(self, dst_bcast_ip, dst_ports):
         '''
         @summary: Check directed broadcast BOOTP packet forwarding and receiving on all member ports.
         '''
@@ -167,16 +141,16 @@ def check_bootp_dir_bcast(self, dst_bcast_ip, dst_port_list):
         masked_exp_pkt.set_do_not_care_scapy(scapy.IP, "chksum")
         masked_exp_pkt.set_do_not_care_scapy(scapy.IP, "ttl")
 
-        src_port = random.choice([port for port in self.src_ports if port not in dst_port_list])
+        src_port = random.choice([port for port in self.src_ports if port not in dst_ports])
         send_packet(self, src_port, pkt)
         logging.info("Sending BOOTP packet from port " + str(src_port) + " to " + ip_dst)
 
-        pkt_count = count_matched_packets_all_ports(self, masked_exp_pkt, dst_port_list)
+        pkt_count = count_matched_packets_all_ports(self, masked_exp_pkt, dst_ports)
         '''
         Check if broadcast BOOTP packet is received on all member ports of vlan
         '''
-        logging.info("Received " + str(pkt_count) + " broadcast BOOTP packets, expecting " + str(len(dst_port_list)))
-        assert (pkt_count == len(dst_port_list)), "received {} expected {}".format(pkt_count, len(dst_port_list))
+        logging.info("Received " + str(pkt_count) + " broadcast BOOTP packets, expecting " + str(len(dst_ports)))
+        assert (pkt_count == len(dst_ports)), "received {} expected {}".format(pkt_count, len(dst_ports))
 
         return
 

tests/common/plugins/conditional_mark/tests_mark_conditions.yaml

@@ -360,7 +360,12 @@ ipfwd/test_dir_bcast.py:
   skip:
     reason: "Unsupported topology."
     conditions:
-      - "topo_type not in ['t0', 'm0']"
+      - "topo_type not in ['t0', 'm0', 'dualtor']"
+  xfail:
+    reason: "Dualtor do not support now, need to fix in buildimage."
+    conditions:
+      - "topo_type in ['dualtor']"
+      - https://github.com/sonic-net/sonic-buildimage/issues/12167
 
 ipfwd/test_mtu.py:
   skip:

tests/ipfwd/test_dir_bcast.py

@@ -1,47 +1,82 @@
 import pytest
+import json
+import logging
 
 from tests.common.fixtures.ptfhost_utils import copy_ptftests_directory   # lgtm[py/unused-import]
 from tests.ptf_runner import ptf_runner
 from datetime import datetime
+from tests.common.dualtor.mux_simulator_control import toggle_all_simulator_ports_to_rand_selected_tor_m    # lgtm[py/unused-import]
 
 pytestmark = [
     pytest.mark.topology('t0', 'm0')
 ]
 
-def test_dir_bcast(duthosts, rand_one_dut_hostname, ptfhost, tbinfo):
+logger = logging.getLogger(__name__)
+
+PTF_TEST_PORT_MAP = '/root/ptf_test_port_map.json'
+
+def get_ptf_src_ports(mg_facts):
+    # Source ports are upstream ports
+    pc_interfaces = [pc["attachto"] for pc in mg_facts["minigraph_portchannel_interfaces"]]
+    ptf_src_ports = []
+    for pc in pc_interfaces:
+        for member in mg_facts['minigraph_portchannels'][pc]['members']:
+            ptf_src_ports.append(mg_facts['minigraph_ptf_indices'][member])
+    return ptf_src_ports
+
+def get_ptf_dst_ports(duthost, mg_facts, testbed_type):
+    if "dualtor" in testbed_type:
+        # In dualtor, only active port in active tor could be dst port
+        mux_status_out = duthost.show_and_parse("show mux status")
+        mux_active_ports = []
+        for mux_info in mux_status_out:
+            if mux_info['status'] == 'active':
+                mux_active_ports.append(mux_info['port'])
+
+    vlan_ip_port_pair = {}
+    for vlan_intf in mg_facts['minigraph_vlan_interfaces']:
+        vlan_subnet = vlan_intf["subnet"]
+        vlan_name = vlan_intf["attachto"]
+
+        ptf_dst_ports = []
+        for member in mg_facts['minigraph_vlans'][vlan_name]['members']:
+            if "Ethernet" in member:
+                if "dualtor" not in testbed_type:
+                    ptf_dst_ports.append(mg_facts['minigraph_port_indices'][member])
+                elif member in mux_active_ports:
+                    ptf_dst_ports.append(mg_facts['minigraph_port_indices'][member])
+
+        if ptf_dst_ports:
+            vlan_ip_port_pair[vlan_subnet] = ptf_dst_ports
+
+    return vlan_ip_port_pair
+
+def ptf_test_port_map(duthost, ptfhost, mg_facts, testbed_type):
+    ptf_test_port_map = {}
+    ptf_src_ports = get_ptf_src_ports(mg_facts)
+    vlan_ip_port_pair = get_ptf_dst_ports(duthost, mg_facts, testbed_type)
+
+    ptf_test_port_map = {
+        'ptf_src_ports': ptf_src_ports,
+        'vlan_ip_port_pair': vlan_ip_port_pair
+    }
+    ptfhost.copy(content=json.dumps(ptf_test_port_map), dest=PTF_TEST_PORT_MAP)
+
+
+def test_dir_bcast(duthosts, rand_one_dut_hostname, ptfhost, tbinfo, toggle_all_simulator_ports_to_rand_selected_tor_m):
     duthost = duthosts[rand_one_dut_hostname]
     testbed_type = tbinfo['topo']['name']
 
     # Copy VLAN information file to PTF-docker
     mg_facts = duthost.get_extended_minigraph_facts(tbinfo)
 
-    # Filter expected_vlans and minigraph_vlans to support t0-56-po2vlan topology
-    expected_vlans = []
-    minigraph_vlans = {}
-    for vlan in mg_facts['minigraph_vlan_interfaces']:
-        vlan_name = vlan['attachto']
-        if len(mg_facts['minigraph_vlans'][vlan_name]['members']) > 1:
-            expected_vlans.append(vlan)
-            vlan_members = []
-            for vl_m in mg_facts['minigraph_vlans'][vlan_name]['members']:
-                if 'PortChannel' not in vl_m:
-                    vlan_members.append(vl_m)
-            minigraph_vlans[vlan_name] = {'name': vlan_name, 'members': vlan_members}
-
-    extra_vars = {
-        'minigraph_vlan_interfaces': expected_vlans,
-        'minigraph_vlans':           minigraph_vlans,
-        'minigraph_port_indices':    mg_facts['minigraph_ptf_indices'],
-        'minigraph_portchannels':    mg_facts['minigraph_portchannels']
-    }
-    ptfhost.host.options['variable_manager'].extra_vars.update(extra_vars)
-    ptfhost.template(src="../ansible/roles/test/templates/fdb.j2", dest="/root/vlan_info.txt")
+    ptf_test_port_map(duthost, ptfhost, mg_facts, testbed_type)
 
     # Start PTF runner
     params = {
         'testbed_type': testbed_type,
         'router_mac': duthost.facts['router_mac'],
-        'vlan_info': '/root/vlan_info.txt'
+        'ptf_test_port_map': PTF_TEST_PORT_MAP
     }
     log_file = "/tmp/dir_bcast.BcastTest.{}.log".format(datetime.now().strftime("%Y-%m-%d-%H:%M:%S"))
     ptf_runner(