[Test Gap] A VLAN interface should stay up when all of its member ports are operationally down

.azure-pipelines/pr_test_scripts.yaml

@@ -181,6 +181,7 @@ t0:
   - vlan/test_host_vlan.py
   - vlan/test_vlan.py
   - vlan/test_vlan_ping.py
+  - vlan/test_vlan_ports_down.py
   - vxlan/test_vnet_route_leak.py
   - vxlan/test_vnet_vxlan.py
   - vxlan/test_vxlan_decap.py
@@ -245,6 +246,7 @@ t0-2vlans:
   - dhcp_relay/test_dhcpv6_relay.py
   - vlan/test_host_vlan.py
   - vlan/test_vlan_ping.py
+  - vlan/test_vlan_ports_down.py
 
 t0-sonic:
   - bgp/test_bgp_fact.py

docs/api_wiki/README.md

@@ -171,6 +171,8 @@ def test_fun(duthosts, rand_one_dut_hostname, ptfhost):
 
 - [get_interfaces_status](sonichost_methods/get_interfaces_status.md) - Get interfaces status on the DUT and parse the result into a dict.
 
+- [show_ipv6_interfaces](sonichost_methods/show_ipv6_interfaces.md) - Retrieve information about IPv6 interfaces and parse the result into a dict.
+
 - [get_intf_link_local_ipv6_addr](sonichost_methods/get_intf_link_local_ipv6_addr.md) - Get the link local ipv6 address of the interface
 
 - [get_ip_route_info](sonichost_methods/get_ip_route_info.md) - Returns route information for a destionation. The destination could an ip address or ip prefix.

docs/api_wiki/sonichost_methods/show_ipv6_interfaces.md

