Added ipv6-only support to vlan_ping (#21018)

yanmo96 · web-flow · commit e2e5c334f4e1 · 2025-11-10T08:37:45.000-08:00
diff --git a/tests/common/plugins/conditional_mark/tests_mark_conditions.yaml b/tests/common/plugins/conditional_mark/tests_mark_conditions.yaml
@@ -4815,10 +4815,6 @@ vlan/test_vlan_ping.py:
     conditions:
       - "asic_type in ['broadcom']"
       - "https://github.com/sonic-net/sonic-mgmt/issues/15061 and 'dualtor-aa' in topo_name"
-  xfail:
-    reason: "xfail for IPv6-only topologies, need to add support for IPv6-only"
-    conditions:
-      - "https://github.com/sonic-net/sonic-mgmt/issues/19929 and '-v6-' in topo_name"
 
 #######################################
 #####             voq             #####
diff --git a/tests/vlan/test_vlan_ping.py b/tests/vlan/test_vlan_ping.py
@@ -6,7 +6,6 @@
 import ptf.packet as scapy
 from ptf.mask import Mask
 import six
-from ipaddress import ip_address, IPv4Address
 from tests.common.helpers.assertions import pytest_assert as py_assert
 from tests.common.dualtor.mux_simulator_control import toggle_all_simulator_ports_to_rand_selected_tor_m   # noqa: F401
 from tests.common.dualtor.dual_tor_utils import lower_tor_host   # noqa: F401
@@ -24,18 +23,15 @@ def static_neighbor_entry(duthost, dic, oper, ip_version="both"):
     Performs addition or deletion of static entries of ipv4 and v6 neighbors in DUT based on 'oper' parameter
     """
     for member in list(dic.values()):
-        if ip_version in ["4", "both"]:
+        if ip_version in ["4", "both"] and 'ipv4' in member:
             if oper == "add":
                 logger.debug("adding ipv4 static arp entry for ip %s on DUT" % (member['ipv4']))
                 duthost.shell("sudo arp -s {0} {1}".format(member['ipv4'], member['mac']))
-
             elif oper == "del":
                 logger.debug("deleting ipv4 static arp entry for ip %s on DUT" % (member['ipv4']))
                 duthost.shell("sudo arp -d {0}".format(member['ipv4']))
-            else:
-                logger.debug("unknown operation")
 
-        if ip_version in ["6", "both"]:
+        if ip_version in ["6", "both"] and 'ipv6' in member:
             if oper == "add":
                 logger.debug("adding ipv6 static arp entry for ip %s on DUT" % (member['ipv6']))
                 duthost.shell(
@@ -47,7 +43,6 @@ def static_neighbor_entry(duthost, dic, oper, ip_version="both"):
                                                                                        member['Vlanid']))
             else:
                 logger.debug("unknown operation")
-
         else:
             logger.debug("unknown IP version")
 
@@ -69,23 +64,35 @@ def vlan_ping_setup(duthosts, rand_one_dut_hostname, ptfhost, nbrhosts, tbinfo,
             break
 
     py_assert(vm_name is not None, "Can't get neighbor vm")
+
+    # Determine which interface to use
     if topo_type == "mx":
-        vm_ip_with_prefix = six.ensure_text(vm_info['conf']['interfaces']['Ethernet1']['ipv4'])
-        output = vm_info['host'].command("ip addr show dev eth1")
+        interface_name = 'Ethernet1'
+        dev_name = 'eth1'
     else:
         if 'Port-Channel1' in vm_info['conf']['interfaces']:
-            vm_ip_with_prefix = six.ensure_text(vm_info['conf']['interfaces']['Port-Channel1']['ipv4'])
-            output = vm_info['host'].command("ip addr show dev po1")
+            interface_name = 'Port-Channel1'
+            dev_name = 'po1'
         else:
-            vm_ip_with_prefix = six.ensure_text(vm_info['conf']['interfaces']['Ethernet1']['ipv4'])
-            output = vm_info['host'].command("ip addr show dev eth1")
+            interface_name = 'Ethernet1'
+            dev_name = 'eth1'
         # in case of lower tor host we need to use the next portchannel
         if "dualtor-aa" in tbinfo["topo"]["name"] and rand_one_dut_hostname == lower_tor_host.hostname:
-            vm_ip_with_prefix = six.ensure_text(vm_info['conf']['interfaces']['Port-Channel2']['ipv4'])
-            output = vm_info['host'].command("ip addr show dev po2")
+            interface_name = 'Port-Channel2'
+            dev_name = 'po2'
+
+    # Get IPv4 and IPv6 addresses if available
+    vm_interface = vm_info['conf']['interfaces'][interface_name]
+    if 'ipv4' in vm_interface:
+        vm_host_info["ipv4"] = ipaddress.IPv4Interface(six.ensure_text(vm_interface['ipv4'])).ip
+    if 'ipv6' in vm_interface:
+        vm_host_info["ipv6"] = ipaddress.IPv6Interface(six.ensure_text(vm_interface['ipv6'])).ip
+
+    py_assert('ipv4' in vm_host_info or 'ipv6' in vm_host_info,
+              "VM interface {} has neither IPv4 nor IPv6 address".format(interface_name))
+
+    output = vm_info['host'].command("ip addr show dev {}".format(dev_name))
     vm_host_info["mac"] = output['stdout_lines'][1].split()[1]
-    vm_ip_intf = ipaddress.IPv4Interface(vm_ip_with_prefix).ip
-    vm_host_info["ipv4"] = vm_ip_intf
     duthost = duthosts[rand_one_dut_hostname]
     mg_facts = duthost.get_extended_minigraph_facts(tbinfo)
     if "dualtor-aa" in tbinfo["topo"]["name"]:
@@ -105,21 +112,22 @@ def vlan_ping_setup(duthosts, rand_one_dut_hostname, ptfhost, nbrhosts, tbinfo,
     ptfhost_info = {}
     ip4 = None
     ip6 = None
+
+    # Find the interface that connects to the selected VM
+    vm_addr = vm_host_info.get('ipv4') or vm_host_info.get('ipv6')
     for a_bgp_nbr in mg_facts['minigraph_bgp']:
-        # Get the bgp neighbor connected to the selected VM
-        if a_bgp_nbr['name'] == vm_name and a_bgp_nbr['addr'] == str(vm_host_info['ipv4']):
-            # Find the interface that connects to the selected VM
+        if a_bgp_nbr['name'] == vm_name:
             if topo_type == "mx":
                 for intf in mg_facts['minigraph_interfaces']:
-                    if intf['peer_addr'] == str(vm_host_info['ipv4']):
+                    if intf['peer_addr'].lower() == str(vm_addr).lower():
                         vm_host_info['port_index_list'] = [mg_facts['minigraph_ptf_indices'][intf['attachto']]]
                         break
             else:
                 ifaces_list = []
                 # UL pkt may take any of the tor in case of dualtor-aa
                 if "dualtor-aa" in tbinfo["topo"]["name"]:
                     for intf in mg_facts['minigraph_portchannel_interfaces']:
-                        if type(ip_address(intf['peer_addr'])) is IPv4Address:
+                        if intf['peer_addr'].lower() == str(vm_addr).lower():
                             portchannel = intf['attachto']
                             for iface in mg_facts['minigraph_portchannels'][portchannel]['members']:
                                 ifaces_list.append(mg_facts['minigraph_ptf_indices'][iface])
@@ -128,7 +136,7 @@ def vlan_ping_setup(duthosts, rand_one_dut_hostname, ptfhost, nbrhosts, tbinfo,
                 else:
                     if mg_facts['minigraph_portchannel_interfaces']:
                         for intf in mg_facts['minigraph_portchannel_interfaces']:
-                            if intf['peer_addr'] == str(vm_host_info['ipv4']):
+                            if intf['peer_addr'].lower() == str(vm_addr).lower():
                                 portchannel = intf['attachto']
                                 for iface in mg_facts['minigraph_portchannels'][portchannel]['members']:
                                     ifaces_list.append(mg_facts['minigraph_ptf_indices'][iface])
@@ -139,6 +147,9 @@ def vlan_ping_setup(duthosts, rand_one_dut_hostname, ptfhost, nbrhosts, tbinfo,
                 vm_host_info['port_index_list'] = ifaces_list
             break
 
+    py_assert('port_index_list' in vm_host_info,
+              "Could not find interface connecting to VM {} with address {}".format(vm_name, vm_addr))
+
     # getting the ipv4, ipv6 and vlan id of a vlan in DUT with 2 or more vlan members
     for k, v in list(my_cfg_facts['VLAN'].items()):
         vlanid = v['vlanid']
@@ -154,34 +165,41 @@ def vlan_ping_setup(duthosts, rand_one_dut_hostname, ptfhost, nbrhosts, tbinfo,
         else:
             continue
 
-    # ip prefixes of the vlan
-    vlan_ip_address_v4 = ipaddress.IPv4Interface(ip4).ip
-    vlan_ip_network_v4 = ipaddress.IPv4Interface(ip4).network
+    # Ensure at least one IP version is configured on the VLAN
+    py_assert(ip4 or ip6, "VLAN{} has neither IPv4 nor IPv6 address configured".format(vlanid))
 
     # selecting 2 random vlan members of DUT
     # Remove portchannel in vlan member list
     filter_vlan_member_list = [member for member in list(my_cfg_facts['VLAN_MEMBER']['Vlan' + vlanid].keys())
                                if member in mg_facts['minigraph_ptf_indices']]
     rand_vlan_member_list = random.sample(filter_vlan_member_list, 2)
-    exclude_ip = []
-    exclude_ip.extend(
-        [ipaddress.IPv4Interface(ip4).network.network_address, ipaddress.IPv4Interface(ip4).network.broadcast_address,
-         vlan_ip_address_v4]
-    )
+
+    # Prepare IPv4 address pool if VLAN has IPv4
+    ips_in_vlan = []
+    if ip4:
+        vlan_ip_network_v4 = ipaddress.IPv4Interface(ip4).network
+        vlan_ip_address_v4 = ipaddress.IPv4Interface(ip4).ip
+        exclude_ip = [vlan_ip_network_v4.network_address, vlan_ip_network_v4.broadcast_address, vlan_ip_address_v4]
+        ips_in_vlan = [x for x in vlan_ip_network_v4 if x not in exclude_ip]
+
     # getting port index, mac, ipv4 and ipv6 of ptf ports into a dict
-    ips_in_vlan = [x for x in vlan_ip_network_v4 if x not in exclude_ip]
     for member in rand_vlan_member_list:
-        # Get first and last ip in vlan for two vlan members
-        ip_in_vlan = ips_in_vlan[0 if len(list(ptfhost_info.keys())) == 0 else -1]
         ptfhost_info[member] = {}
         ptfhost_info[member]["Vlanid"] = vlanid
         ptfhost_info[member]["port_index_list"] = [mg_facts['minigraph_ptf_indices'][member]]
         ptfhost_info[member]["mac"] = (ptfhost.shell(
             "ifconfig eth%d | grep -o -E '([[:xdigit:]]{1,2}:){5}[[:xdigit:]]{1,2}'" % ptfhost_info[member][
                 "port_index_list"][0]))['stdout']
-        ptfhost_info[member]["ipv4"] = str(ip_in_vlan)
-        ptfhost_info[member]["ipv6"] = str(
-            ipaddress.IPv6Interface(ip6).network[ptfhost_info[member]["port_index_list"][0]])
+
+        # Assign IPv4 address if VLAN has IPv4
+        if ip4:
+            ip_in_vlan = ips_in_vlan[0 if len(ptfhost_info) == 1 else -1]
+            ptfhost_info[member]["ipv4"] = str(ip_in_vlan)
+
+        # Assign IPv6 address if VLAN has IPv6
+        if ip6:
+            ptfhost_info[member]["ipv6"] = str(
+                ipaddress.IPv6Interface(ip6).network[ptfhost_info[member]["port_index_list"][0]])
 
     yield vm_host_info, ptfhost_info
 
@@ -190,35 +208,50 @@ def vlan_ping_setup(duthosts, rand_one_dut_hostname, ptfhost, nbrhosts, tbinfo,
 
 
 def verify_icmp_packet(dut_mac, src_port, dst_port, ptfadapter, tbinfo,
-                       vlan_mac=None, dtor_ul=False, dtor_dl=False):
-    if dtor_ul is True:
-        # use vlan int mac in case of dualtor UL test pkt
-        pkt = testutils.simple_icmp_packet(eth_src=str(src_port['mac']),
-                                           eth_dst=str(vlan_mac),
-                                           ip_src=str(src_port['ipv4']),
-                                           ip_dst=str(dst_port['ipv4']), ip_ttl=64)
-    else:
-        # use dut mac addr for all other test pkts
-        pkt = testutils.simple_icmp_packet(eth_src=str(src_port['mac']),
-                                           eth_dst=str(dut_mac),
-                                           ip_src=str(src_port['ipv4']),
-                                           ip_dst=str(dst_port['ipv4']), ip_ttl=64)
-    if dtor_dl is True:
-        # expect vlan int mac as src mac in dualtor DL test pkt
-        exptd_pkt = testutils.simple_icmp_packet(eth_src=str(vlan_mac),
-                                                 eth_dst=str(dst_port['mac']),
-                                                 ip_src=str(src_port['ipv4']),
-                                                 ip_dst=str(dst_port['ipv4']), ip_ttl=63)
+                       vlan_mac=None, dtor_ul=False, dtor_dl=False, is_ipv6=False):
+
+    # Determine IP key and packet function based on IP version
+    ip_key = 'ipv6' if is_ipv6 else 'ipv4'
+    packet_func = testutils.simple_icmpv6_packet if is_ipv6 else testutils.simple_icmp_packet
+
+    # Build packet parameters based on IP version
+    if is_ipv6:
+        ip_params = {
+            'ipv6_src': str(src_port[ip_key]),
+            'ipv6_dst': str(dst_port[ip_key]),
+            'ipv6_hlim': 64
+        }
+        exp_ip_params = {
+            'ipv6_src': str(src_port[ip_key]),
+            'ipv6_dst': str(dst_port[ip_key]),
+            'ipv6_hlim': 63
+        }
     else:
-        # expect dut mac as src mac for non dualtor DL test pkt
-        exptd_pkt = testutils.simple_icmp_packet(eth_src=str(dut_mac),
-                                                 eth_dst=str(dst_port['mac']),
-                                                 ip_src=str(src_port['ipv4']),
-                                                 ip_dst=str(dst_port['ipv4']), ip_ttl=63)
-    # skip smac check for dualtor-aa UL test pkt
+        ip_params = {
+            'ip_src': str(src_port[ip_key]),
+            'ip_dst': str(dst_port[ip_key]),
+            'ip_ttl': 64
+        }
+        exp_ip_params = {
+            'ip_src': str(src_port[ip_key]),
+            'ip_dst': str(dst_port[ip_key]),
+            'ip_ttl': 63
+        }
+
+    # Create test packet - use vlan mac for dualtor UL, otherwise use dut mac
+    eth_dst = str(vlan_mac) if dtor_ul else str(dut_mac)
+    pkt = packet_func(eth_src=str(src_port['mac']), eth_dst=eth_dst, **ip_params)
+
+    # Create expected packet - use vlan mac for dualtor DL, otherwise use dut mac
+    exp_eth_src = str(vlan_mac) if dtor_dl else str(dut_mac)
+    exptd_pkt = packet_func(eth_src=exp_eth_src, eth_dst=str(dst_port['mac']), **exp_ip_params)
+
+    # Skip smac check for dualtor-aa UL test pkt
     if "dualtor-aa" in tbinfo["topo"]["name"] and dtor_ul is True:
         exptd_pkt = Mask(exptd_pkt)
         exptd_pkt.set_do_not_care_scapy(scapy.Ether, "src")
+
+    # Send packet and verify with retry logic
     for i in range(5):
         testutils.send_packet(ptfadapter, src_port['port_index_list'][0], pkt)
         try:
@@ -257,14 +290,31 @@ def test_vlan_ping(vlan_ping_setup, duthosts, rand_one_dut_hostname, ptfadapter,
     logger.info("Checking connectivity to ptf ports")
 
     for member in ptfhost_info:
-        if 'dualtor' in tbinfo["topo"]["name"]:
-            verify_icmp_packet(duthost.facts['router_mac'], ptfhost_info[member],
-                               vmhost_info, ptfadapter, tbinfo, vlan_mac, dtor_ul=True)
-            verify_icmp_packet(duthost.facts['router_mac'], vmhost_info, ptfhost_info[member],
-                               ptfadapter, tbinfo, vlan_mac, dtor_dl=True)
-        else:
-            verify_icmp_packet(duthost.facts['router_mac'], ptfhost_info[member], vmhost_info, ptfadapter, tbinfo)
-            verify_icmp_packet(duthost.facts['router_mac'], vmhost_info, ptfhost_info[member], ptfadapter, tbinfo)
+        # Test IPv4 connectivity if VM has IPv4
+        if 'ipv4' in vmhost_info:
+            if 'dualtor' in tbinfo["topo"]["name"]:
+                verify_icmp_packet(duthost.facts['router_mac'], ptfhost_info[member],
+                                   vmhost_info, ptfadapter, tbinfo, vlan_mac, dtor_ul=True, is_ipv6=False)
+                verify_icmp_packet(duthost.facts['router_mac'], vmhost_info, ptfhost_info[member],
+                                   ptfadapter, tbinfo, vlan_mac, dtor_dl=True, is_ipv6=False)
+            else:
+                verify_icmp_packet(duthost.facts['router_mac'], ptfhost_info[member], vmhost_info,
+                                   ptfadapter, tbinfo, is_ipv6=False)
+                verify_icmp_packet(duthost.facts['router_mac'], vmhost_info, ptfhost_info[member],
+                                   ptfadapter, tbinfo, is_ipv6=False)
+
+        # Test IPv6 connectivity if VM has IPv6
+        if 'ipv6' in vmhost_info:
+            if 'dualtor' in tbinfo["topo"]["name"]:
+                verify_icmp_packet(duthost.facts['router_mac'], ptfhost_info[member],
+                                   vmhost_info, ptfadapter, tbinfo, vlan_mac, dtor_ul=True, is_ipv6=True)
+                verify_icmp_packet(duthost.facts['router_mac'], vmhost_info, ptfhost_info[member],
+                                   ptfadapter, tbinfo, vlan_mac, dtor_dl=True, is_ipv6=True)
+            else:
+                verify_icmp_packet(duthost.facts['router_mac'], ptfhost_info[member], vmhost_info,
+                                   ptfadapter, tbinfo, is_ipv6=True)
+                verify_icmp_packet(duthost.facts['router_mac'], vmhost_info, ptfhost_info[member],
+                                   ptfadapter, tbinfo, is_ipv6=True)
 
     # flushing and re-adding ipv6 static arp entry
     static_neighbor_entry(duthost, ptfhost_info, "del", "6")
@@ -281,11 +331,28 @@ def test_vlan_ping(vlan_ping_setup, duthosts, rand_one_dut_hostname, ptfadapter,
     # Checking for connectivity
     logger.info("Check connectivity to both ptfhost")
     for member in ptfhost_info:
-        if 'dualtor' in tbinfo["topo"]["name"]:
-            verify_icmp_packet(duthost.facts['router_mac'], ptfhost_info[member],
-                               vmhost_info, ptfadapter, tbinfo, vlan_mac, dtor_ul=True)
-            verify_icmp_packet(duthost.facts['router_mac'], vmhost_info, ptfhost_info[member],
-                               ptfadapter, tbinfo, vlan_mac, dtor_dl=True)
-        else:
-            verify_icmp_packet(duthost.facts['router_mac'], ptfhost_info[member], vmhost_info, ptfadapter, tbinfo)
-            verify_icmp_packet(duthost.facts['router_mac'], vmhost_info, ptfhost_info[member], ptfadapter, tbinfo)
+        # Test IPv4 connectivity if VM has IPv4
+        if 'ipv4' in vmhost_info:
+            if 'dualtor' in tbinfo["topo"]["name"]:
+                verify_icmp_packet(duthost.facts['router_mac'], ptfhost_info[member],
+                                   vmhost_info, ptfadapter, tbinfo, vlan_mac, dtor_ul=True, is_ipv6=False)
+                verify_icmp_packet(duthost.facts['router_mac'], vmhost_info, ptfhost_info[member],
+                                   ptfadapter, tbinfo, vlan_mac, dtor_dl=True, is_ipv6=False)
+            else:
+                verify_icmp_packet(duthost.facts['router_mac'], ptfhost_info[member], vmhost_info,
+                                   ptfadapter, tbinfo, is_ipv6=False)
+                verify_icmp_packet(duthost.facts['router_mac'], vmhost_info, ptfhost_info[member],
+                                   ptfadapter, tbinfo, is_ipv6=False)
+
+        # Test IPv6 connectivity if VM has IPv6
+        if 'ipv6' in vmhost_info:
+            if 'dualtor' in tbinfo["topo"]["name"]:
+                verify_icmp_packet(duthost.facts['router_mac'], ptfhost_info[member],
+                                   vmhost_info, ptfadapter, tbinfo, vlan_mac, dtor_ul=True, is_ipv6=True)
+                verify_icmp_packet(duthost.facts['router_mac'], vmhost_info, ptfhost_info[member],
+                                   ptfadapter, tbinfo, vlan_mac, dtor_dl=True, is_ipv6=True)
+            else:
+                verify_icmp_packet(duthost.facts['router_mac'], ptfhost_info[member], vmhost_info,
+                                   ptfadapter, tbinfo, is_ipv6=True)
+                verify_icmp_packet(duthost.facts['router_mac'], vmhost_info, ptfhost_info[member],
+                                   ptfadapter, tbinfo, is_ipv6=True)
