[PATCH v4 2/2] dts: add dual VLAN test suite

[Paul Szczepanek <paul.szczepanek@arm.com>]

From: Jeremy Spewock <jspewock@iol.unh.edu>

Test functionality of VLAN functions such as stripping,
inserting and filtering with two tags present.

Signed-off-by: Jeremy Spewock <jspewock@iol.unh.edu>
Signed-off-by: Paul Szczepanek <paul.szczepanek@arm.com>
---
 dts/tests/TestSuite_dual_vlan.py | 287 +++++++++++++++++++++++++++++++
 1 file changed, 287 insertions(+)
 create mode 100644 dts/tests/TestSuite_dual_vlan.py

diff --git a/dts/tests/TestSuite_dual_vlan.py b/dts/tests/TestSuite_dual_vlan.py
new file mode 100644
index 0000000000..bdbee7e8d1
--- /dev/null
+++ b/dts/tests/TestSuite_dual_vlan.py
@@ -0,0 +1,287 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2024 University of New Hampshire
+
+"""Dual VLAN functionality testing suite.
+
+The main objective of this test suite is to ensure that standard VLAN functions such as stripping
+and filtering both still carry out their expected behavior in the presence of a packet which
+contains two VLAN headers. These functions should carry out said behavior not just in isolation,
+but also when other VLAN functions are configured on the same port. In addition to this, the
+priority attributes of VLAN headers should be unchanged in the case of multiple VLAN headers
+existing on a single packet, and a packet with only a single VLAN header should be able to have one
+additional VLAN inserted into it.
+"""
+
+from enum import Flag, auto
+from typing import ClassVar
+
+from scapy.layers.l2 import Dot1Q, Ether
+from scapy.packet import Packet, Raw
+
+from framework.params.testpmd import SimpleForwardingModes
+from framework.remote_session.testpmd_shell import TestPmdShell
+from framework.test_suite import TestSuite, func_test
+
+
+class TestDualVlan(TestSuite):
+    """DPDK Dual VLAN test suite.
+
+    This suite tests the behavior of VLAN functions and properties in the presence of two VLAN
+    headers. All VLAN functions which are tested in this suite are specified using the inner class
+    :class:`TestCaseOptions` and should have cases for configuring them in
+    :meth:`configure_testpmd` as well as cases for testing their behavior in
+    :meth:`verify_vlan_functions`. Every combination of VLAN functions being enabled should be
+    tested. Additionally, attributes of VLAN headers, such as priority, are tested to ensure they
+    are not modified in the case of two VLAN headers.
+    """
+
+    class TestCaseOptions(Flag):
+        """Flag for specifying which VLAN functions to configure.
+
+        These flags are specific to configuring this testcase and are used to
+        create a matrix of all configuration combinations.
+        """
+
+        #:
+        VLAN_STRIP = auto()
+        #:
+        VLAN_FILTER_INNER = auto()
+        #:
+        VLAN_FILTER_OUTER = auto()
+
+    #: ID to set on inner VLAN tags.
+    inner_vlan_tag: ClassVar[int] = 2
+    #: ID to set on outer VLAN tags.
+    outer_vlan_tag: ClassVar[int] = 1
+    #: ID to use when inserting VLAN tags.
+    vlan_insert_tag: ClassVar[int] = 3
+    #:
+    rx_port: ClassVar[int] = 0
+    #:
+    tx_port: ClassVar[int] = 1
+
+    def is_relevant_packet(self, pkt: Packet) -> bool:
+        """Check if a packet was sent by functions in this suite.
+
+        All functions in this test suite send packets with a payload that is packed with 20 "X"
+        characters. This method, therefore, can determine if the packet was sent by this test suite
+        by just checking to see if this payload exists on the received packet.
+
+        Args:
+            pkt: Packet to check for relevancy.
+
+        Returns:
+            :data:`True` if the packet contains the expected payload, :data:`False` otherwise.
+        """
+        return hasattr(pkt, "load") and "X" * 20 in str(pkt.load)
+
+    def pkt_payload_contains_layers(self, pkt: Packet, *expected_layers: Packet) -> bool:
+        """Verify that the payload of the packet matches `expected_layers`.
+
+        The layers in the payload of `pkt` must match the type and the user-defined fields of the
+        layers in `expected_layers` in order.
+
+        Args:
+            pkt: Packet to check the payload of.
+            *expected_layers: Layers expected to be in the payload of `pkt`.
+
+        Returns:
+            :data:`True` if the payload of `pkt` matches the layers in `expected_layers` in order,
+            :data:`False` otherwise.
+        """
+        current_pkt_layer = pkt.payload
+        ret = True
+        for layer in expected_layers:
+            ret &= isinstance(current_pkt_layer, type(layer))
+            if not ret:
+                break
+            for field, val in layer.fields.items():
+                ret &= (
+                    hasattr(current_pkt_layer, field) and getattr(current_pkt_layer, field) == val
+                )
+            current_pkt_layer = current_pkt_layer.payload
+        return ret
+
+    def verify_vlan_functions(self, send_packet: Packet, options: TestCaseOptions) -> None:
+        """Send packet and verify the received packet has the expected structure.
+
+        Expected structure is defined by `options` according to the following table:
+        +----------------------------------------------+-----------------------+
+        |                  Configure setting           |       Result          |
+        +=======+=======+========+============+========+=======+=======+=======+
+        | Outer | Inner |  VLAN  |   VLAN     | VLAN   | Pass/ | Outer | Inner |
+        | vlan  | vlan  |  strip |   filter   | insert | Drop  | vlan  | vlan  |
+        +-------+-------+--------+------------+--------+-------+-------+-------+
+        |  0x1  |  0x2  |   no   |     no     |   no   | pass  |  0x1  |  0x2  |
+        +-------+-------+--------+------------+--------+-------+-------+-------+
+        |  0x1  |  0x2  |  yes   |     no     |   no   | pass  |  no   |  0x2  |
+        +-------+-------+--------+------------+--------+-------+-------+-------+
+        |  0x1  |  0x2  |   no   |  yes,0x1   |   no   | pass  |  0x1  |  0x2  |
+        +-------+-------+--------+------------+--------+-------+-------+-------+
+        |  0x1  |  0x2  |   no   |  yes,0x2   |   no   | pass  |  0x1  |  0x2  |
+        +-------+-------+--------+------------+--------+-------+-------+-------+
+        |  0x1  |  0x2  |   no   | yes,0x1,0x2|   no   | pass  |  0x1  |  0x2  |
+        +-------+-------+--------+------------+--------+-------+-------+-------+
+        |  0x1  |  0x2  |  yes   |  yes,0x1   |   no   | pass  |  no   |  0x2  |
+        +-------+-------+--------+------------+--------+-------+-------+-------+
+        |  0x1  |  0x2  |  yes   |  yes,0x2   |   no   | pass  |  no   |  0x2  |
+        +-------+-------+--------+------------+--------+-------+-------+-------+
+        |  0x1  |  0x2  |  yes   | yes,0x1,0x2|   no   | pass  |  no   |  0x2  |
+        +-------+-------+--------+------------+--------+-------+-------+-------+
+
+        Args:
+            send_packet: Packet to send for testing.
+            options: Flag which defines the currents configured settings in testpmd.
+        """
+        recv = self.send_packet_and_capture(send_packet)
+        recv = list(filter(self.is_relevant_packet, recv))
+        expected_layers: list[Packet] = []
+
+        if self.TestCaseOptions.VLAN_STRIP not in options:
+            expected_layers.append(Dot1Q(vlan=self.outer_vlan_tag))
+        expected_layers.append(Dot1Q(vlan=self.inner_vlan_tag))
+
+        self.verify(
+            len(recv) > 0,
+            f"Expected to receive packet with the payload {expected_layers} but got nothing.",
+        )
+
+        for pkt in recv:
+            self.verify(
+                self.pkt_payload_contains_layers(pkt, *expected_layers),
+                f"Received packet ({pkt.summary()}) did not match the expected sequence of layers "
+                f"{expected_layers} with options {options}.",
+            )
+
+    def configure_testpmd(self, shell: TestPmdShell, options: TestCaseOptions, add: bool) -> None:
+        """Configure VLAN functions in testpmd based on `options`.
+
+        Args:
+            shell: Testpmd session to configure the settings on.
+            options: Settings to modify in `shell`.
+            add: If :data:`True`, turn the settings in `options` on, otherwise turn them off.
+        """
+        if (
+            self.TestCaseOptions.VLAN_FILTER_INNER in options
+            or self.TestCaseOptions.VLAN_FILTER_OUTER in options
+        ):
+            if add:
+                # If we are adding a filter, filtering has to be enabled first
+                shell.set_vlan_filter(self.rx_port, True)
+
+            if self.TestCaseOptions.VLAN_FILTER_INNER in options:
+                shell.rx_vlan(self.inner_vlan_tag, self.rx_port, add)
+            if self.TestCaseOptions.VLAN_FILTER_OUTER in options:
+                shell.rx_vlan(self.outer_vlan_tag, self.rx_port, add)
+
+            if not add:
+                # If we are removing filters then we need to remove the filters before we can
+                # disable filtering.
+                shell.set_vlan_filter(self.rx_port, False)
+        if self.TestCaseOptions.VLAN_STRIP in options:
+            shell.set_vlan_strip(self.rx_port, add)
+
+    @func_test
+    def insert_second_vlan(self) -> None:
+        """Test that a packet with a single VLAN can have an additional one inserted into it.
+
+        Steps:
+            Set VLAN tag on the tx port.
+            Create a packet with a VLAN tag.
+            Send and receive the packet.
+        Verify:
+            Packets are received.
+            Packet contains two VLAN tags.
+        """
+        with TestPmdShell(self.sut_node, forward_mode=SimpleForwardingModes.mac) as testpmd:
+            testpmd.tx_vlan_set(port=self.tx_port, enable=True, vlan=self.vlan_insert_tag)
+            testpmd.start()
+            recv = self.send_packet_and_capture(
+                Ether() / Dot1Q(vlan=self.outer_vlan_tag) / Raw(b"X" * 20)
+            )
+            self.verify(len(recv) > 0, "Did not receive any packets when testing VLAN insertion.")
+            self.verify(
+                any(
+                    self.is_relevant_packet(p)
+                    and self.pkt_payload_contains_layers(
+                        p, Dot1Q(vlan=self.vlan_insert_tag), Dot1Q(vlan=self.outer_vlan_tag)
+                    )
+                    for p in recv
+                ),
+                "Packet was unable to insert a second VLAN tag.",
+            )
+
+    @func_test
+    def all_vlan_functions(self) -> None:
+        """Test that all combinations of :class:`TestCaseOptions` behave as expected.
+
+        Steps:
+            Send packets with VLAN functions disabled.
+            Send packets with a set of combinations of VLAN functions enabled.
+            Disable VLAN function to allow for a clean environment for the next test.
+        Verify:
+            Packet with two VLANs is unchanged without the VLAN modification functions enabled.
+            VLAN functions work as expected.
+        """
+        send_pkt = (
+            Ether()
+            / Dot1Q(vlan=self.outer_vlan_tag)
+            / Dot1Q(vlan=self.inner_vlan_tag)
+            / Raw(b"X" * 20)
+        )
+        with TestPmdShell(self.sut_node, forward_mode=SimpleForwardingModes.mac) as testpmd:
+            testpmd.start()
+            recv = self.send_packet_and_capture(send_pkt)
+            self.verify(len(recv) > 0, "Unmodified packet was not received.")
+            self.verify(
+                any(
+                    self.is_relevant_packet(p)
+                    and self.pkt_payload_contains_layers(
+                        p, Dot1Q(vlan=self.outer_vlan_tag), Dot1Q(vlan=self.inner_vlan_tag)
+                    )
+                    for p in recv
+                ),
+                "Packet was modified without any VLAN functions applied.",
+            )
+            testpmd.stop()
+            for i in range(2 ** len(self.TestCaseOptions)):
+                options = self.TestCaseOptions(i)
+                self.configure_testpmd(testpmd, options, True)
+                testpmd.start()
+                self.verify_vlan_functions(send_pkt, options)
+                testpmd.stop()
+                self.configure_testpmd(testpmd, options, False)
+
+    @func_test
+    def maintains_priority(self) -> None:
+        """Test that priorities of multiple VLAN tags are preserved by the PMD.
+
+        Steps:
+            Create packets with VLAN tags with priorities.
+            Send and receive packets.
+        Verify:
+            Packets are received.
+            Priorities are unchanged.
+        """
+        pkt = (
+            Ether()
+            / Dot1Q(vlan=self.outer_vlan_tag, prio=1)
+            / Dot1Q(vlan=self.inner_vlan_tag, prio=2)
+            / Raw(b"X" * 20)
+        )
+        with TestPmdShell(self.sut_node, forward_mode=SimpleForwardingModes.mac) as testpmd:
+            testpmd.start()
+            recv = self.send_packet_and_capture(pkt)
+            self.verify(len(recv) > 0, "Did not receive any packets when testing VLAN priority.")
+            self.verify(
+                any(
+                    self.is_relevant_packet(p)
+                    and self.pkt_payload_contains_layers(
+                        p,
+                        Dot1Q(vlan=self.outer_vlan_tag, prio=1),
+                        Dot1Q(vlan=self.inner_vlan_tag, prio=2),
+                    )
+                    for p in recv
+                ),
+                "VLAN headers did not maintain their priorities.",
+            )
-- 
2.39.2