@@ -0,0 +1,55 @@
+# show_ipv6_interfaces
+
+- [Overview](#overview)
+- [Examples](#examples)
+- [Arguments](#arguments)
+- [Expected Output](#expected-output)
+
+## Overview
+Retrieve information about IPv6 interfaces and parse the result into a dict.
+
+## Examples
+```python
+def test_fun(duthosts, rand_one_dut_hostname):
+    duthost = duthosts[rand_one_dut_hostname]
+
+    ipv6_ifs = duthost.show_ipv6_interfaces()
+```
+
+## Arguments
+This function takes no arguments.
+
+## Expected Output
+Returns a dictionary containing information about the DUT's IPv6 interfaces.
+Note: The result does NOT contain link-local IPv6 addresses.
+
+Example output:
+
+```json
+{
+    "Ethernet16": {
+        "master": "Bridge",
+        "ipv6 address/mask": "fe80::2048:23ff:fe27:33d8%Ethernet16/64",
+        "admin": "up",
+        "oper": "up",
+        "bgp neighbor": "N/A",
+        "neighbor ip": "N/A"
+    },
+    "PortChannel101": {
+        "master": "",
+        "ipv6 address/mask": "fc00::71/126",
+        "admin": "up",
+        "oper": "up",
+        "bgp neighbor": "ARISTA01T1",
+        "neighbor ip": "fc00::72"
+    },
+    "eth5": {
+        "master": "",
+        "ipv6 address/mask": "fe80::5054:ff:fee6:bea6%eth5/64",
+        "admin": "up",
+        "oper": "up",
+        "bgp neighbor": "N/A",
+        "neighbor ip": "N/A"
+    }
+}
+```

tests/common/devices/sonic.py

@@ -1896,6 +1896,54 @@ def get_interfaces_status(self):
         '''
         return {x.get('interface'): x for x in self.show_and_parse('show interfaces status')}
 
+    def show_ipv6_interfaces(self):
+        '''
+        Retrieves information about IPv6 interfaces by running "show ipv6 interfaces" on the DUT
+        and then parses the result into a dict.
+
+        Example output:
+            {
+                "Ethernet16": {
+                    'master': 'Bridge',
+                    'ipv6 address/mask': 'fe80::2048:23ff:fe27:33d8%Ethernet16/64',
+                    'admin': 'up',
+                    'oper': 'up',
+                    'bgp neighbor': 'N/A',
+                    'neighbor ip': 'N/A'
+                },
+                "PortChannel101": {
+                    'master': '',
+                    'ipv6 address/mask': 'fc00::71/126',
+                    'admin': 'up',
+                    'oper': 'up',
+                    'bgp neighbor': 'ARISTA01T1',
+                    'neighbor ip': 'fc00::72'
+                },
+                "eth5": {
+                    'master': '',
+                    'ipv6 address/mask': 'fe80::5054:ff:fee6:bea6%eth5/64',
+                    'admin': 'up',
+                    'oper': 'up',
+                    'bgp neighbor': 'N/A',
+                    'neighbor ip': 'N/A'
+                }
+            }
+        '''
+        result = {iface_info["interface"]: iface_info for iface_info in self.show_and_parse("show ipv6 interfaces")}
+        # Some interfaces have two IPv6 addresses: One public and one link-local address.
+        # Since show_and_parse parses each line separately, it cannot handle this case properly.
+        # So for interfaces that have two IPv6 addresses, we ignore the second line (which corresponds
+        # to the link-local address).
+        if "" in result:
+            del result[""]
+        for iface in result.keys():
+            del result[iface]["interface"]  # redundant, because it is equal to iface
+            admin_oper = result[iface]["admin/oper"].split('/')
+            del result[iface]["admin/oper"]
+            result[iface]["admin"] = admin_oper[0]
+            result[iface]["oper"] = admin_oper[1]
+        return result
+
     def get_crm_facts(self):
         """Run various 'crm show' commands and parse their output to gather CRM facts
 

tests/vlan/test_autostate_disabled.py

@@ -37,8 +37,8 @@ class TestAutostateDisabled:
     up/up status when at least one Layer 2 port becomes active in that VLAN.
 
     In SONiC, all vlans are bound to a single bridge interface, so the vlan interface will go down only if the bridge
-    is down. Since bridge goes down when all the associated interfaces are down, if all the vlan members across all
-    the vlans go down, the bridge will go down and the vlan interface will go down.
+    is down. If all the vlan members across all the vlans go down, the bridge should still remain up so as to prevent
+    the vlan interface from going down.
 
     For more information about autostate, see:
       * https://www.cisco.com/c/en/us/support/docs/switches/catalyst-6500-series-switches/41141-188.html
@@ -48,7 +48,6 @@ def test_autostate_disabled(self, duthosts, enum_frontend_dut_hostname):
         """
         Verify vlan interface autostate is disabled on SONiC.
         """
-        pytest.skip("Temporarily skipped to let the sonic-swss submodule be updated.")
 
         duthost = duthosts[enum_frontend_dut_hostname]
         dut_hostname = duthost.hostname
@@ -84,16 +83,10 @@ def test_autostate_disabled(self, duthosts, enum_frontend_dut_hostname):
 
             # Check whether the oper_state of vlan interface is changed as expected.
             ip_ifs = duthost.show_ip_interface()['ansible_facts']['ip_interfaces']
-            if len(vlan_available) > 1:
-                # If more than one vlan comply with the above test requirements, then there are members in other vlans
-                # that are still up. Therefore, the bridge is still up, and vlan interface should be up.
-                pytest_assert(ip_ifs.get(vlan, {}).get('oper_state') == "up",
-                              'vlan interface of {vlan} is not up as expected'.format(vlan=vlan))
-            else:
-                # If only one vlan comply with the above test requirements, then all the vlan members across all the
-                # vlans are down. Therefore, the bridge is down, and vlan interface should be down.
-                pytest_assert(ip_ifs.get(vlan, {}).get('oper_state') == "down",
-                              'vlan interface of {vlan} is not down as expected'.format(vlan=vlan))
+            # Even if all member ports of all vlans are down, the dummy interface is still expected to be
+            # up. Therefore, the bridge should still be up, which means that vlan interfaces should be up.
+            pytest_assert(ip_ifs.get(vlan, {}).get('oper_state') == "up",
+                          'vlan interface of {vlan} is not up as expected'.format(vlan=vlan))
         finally:
             # Restore all interfaces to their original admin_state.
             self.restore_interface_admin_state(duthost, ifs_status)

tests/vlan/test_vlan_ports_down.py

@@ -0,0 +1,128 @@
+import pytest
+import logging
+import ptf.testutils as testutils
+import ptf.mask as mask
+import time
+
+from netaddr import IPNetwork, NOHOST
+from tests.common.helpers.assertions import pytest_assert
+from scapy.all import IP, Ether
+
+logger = logging.getLogger(__name__)
+
+pytestmark = [
+    pytest.mark.topology('t0')
+]
+
+
+@pytest.fixture(scope='module')
+def vlan_ports_setup(duthosts, rand_one_dut_hostname):
+    """
+    Setup:      Brings down all member ports of a VLAN.
+    Teardown:   Restores the admin state of all member ports of the VLAN selected in the Setup phase.
+    """
+    duthost = duthosts[rand_one_dut_hostname]
+    vlan_brief = duthost.get_vlan_brief()
+    if not vlan_brief:
+        pytest.skip("The testbed does not have any VLANs.")
+    # Selecting the first VLAN in 'vlan_brief'
+    vlan_name = next(iter(vlan_brief))
+    vlan_members = vlan_brief[vlan_name]["members"]
+    ifs_status = duthost.get_interfaces_status()
+    vlan_up_members = [port for port in vlan_members if ifs_status[port]["admin"] == "up"]
+    logger.info(f"Bringing down all member ports of {vlan_name}...")
+    for vlan_up_port in vlan_up_members:
+        duthost.shell(f"sudo config interface shutdown {vlan_up_port}")
+    time.sleep(5)  # Sleep for 5 seconds to ensure T1 switches update their routing table
+    yield vlan_name
+    logger.info(f"Restoring the previous admin state of all member ports of {vlan_name}...")
+    for vlan_port in vlan_up_members:
+        duthost.shell(f"sudo config interface startup {vlan_port}")
+
+
+def test_vlan_ports_down(vlan_ports_setup, duthosts, rand_one_dut_hostname, nbrhosts, tbinfo, ptfadapter):
+    """
+    Asserts the following conditions when all member ports of a VLAN interface are down:
+        1. The VLAN interface's oper status remains Up.
+        2. The VLAN's subnet IP is advertised to the T1 neighbors.
+        3. The IP decapsulation feature works for packets that are sent to the VLAN interfaces's IP address.
+    """
+    duthost = duthosts[rand_one_dut_hostname]
+    vlan_name = vlan_ports_setup
+    ip_interfaces = duthost.show_ip_interface()["ansible_facts"]["ip_interfaces"]
+    vlan_info = ip_interfaces[vlan_name]
+    logger.info(f"Checking if {vlan_name} is oper UP...")
+    # check if the VLAN interface is operationally Up (IPv4)
+    pytest_assert(vlan_info["oper_state"] == "up", f"{vlan_name} is operationally down.")
+
+    ipv6_interfaces = duthost.show_ipv6_interfaces()
+    vlan_info_ipv6 = ipv6_interfaces[vlan_name]
+    # check if the VLAN interface is operationally Up (IPv6)
+    pytest_assert(vlan_info_ipv6["oper"] == "up", f"{vlan_name} is operationally down.")
+
+    logger.info("Checking BGP routes on T1 neighbors...")
+    vlan_subnet = str(IPNetwork(f"{vlan_info['ipv4']}/{vlan_info['prefix_len']}", flags=NOHOST))
+    vlan_subnet_ipv6 = str(IPNetwork(vlan_info_ipv6["ipv6 address/mask"], flags=NOHOST))
+    nbrcount = 0
+    for nbrname, nbrhost in nbrhosts.items():
+        nbrhost = nbrhost["host"]
+        # check IPv4 routes on nbrhost
+        logger.info(f"Checking IPv4 routes on {nbrname}...")
+        try:
+            vlan_route = nbrhost.get_route(vlan_subnet)["vrfs"]["default"]
+        except Exception:
+            # nbrhost might be unreachable. Skip it.
+            logger.info(f"{nbrname} might be unreachable.")
+            continue
+        pytest_assert(vlan_route["bgpRouteEntries"],
+                      f"{vlan_name}'s IPv4 subnet is not advertised to the T1 neighbor {nbrname}.")
+        # check IPv6 routes on nbrhost
+        logger.info(f"Checking IPv6 routes on {nbrname}...")
+        try:
+            vlan_route_ipv6 = nbrhost.get_route(vlan_subnet_ipv6)["vrfs"]["default"]
+        except Exception:
+            # nbrhost might be unreachable. Skip it.
+            logger.info(f"{nbrname} might be unreachable.")
+            continue
+        pytest_assert(vlan_route_ipv6["bgpRouteEntries"],
+                      f"{vlan_name}'s IPv6 subnet is not advertised to the T1 neighbor {nbrname}.")
+        nbrcount += 1
+    if nbrcount == 0:
+        pytest.skip("Could not get routing info from any T1 neighbors.")
+    if duthost.facts["asic_type"].lower() == "vs":
+        logger.info("Skipping IP-in-IP decapsulation test for the 'vs' ASIC type.")
+        return
+    logger.info("Starting the IP-in-IP decapsulation test...")
+    mg_facts = duthost.get_extended_minigraph_facts(tbinfo)
+    # Use the first Ethernet port associated with the first portchannel to send test packets to the DUT
+    portchannel_info = next(iter(mg_facts["minigraph_portchannels"].values()))
+    ptf_src_port = portchannel_info["members"][0]
+    ptf_src_port_index = mg_facts["minigraph_ptf_indices"][ptf_src_port]
+    ptf_dst_port_indices = list(mg_facts["minigraph_ptf_indices"].values())
+    # Test IPv4 in IPv4 decapsulation.
+    # Outer IP packet:
+    #   src: 1.1.1.1
+    #   dst: VLAN interface's IPv4 address
+    # Inner IP packet:
+    #   src: 2.2.2.2
+    #   dst: 3.3.3.3
+    # Expectation: The T0 switch (DUT) decapsulates the outer IP packet and sends
+    # the inner IP packet to the default gateway (one of the connected T1 switches).
+    inner_pkt = testutils.simple_udp_packet(ip_src="2.2.2.2",
+                                            ip_dst="3.3.3.3")
+    outer_pkt = testutils.simple_ipv4ip_packet(eth_src=ptfadapter.dataplane.get_mac(0, ptf_src_port_index),
+                                               eth_dst=duthost.facts["router_mac"],
+                                               ip_src="1.1.1.1",
+                                               ip_dst=vlan_info["ipv4"],
+                                               inner_frame=inner_pkt["IP"])
+    exp_pkt = inner_pkt.copy()
+    exp_pkt = mask.Mask(exp_pkt)
+    exp_pkt.set_do_not_care_packet(Ether, "src")
+    exp_pkt.set_do_not_care_packet(Ether, "dst")
+    exp_pkt.set_do_not_care_packet(IP, "ttl")
+    exp_pkt.set_do_not_care_packet(IP, "chksum")
+    exp_pkt.set_do_not_care_packet(IP, "tos")
+    logger.info("Sending the IP-in-IP packet...")
+    testutils.send(ptfadapter, ptf_src_port_index, outer_pkt)
+    logger.info("IP-in-IP packet sent.")
+    testutils.verify_packet_any_port(ptfadapter, exp_pkt, ports=ptf_dst_port_indices)