Re: [PATCH v1 2/2] dts: add qinq test suite

[Luca Vizzarro <Luca.Vizzarro@arm.com>]

Hi Dean,

thank you so much for your patch. Looks mostly correct, just lots of 
polishing suggestions for a more readable code.

On 17/07/2025 21:57, Dean Marx wrote:
> +"""QinQ (802.1ad) Test Suite.
> +
> +This test suite verifies the correctness and capability of DPDK Poll Mode Drivers (PMDs)
> +in handling QinQ-tagged Ethernet frames, which contain a pair of stacked VLAN headers
> +(outer S-VLAN and inner C-VLAN). These tests ensure that both software and hardware offloads
> +related to QinQ behave as expected across different NIC vendors and PMD implementations.
> +
extra new line not needed.
> +"""

> +class TestQinq(TestSuite):
> <snip>
> +    def send_packet_and_verify(
> +        self, packet: Packet, testpmd: TestPmdShell, should_receive: bool
> +    ) -> None:
> <snip>
> +        if should_receive:
> +            self.verify(received != [], "Packet was dropped when it should have been received.")
> +        else:
> +            self.verify(received == [], "Packet was received when it should have been dropped.")
Quite an unusual way to check if a list is (not) empty. I'd just exploit 
the boolean check on lists in this case. Also send_packet_and_capture 
returns a list of packets, received is a misleading name as it indicates 
it returns a boolean instead. For clarity, you could do:

   packets = send_packet_and_capture(..)
   not_received = not packets

> +    @requires(NicCapability.RX_OFFLOAD_VLAN_EXTEND)
> +    @func_test
> +    def test_qinq_filter(self) -> None:
> <snip>
> +        with TestPmdShell() as testpmd:
> +            testpmd.stop_all_ports()
> +            testpmd.set_vlan_filter(0, True)
> +            testpmd.set_vlan_extend(0, True)
> +            testpmd.start_all_ports()
> +            testpmd.rx_vlan(100, 0, True)
> +            self.send_packet_and_verify(packets[0], testpmd, should_receive=True)
> +            self.send_packet_and_verify(packets[1], testpmd, should_receive=False)

Help me understand what's happening here. set_vlan_filter and 
set_vlan_extend are decorated with @requires_stopped_ports. That means 
that stop_all_ports will be called automatically, so that removes the 
need for it.

What confuses me is that rx_vlan is also decorated with 
@requires_stopped_ports, which makes the preceding start_all_ports a 
redundant action.

The underlying testpmd.start() in send_packet_and_verify is decorated 
with @requires_started_ports, so it'd also make calling start_all_ports 
redundant even if it were placed in the correct order.

My current understanding is that both stop_all_ports and start_all_ports 
are redundant, and the latter is a potential bug (due to its bad 
placement) and the test case only works thanks to the decorators. If 
this is the case then both stop_all_ports and start_all_ports should go.

> +
> +    @func_test
> +    def test_qinq_forwarding(self) -> None:
> +        """QinQ Rx filter test case.
> +
> +        Steps:
> +            Launch testpmd with mac forwarding mode.
> +            Disable VLAN filter mode on port 0.
> +            Send test packet and capture verbose output.
> +
> +        Verify:
> +            Check that the received packet has two separate VLAN layers in proper QinQ fashion.
> +            Check that the received packet outer and inner VLAN layer has the appropriate ID.
> +        """
> +        test_packet = (
> +            Ether(dst="ff:ff:ff:ff:ff:ff")
> +            / Dot1AD(vlan=100)
> +            / Dot1Q(vlan=200)
> +            / IP(dst="1.2.3.4")
> +            / UDP(dport=1234, sport=4321)
> +        )
> +        with TestPmdShell() as testpmd:
> +            testpmd.stop_all_ports()
not needed
> +            testpmd.set_vlan_filter(0, False)
> +            testpmd.start_all_ports()
not needed
> +            testpmd.start()
> +            received_packet = self.send_packet_and_capture(test_packet)
received_packets
> +
> +            self.verify(
> +                received_packet != [], "Packet was dropped when it should have been received."
> +            )
not received_packets
> +
> +            for packet in received_packet:
> +                vlan_tags = packet.getlayer(Dot1AD), packet.getlayer(Dot1Q)
> +                self.verify(len(vlan_tags) == 2, f"Expected 2 VLAN tags, found {len(vlan_tags)}")
I am expecting this will always be True. You are creating a fixed-size 
tuple[Dot1AD | None, Dot1Q | None]. So even if both are None vlan_tags 
will still be (None, None) which is 2 None items. Since you don't 
actually use vlan_tags, it may easier to verify the presence of the 
layer of directly, but for Dot1Q it already happens herebelow. So I'd 
just check Dot1AD:

   self.verify(packet.haslayer(Dot1AD), ...)

