[PATCH v3 1/2] dts: add dual_vlan testing suite

[Patrick Robb]

Looks good other than updating the testpmd_shell method names

Reviewed-by: Patrick Robb <probb at iol.unh.edu>

On Wed, Jul 24, 2024 at 1:13 PM <jspewock at iol.unh.edu> wrote:
> +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."""
> +
> +        #:
> +        VLAN_STRIP = auto()
> +        #:
> +        VLAN_FILTER_INNER = auto()
> +        #:
> +        VLAN_FILTER_OUTER = auto()

I think this can now be dropped and we can import VLANOffloadFlag from
testpmd_shell in its place.

> +
> +    #: 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: Dot1Q) -> 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. Expected to already be running
> +                with all ports stopped before being passed into this function.

Instead of having this requirement that all ports be stopped for the
testpmd session, we can just use testpmd_shell.requires_stopped_ports
decorator now.

> +            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.vlan_filter_set(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.vlan_filter_set(self.rx_port, False)
> +        if self.TestCaseOptions.VLAN_STRIP in options:
> +            shell.vlan_strip_set(self.rx_port, add)
> +
> +    def test_insert_second_vlan(self) -> None:
> +        """Test that a packet with a single VLAN can have an additional one inserted into it."""
> +        with TestPmdShell(self.sut_node, forward_mode=SimpleForwardingModes.mac) as testpmd:
> +            testpmd.port_stop_all()

Should be testpmd.stop_all_ports() now per luca's pktegen/testpmd patch.

> +            testpmd.tx_vlan_set(self.tx_port, self.vlan_insert_tag)
> +            testpmd.port_start_all()

start_all_ports()

> +            testpmd.start()
> +            recv = self.send_packet_and_capture(
> +                Ether() / Dot1Q(vlan=self.outer_vlan_tag) / Raw("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.",
> +            )
> +
> +    def test_all_vlan_functions(self) -> None:
> +        """Test that all combinations of :class:`TestCaseOptions` behave as expected.
> +
> +        To test this, the base case is tested first, ensuring that a packet with two VLANs is
> +        unchanged without the VLAN modification functions enabled. Then the same Testpmd shell is
> +        modified to enable all necessary VLAN functions, followed by verification that the
> +        functions work as expected, and finally the functions are disabled to allow for a clean
> +        environment for the next test.
> +        """
> +        send_pakt = (
> +            Ether()
> +            / Dot1Q(vlan=self.outer_vlan_tag)
> +            / Dot1Q(vlan=self.inner_vlan_tag)
> +            / Raw("X" * 20)
> +        )
> +        with TestPmdShell(self.sut_node, forward_mode=SimpleForwardingModes.mac) as testpmd:
> +            testpmd.start()
> +            recv = self.send_packet_and_capture(send_pakt)
> +            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()
> +            testpmd.port_stop_all()

stop_all_ports(). I'll stop mentioning this below. :)