> +
> +                if packet.haslayer(Dot1Q):
> +                    outer_vlan_id = packet[Dot1Q].vlan
It may be more readable to take advantage of getlayer in the if:

   if outer_vlan := packet.getlayer(Dot1Q):
       outer_vlan_id = outer_vlan.vlan
> +                    self.verify(
> +                        outer_vlan_id == 100,
> +                        f"Outer VLAN ID was {outer_vlan_id} when it should have been 100.",
> +                    )
> +                else:
> +                    self.verify(False, "VLAN layer not found in received packet.")
> +
> +                if packet[Dot1Q].haslayer(Dot1Q):
> +                    inner_vlan_id = packet[Dot1Q].payload[Dot1Q].vlan
the above approach helps to avoid this chain:

     if inner_vlan := outer_vlan.getlayer(Dot1Q):
         inner_vlan_id = inner_vlan.vlan
> +                    self.verify(
> +                        inner_vlan_id == 200,
> +                        f"Inner VLAN ID was {inner_vlan_id} when it should have been 200",
> +                    )
> +
> +    @func_test
> +    def test_mismatched_tpid(self) -> None:
> +        """Test behavior when outer VLAN tag has a non-standard TPID (not 0x8100 or 0x88a8).
> +
> +        Steps:
> +            Launch testpmd in mac forward mode.
> +            Set verbose level to 1.
> +            Disable VLAN filtering on port 0.
> +            Send and capture test packet.
> +
> +        Verify:
> +            Only 1 VLAN tag is in received packet.
> +            Inner VLAN ID matches the original packet.
> +        """
> +        with TestPmdShell() as testpmd:
> +            testpmd.set_verbose(level=1)
> +            testpmd.stop_all_ports()
not needed
> +            testpmd.set_vlan_filter(0, False)
> +            testpmd.start_all_ports()
not needed
> +            testpmd.start()
> +
> +            mismatched_packet = (
> +                Ether(dst="ff:ff:ff:ff:ff:ff")
> +                / Dot1Q(vlan=100, type=0x1234)
> +                / Dot1Q(vlan=200)
> +                / IP(dst="1.2.3.4")
> +                / UDP(dport=1234, sport=4321)
> +            )
> +
> +            received_packet = self.send_packet_and_capture(mismatched_packet)
reiceved_packets
> +
> +            self.verify(
> +                received_packet != [], "Packet was dropped when it should have been received."
> +            )
not received_packets
> +
> +            for packet in received_packet:
> +                vlan_tags = [layer for layer in packet.layers() if layer == Dot1Q]
> +                self.verify(
> +                    len(vlan_tags) == 1,
> +                    f"Expected 1 VLAN tag due to mismatched TPID, found {len(vlan_tags)}",
> +                )
> +
> +                vlan_id = packet[Dot1Q].vlan
you already collected the layer in vlan_tags, and the above verify 
guarantees we have one tag. This should just be:
   vlan_tags[0].vlan

> +                self.verify(vlan_id == 200, f"Expected inner VLAN ID 200, got {vlan_id}")
> +
> +    def strip_verify(
> +        self, packet: Packet, expected_num_tags: int, context: str, check_id: bool = False
since expected_num_tags can be only 0 or 1, I'd make this a bool, since 
this function doesn't handle any other number (and could become a problem).
> +    ) -> None:
> +        """Helper method for verifying packet stripping functionality."""
> +        if expected_num_tags == 0:
> +            has_vlan = bool(packet.haslayer(Dot1Q))
> +            if not has_vlan:
> +                self.verify(True, "Packet contained VLAN layer")
this verify will never trigger, meant to be False? Although I think this 
whole block is redundant.
> +            else:
> +                vlan_layer = packet.getlayer(Dot1Q)
> +                self.verify(
> +                    vlan_layer is not None
all of the above could be greatly simplified with the approach suggested 
so far:

     if vlan_layer := packet.getlayer(Dot1Q):
         self.verify(vlan_layer.type != ...)
         ...

No `else` needed.
> +                    and vlan_layer.type != 0x8100
> +                    and vlan_layer.type != 0x88A8,
> +                    f"""VLAN tags found in packet when should have been stripped: {packet.summary()}
> +                    sent packet: {context}""",
> +                )
> +        if expected_num_tags == 1:
> +
> +            def count_vlan_tags(packet: Packet) -> int:
> +                """Method for counting the number of VLAN layers in a packet."""
> +                count = 0
> +                layer = packet.getlayer(Dot1Q)
> +                while layer:
> +                    if layer.type == 0x8100:
> +                        count += 1
> +                    layer = layer.payload.getlayer(Dot1Q)
> +                return count
Unless I am misunderstanding this could be greater simplified with:

     tag_count = [
         type(layer) is Dot1Q and layer.type == 0x8100
         for layer in packet.layers()
     ].count(True)

> +
> +            tag_count = count_vlan_tags(packet)
You are calling this only once as far as I can see. So why make it a 
local function (not a method)?
> +            self.verify(
> +                tag_count == 1,
> +                f"""Expected one 0x8100 VLAN tag but found {tag_count}: {packet.summary()}
> +                sent packet: {context}""",
please no triple quotes
> +            )
> +            first_dot1q = packet.getlayer(Dot1Q)
since you actually want the first layer, I'd then change the list 
comprehension above so that you can avoid checking for None again:

    dot1q_layers = [
        layer
        for layer in packet.layers()
        if type(layer) is Dot1Q and layer.type == 0x8100
    ]

    self.verify(len(dot1q_layers) == 1, ...)

    ...

    first_dot1q = dot1q_layers[0]
> +            self.verify(
> +                first_dot1q is not None and first_dot1q.type == 0x8100,
> +                f"""VLAN tag 0x8100 not found in packet: {packet.summary()}
> +                sent packet: {context}""",
no triple quotes
> +            )
> +            if check_id:
> +                self.verify(
> +                    packet[Dot1Q].vlan == 200,
is this just meant to be first_dot1q?
> +                    f"""VLAN ID 200 not found in received packet: {packet.summary()}
> +                    sent packet: {context}""",
no triple quotes
> +                )
> +
> +    @requires(NicCapability.RX_OFFLOAD_VLAN_STRIP)
> +    @func_test
> +    def test_vlan_strip(self) -> None:
> +        """Test combinations of VLAN/QinQ strip settings with various QinQ packets.
> +
> +        Steps:
> +            Launch testpmd with VLAN strip enabled.
> +            Send four VLAN/QinQ related test packets.
> +
> +        Verify:
> +            Check received packets have the expected VLAN/QinQ layers/tags.
> +        """
> +        test_packets = [
> +            Ether() / Dot1Q(type=0x8100) / IP() / UDP(dport=1234, sport=4321),
> +            Ether()
> +            / Dot1Q(vlan=100, type=0x8100)
> +            / Dot1Q(vlan=200, type=0x8100)
> +            / IP()
> +            / UDP(dport=1234, sport=4321),
> +            Ether() / Dot1Q(type=0x88A8) / IP() / UDP(dport=1234, sport=4321),
> +            Ether() / Dot1Q(type=0x88A8) / Dot1Q(type=0x8100) / IP() / UDP(dport=1234, sport=4321),
> +        ]
> +        with TestPmdShell() as testpmd:
> +            testpmd.stop_all_ports()
not needed
> +            testpmd.set_vlan_strip(0, True)
> +            testpmd.start_all_ports()
not needed
> +            testpmd.start()
> +
> +            rec1 = self.send_packet_and_capture(test_packets[0])
> +            rec2 = self.send_packet_and_capture(test_packets[1])
> +            rec3 = self.send_packet_and_capture(test_packets[2])
> +            rec4 = self.send_packet_and_capture(test_packets[3])
> +
> +            testpmd.stop()
> +
> +            try:
> +                context = "Single VLAN"
> +                self.strip_verify(rec1[0], 0, context)
> +                context = "Stacked VLAN"
> +                self.strip_verify(rec2[0], 1, context, check_id=True)
> +                context = "Single S-VLAN"
> +                self.strip_verify(rec3[0], 0, context)
> +                context = "QinQ"
> +                self.strip_verify(rec4[0], 1, context)
> +            except IndexError:
> +                self.verify(
> +                    False, f"{context} packet was dropped when it should have been received."
> +                )
what guarantees us that the packet we sent is the first one to be received?
> +
> +    @requires(NicCapability.RX_OFFLOAD_QINQ_STRIP)
> +    @func_test
> +    def test_qinq_strip(self) -> None:
> +        """Test combinations of VLAN/QinQ strip settings with various QinQ packets.
> +
> +        Steps:
> +            Launch testpmd with QinQ and VLAN strip enabled.
> +            Send four VLAN/QinQ related test packets.
> +
> +        Verify:
> +            Check received packets have the expected VLAN/QinQ layers/tags.
> +        """
> +        test_packets = [
> +            Ether() / Dot1Q(type=0x8100) / IP() / UDP(dport=1234, sport=4321),
> +            Ether()
> +            / Dot1Q(vlan=100, type=0x8100)
> +            / Dot1Q(vlan=200, type=0x8100)
> +            / IP()
> +            / UDP(dport=1234, sport=4321),
> +            Ether() / Dot1Q(type=0x88A8) / IP() / UDP(dport=1234, sport=4321),
> +            Ether() / Dot1Q(type=0x88A8) / Dot1Q(type=0x8100) / IP() / UDP(dport=1234, sport=4321),
> +        ]
> +        with TestPmdShell() as testpmd:
> +            testpmd.stop_all_ports()
not needed
> +            testpmd.set_qinq_strip(0, True)
> +            testpmd.start_all_ports()
not needed
> +            testpmd.start()
> +
> +            rec1 = self.send_packet_and_capture(test_packets[0])
> +            rec2 = self.send_packet_and_capture(test_packets[1])
> +            rec3 = self.send_packet_and_capture(test_packets[2])
> +            rec4 = self.send_packet_and_capture(test_packets[3])
> +
> +            testpmd.stop()
> +
> +            try:
> +                context = "Single VLAN"
> +                self.strip_verify(rec1[0], 0, context)
> +                context = "Stacked VLAN"
> +                self.strip_verify(rec2[0], 1, context, check_id=True)
> +                context = "Single S-VLAN"
> +                self.strip_verify(rec3[0], 0, context)
> +                context = "QinQ"
> +                self.strip_verify(rec4[0], 0, context)
> +            except IndexError:
> +                self.log(f"{context} packet was dropped when it should have been received.")
this is actually virtually identical (with only one line of difference) 
to test_vlan_strip. Could we remove this duplication?

Luca