[RFC PATCH 0/2] add qinq test suite

[RFC PATCH 0/2] add qinq test suite

[Dean Marx]

Hi all,

This patch series introduces a new functional test suite for
verifying QinQ (802.1ad) behavior across Poll Mode Drivers in DPDK.
While existing VLAN test coverage in DTS
includes basic single VLAN and dual VLAN tagging scenarios, 
it does not specifically validate QinQ encapsulation,
which is a distinct Ethernet frame format involving an
outer S-VLAN (Service VLAN) and an inner C-VLAN (Customer VLAN) tag.
This test suite focuses on validating both the software classification
and hardware offload capabilities provided by PMDs
for QinQ-tagged frames. The suite includes tests for:

QinQ Rx Parse - Ensures that double-tagged frames are correctly 
  classified with appropriate offload flags.
QinQ Strip - Verifies QinQ hardware stripping functionality,
  including correct VLAN Tag Control Information (TCI) handling.
QinQ Filter - Tests packet filtering logic based on S-VLAN ID.
QinQ Forwarding - Confirms both VLAN tags are preserved across 
  forwarding paths.
Mismatched TPID Handling - Tests behavior when the outer VLAN uses
  a non-standard TPID, ensuring proper fallback behavior.

Dean Marx (2):
  dts: add qinq strip and VLAN extend to testpmd shell
  dts: add qinq test suite

 dts/framework/remote_session/testpmd_shell.py |  52 ++++
 dts/tests/TestSuite_qinq.py                   | 234 ++++++++++++++++++
 2 files changed, 286 insertions(+)
 create mode 100644 dts/tests/TestSuite_qinq.py

-- 
2.49.0

[RFC PATCH 1/2] dts: add qinq strip and VLAN extend to testpmd shell

[Dean Marx]

Add QinQ strip and VLAN extend methods to TestPmdShell class.

Signed-off-by: Dean Marx <dmarx@iol.unh.edu>
---
 dts/framework/remote_session/testpmd_shell.py | 52 +++++++++++++++++++
 1 file changed, 52 insertions(+)

diff --git a/dts/framework/remote_session/testpmd_shell.py b/dts/framework/remote_session/testpmd_shell.py
index ac99bd09ef..aaab267a54 100644
--- a/dts/framework/remote_session/testpmd_shell.py
+++ b/dts/framework/remote_session/testpmd_shell.py
@@ -2099,6 +2099,58 @@ def set_vlan_filter(self, port: int, enable: bool, verify: bool = True) -> None:
                     filter on port {port}"""
                 )
 
+    def set_vlan_extend(self, port: int, enable: bool, verify: bool = True) -> None:
+        """Set vlan extend.
+
+        Args:
+            port: The port number to enable VLAN extend on.
+            enable: Enable extend on `port` if :data:`True`, otherwise disable it.
+            verify: If :data:`True`, the output of the command and show port info
+                is scanned to verify that vlan extend was set successfully.
+
+        Raises:
+            InteractiveCommandExecutionError: If `verify` is :data:`True` and extend
+                fails to update.
+        """
+        extend_cmd_output = self.send_command(f"vlan set extend {'on' if enable else 'off'} {port}")
+        if verify:
+            vlan_settings = self.show_port_info(port_id=port).vlan_offload
+            if enable ^ (vlan_settings is not None and VLANOffloadFlag.EXTEND in vlan_settings):
+                self._logger.debug(
+                    f"""Failed to {"enable" if enable else "disable"}
+                                   extend on port {port}: \n{extend_cmd_output}"""
+                )
+                raise InteractiveCommandExecutionError(
+                    f"""Failed to {"enable" if enable else "disable"} extend on port {port}"""
+                )
+
+    def set_qinq_strip(self, port: int, enable: bool, verify: bool = True) -> None:
+        """Set QinQ strip.
+
+        Args:
+            port: The port number to enable QinQ strip on.
+            enable: Enable stripping on `port` if :data:`True`, otherwise disable it.
+            verify: If :data:`True`, the output of the command and show port info
+                is scanned to verify that QinQ strip was set successfully.
+
+        Raises:
+            InteractiveCommandExecutionError: If `verify` is :data:`True` and QinQ strip
+                fails to update.
+        """
+        qinq_cmd_output = self.send_command(
+            f"vlan set qinq_strip {'on' if enable else 'off'} {port}"
+        )
+        if verify:
+            vlan_settings = self.show_port_info(port_id=port).vlan_offload
+            if enable ^ (vlan_settings is not None and VLANOffloadFlag.QINQ_STRIP in vlan_settings):
+                self._logger.debug(
+                    f"""Failed to {"enable" if enable else "disable"}
+                                   QinQ strip on port {port}: \n{qinq_cmd_output}"""
+                )
+                raise InteractiveCommandExecutionError(
+                    f"""Failed to {"enable" if enable else "disable"} QinQ strip on port {port}"""
+                )
+
     def set_mac_address(self, port: int, mac_address: str, verify: bool = True) -> None:
         """Set port's MAC address.
 
-- 
2.49.0

[RFC PATCH 2/2] dts: add qinq test suite

[Dean Marx]

Add QinQ test suite, which verifies PMD behavior when
sending QinQ (IEEE 802.1ad) packets.

Signed-off-by: Dean Marx <dmarx@iol.unh.edu>
---
 dts/tests/TestSuite_qinq.py | 240 ++++++++++++++++++++++++++++++++++++
 1 file changed, 240 insertions(+)
 create mode 100644 dts/tests/TestSuite_qinq.py

diff --git a/dts/tests/TestSuite_qinq.py b/dts/tests/TestSuite_qinq.py
new file mode 100644
index 0000000000..4bfa2c1920
--- /dev/null
+++ b/dts/tests/TestSuite_qinq.py
@@ -0,0 +1,240 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2025 University of New Hampshire
+
+"""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.
+
+"""
+
+from scapy.layers.inet import IP, UDP
+from scapy.layers.l2 import Dot1Q, Ether
+from scapy.packet import Packet
+
+from framework.params.testpmd import SimpleForwardingModes
+from framework.remote_session.testpmd_shell import PacketOffloadFlag, TestPmdShell
+from framework.test_suite import TestSuite, func_test
+from framework.testbed_model.capability import NicCapability, TopologyType, requires
+
+
+@requires(topology_type=TopologyType.two_links)
+@requires(NicCapability.RX_OFFLOAD_VLAN_FILTER)
+@requires(NicCapability.RX_OFFLOAD_VLAN_EXTEND)
+class TestQinq(TestSuite):
+    """QinQ test suite.
+
+    This suite consists of 3 test cases:
+    1. QinQ Rx parse: Ensures correct classification and detection of double VLAN packets (QinQ).
+    2. QinQ strip: Validates hardware offload of VLAN header removal and correct TCI population.
+    3. QinQ filter:
+
+    """
+
+    def send_packet_and_verify_offload_flags(
+        self, packet: Packet, testpmd: TestPmdShell, offload_flags: list[PacketOffloadFlag]
+    ) -> None:
+        """Send packet and verify offload flags match the stripping action.
+
+        Args:
+            packet: The packet to send to testpmd.
+            testpmd: The testpmd session to send commands to.
+            offload_flags: List of PacketOffloadFlags that should be in verbose output.
+        """
+        testpmd.start()
+        self.send_packet_and_capture(packet=packet)
+        verbose_output = testpmd.extract_verbose_output(testpmd.stop())
+        for flag in offload_flags:
+            self.verify(flag in verbose_output, f"Expected flag {flag} not found in output")
+
+    def send_packet_and_verify(
+        self, packet: Packet, testpmd: TestPmdShell, should_receive: bool
+    ) -> None:
+        """Send packet and verify reception.
+
+        Args:
+            packet: The packet to send to testpmd.
+            testpmd: The testpmd session to send commands to.
+            should_receive: If :data:`True`, verifies packet was received.
+        """
+        testpmd.start()
+        received = self.send_packet_and_capture(packet=packet)
+        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.")
+
+    @property
+    def test_packet(self):
+        """QinQ packet to be used in each test case."""
+        return (
+            Ether(dst="ff:ff:ff:ff:ff:ff")
+            / Dot1Q(vlan=100, type=0x8100)
+            / Dot1Q(vlan=200)
+            / IP(dst="1.2.3.4")
+            / UDP(dport=1234, sport=4321)
+        )
+
+    @func_test
+    def test_qinq_rx_parse(self) -> None:
+        """QinQ Rx parse test case.
+
+        Steps:
+            Launch testpmd with Rxonly forwarding mode.
+            Set verbose level to 1.
+            Enable VLAN filter/extend modes on port 0.
+            Send test packet and capture verbose output.
+
+        Verify:
+            Check that all expected offload flags are in verbose output.
+        """
+        offload_flags = [PacketOffloadFlag.RTE_MBUF_F_RX_QINQ, PacketOffloadFlag.RTE_MBUF_F_RX_VLAN]
+        with TestPmdShell(forward_mode=SimpleForwardingModes.rxonly) as testpmd:
+            testpmd.set_verbose(level=1)
+            testpmd.stop_all_ports()
+            testpmd.set_vlan_filter(0, True)
+            testpmd.set_vlan_extend(0, True)
+            testpmd.start_all_ports()
+            self.send_packet_and_verify_offload_flags(self.test_packet, testpmd, offload_flags)
+
+    @requires(NicCapability.RX_OFFLOAD_QINQ_STRIP)
+    @func_test
+    def test_qinq_strip(self) -> None:
+        """QinQ Rx strip test case.
+
+        Steps:
+            Launch testpmd with Rxonly forwarding mode.
+            Set verbose level to 1.
+            Enable VLAN filter/extend modes on port 0.
+            Enable QinQ strip mode on port 0.
+            Send test packet and capture verbose output.
+
+        Verify:
+            Check that all expected offload flags are in verbose output.
+        """
+        offload_flags = [
+            PacketOffloadFlag.RTE_MBUF_F_RX_QINQ_STRIPPED,
+            PacketOffloadFlag.RTE_MBUF_F_RX_VLAN_STRIPPED,
+            PacketOffloadFlag.RTE_MBUF_F_RX_VLAN,
+            PacketOffloadFlag.RTE_MBUF_F_RX_QINQ,
+        ]
+        with TestPmdShell(forward_mode=SimpleForwardingModes.rxonly) as testpmd:
+            testpmd.set_verbose(level=1)
+            testpmd.stop_all_ports()
+            testpmd.set_vlan_filter(0, True)
+            testpmd.set_vlan_extend(0, True)
+            testpmd.set_qinq_strip(0, True)
+            testpmd.start_all_ports()
+            self.send_packet_and_verify_offload_flags(self.test_packet, testpmd, offload_flags)
+
+    @func_test
+    def test_qinq_filter(self) -> None:
+        """QinQ Rx filter test case.
+
+        Steps:
+            Launch testpmd with Rxonly forwarding mode.
+            Enable VLAN filter/extend modes on port 0.
+            Add VLAN tag 100 to the filter on port 0.
+            Send test packet and capture verbose output.
+
+        Verify:
+            Check that all expected offload flags are in verbose output.
+        """
+        packets = [
+            Ether(dst="00:11:22:33:44:55", src="66:77:88:99:aa:bb")
+            / Dot1Q(vlan=100)
+            / Dot1Q(vlan=200)
+            / IP(dst="192.0.2.1", src="198.51.100.1")
+            / UDP(dport=1234, sport=5678),
+            Ether(dst="00:11:22:33:44:55", src="66:77:88:99:aa:bb")
+            / Dot1Q(vlan=101)
+            / Dot1Q(vlan=200)
+            / IP(dst="192.0.2.1", src="198.51.100.1")
+            / UDP(dport=1234, sport=5678),
+        ]
+        with TestPmdShell(forward_mode=SimpleForwardingModes.rxonly) 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)
+            self.send_packet_and_verify(packets[0], testpmd, should_receive=True)
+            self.send_packet_and_verify(packets[1], testpmd, should_receive=False)
+
+    @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.
+        """
+        with TestPmdShell(forward_mode=SimpleForwardingModes.mac) as testpmd:
+            testpmd.stop_all_ports()
+            testpmd.set_vlan_filter(0, False)
+            testpmd.start_all_ports()
+            received_packet = self.send_packet_and_capture(self.test_packet)
+            for packet in received_packet:
+                vlan_tags = [layer for layer in packet.layers() if layer == Dot1Q]
+                self.verify(len(vlan_tags) == 2, f"Expected 2 VLAN tags, found {len(vlan_tags)}")
+
+                if packet.haslayer(Dot1Q):
+                    outer_vlan_id = packet[Dot1Q].vlan
+                    self.verify(
+                        outer_vlan_id == 100,
+                        f"Outer VLAN ID was {outer_vlan_id} when it should have been 100.",
+                    )
+
+                if packet[Dot1Q].haslayer(Dot1Q):
+                    inner_vlan_id = packet[Dot1Q].payload[Dot1Q].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 rxonly 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(forward_mode=SimpleForwardingModes.rxonly) as testpmd:
+            testpmd.set_verbose(level=1)
+            testpmd.stop_all_ports()
+            testpmd.set_vlan_filter(0, False)
+            testpmd.start_all_ports()
+
+            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)
+
+            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
+                self.verify(vlan_id == 200, f"Expected inner VLAN ID 200, got {vlan_id}")
-- 
2.49.0

[PATCH v1 1/2] dts: add qinq strip and VLAN extend to testpmd shell

[Dean Marx]

Add QinQ strip and VLAN extend methods to TestPmdShell class.

Signed-off-by: Dean Marx <dmarx@iol.unh.edu>
---
 dts/framework/remote_session/testpmd_shell.py | 52 +++++++++++++++++++
 1 file changed, 52 insertions(+)

diff --git a/dts/framework/remote_session/testpmd_shell.py b/dts/framework/remote_session/testpmd_shell.py
index ad8cb273dc..6caa6b2772 100644
--- a/dts/framework/remote_session/testpmd_shell.py
+++ b/dts/framework/remote_session/testpmd_shell.py
@@ -2112,6 +2112,58 @@ def set_vlan_filter(self, port: int, enable: bool, verify: bool = True) -> None:
                     filter on port {port}"""
                 )
 
+    def set_vlan_extend(self, port: int, enable: bool, verify: bool = True) -> None:
+        """Set vlan extend.
+
+        Args:
+            port: The port number to enable VLAN extend on.
+            enable: Enable extend on `port` if :data:`True`, otherwise disable it.
+            verify: If :data:`True`, the output of the command and show port info
+                is scanned to verify that vlan extend was set successfully.
+
+        Raises:
+            InteractiveCommandExecutionError: If `verify` is :data:`True` and extend
+                fails to update.
+        """
+        extend_cmd_output = self.send_command(f"vlan set extend {'on' if enable else 'off'} {port}")
+        if verify:
+            vlan_settings = self.show_port_info(port_id=port).vlan_offload
+            if enable ^ (vlan_settings is not None and VLANOffloadFlag.EXTEND in vlan_settings):
+                self._logger.debug(
+                    f"""Failed to {"enable" if enable else "disable"}
+                                   extend on port {port}: \n{extend_cmd_output}"""
+                )
+                raise InteractiveCommandExecutionError(
+                    f"""Failed to {"enable" if enable else "disable"} extend on port {port}"""
+                )
+
+    def set_qinq_strip(self, port: int, enable: bool, verify: bool = True) -> None:
+        """Set QinQ strip.
+
+        Args:
+            port: The port number to enable QinQ strip on.
+            enable: Enable stripping on `port` if :data:`True`, otherwise disable it.
+            verify: If :data:`True`, the output of the command and show port info
+                is scanned to verify that QinQ strip was set successfully.
+
+        Raises:
+            InteractiveCommandExecutionError: If `verify` is :data:`True` and QinQ strip
+                fails to update.
+        """
+        qinq_cmd_output = self.send_command(
+            f"vlan set qinq_strip {'on' if enable else 'off'} {port}"
+        )
+        if verify:
+            vlan_settings = self.show_port_info(port_id=port).vlan_offload
+            if enable ^ (vlan_settings is not None and VLANOffloadFlag.QINQ_STRIP in vlan_settings):
+                self._logger.debug(
+                    f"""Failed to {"enable" if enable else "disable"}
+                                   QinQ strip on port {port}: \n{qinq_cmd_output}"""
+                )
+                raise InteractiveCommandExecutionError(
+                    f"""Failed to {"enable" if enable else "disable"} QinQ strip on port {port}"""
+                )
+
     def set_mac_address(self, port: int, mac_address: str, verify: bool = True) -> None:
         """Set port's MAC address.
 
-- 
2.50.1

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

[Dean Marx]

Add QinQ test suite, which verifies PMD behavior when
sending QinQ (IEEE 802.1ad) packets.

Signed-off-by: Dean Marx <dmarx@iol.unh.edu>
---
 dts/tests/TestSuite_qinq.py | 329 ++++++++++++++++++++++++++++++++++++
 1 file changed, 329 insertions(+)
 create mode 100644 dts/tests/TestSuite_qinq.py

diff --git a/dts/tests/TestSuite_qinq.py b/dts/tests/TestSuite_qinq.py
new file mode 100644
index 0000000000..7412e9bf04
--- /dev/null
+++ b/dts/tests/TestSuite_qinq.py
@@ -0,0 +1,329 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2025 University of New Hampshire
+
+"""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.
+
+"""
+
+from scapy.layers.inet import IP, UDP
+from scapy.layers.l2 import Dot1AD, Dot1Q, Ether
+from scapy.packet import Packet
+
+from framework.remote_session.testpmd_shell import TestPmdShell
+from framework.test_suite import TestSuite, func_test
+from framework.testbed_model.capability import NicCapability, requires
+
+
+class TestQinq(TestSuite):
+    """QinQ test suite.
+
+    This suite consists of 4 test cases:
+    1. QinQ Filter: Enable VLAN filter and verify packets with mismatched VLAN IDs are dropped,
+        and packets with matching VLAN IDs are received.
+    2. QinQ Forwarding: Send a QinQ packet and verify the received packet contains
+        both QinQ/VLAN layers.
+    3. Mismatched TPID: Send a Qinq packet with an invalid TPID (not 0x8100/0x88a8) and verify the
+        mismatched VLAN layer is interpreted as part of the Ethertype (expected DPDK behavior.)
+    4. VLAN Strip: Enable VLAN stripping and verify sent packets are received with the
+        expected VLAN/QinQ layers.
+    5. QinQ Strip: Enable VLAN/QinQ stripping and verify sent packets are received with the
+        expected VLAN/QinQ layers.
+    """
+
+    def send_packet_and_verify(
+        self, packet: Packet, testpmd: TestPmdShell, should_receive: bool
+    ) -> None:
+        """Send packet and verify reception.
+
+        Args:
+            packet: The packet to send to testpmd.
+            testpmd: The testpmd session to send commands to.
+            should_receive: If :data:`True`, verifies packet was received.
+        """
+        testpmd.start()
+        received = self.send_packet_and_capture(packet=packet)
+        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.")
+
+    @requires(NicCapability.RX_OFFLOAD_VLAN_EXTEND)
+    @func_test
+    def test_qinq_filter(self) -> None:
+        """QinQ Rx filter test case.
+
+        Steps:
+            Launch testpmd with mac forwarding mode.
+            Enable VLAN filter/extend modes on port 0.
+            Add VLAN tag 100 to the filter on port 0.
+            Send test packet and capture verbose output.
+
+        Verify:
+            Packet with matching VLAN ID is received.
+            Packet with mismatched VLAN ID is dropped.
+        """
+        packets = [
+            Ether(dst="00:11:22:33:44:55", src="66:77:88:99:aa:bb")
+            / Dot1Q(vlan=100)
+            / Dot1Q(vlan=200)
+            / IP(dst="192.0.2.1", src="198.51.100.1")
+            / UDP(dport=1234, sport=5678),
+            Ether(dst="00:11:22:33:44:55", src="66:77:88:99:aa:bb")
+            / Dot1Q(vlan=101)
+            / Dot1Q(vlan=200)
+            / IP(dst="192.0.2.1", src="198.51.100.1")
+            / UDP(dport=1234, sport=5678),
+        ]
+        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)
+
+    @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()
+            testpmd.set_vlan_filter(0, False)
+            testpmd.start_all_ports()
+            testpmd.start()
+            received_packet = self.send_packet_and_capture(test_packet)
+
+            self.verify(
+                received_packet != [], "Packet was dropped when it should have been received."
+            )
+
+            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)}")
+
+                if packet.haslayer(Dot1Q):
+                    outer_vlan_id = packet[Dot1Q].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
+                    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()
+            testpmd.set_vlan_filter(0, False)
+            testpmd.start_all_ports()
+            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)
+
+            self.verify(
+                received_packet != [], "Packet was dropped when it should have been received."
+            )
+
+            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
+                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
+    ) -> 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")
+            else:
+                vlan_layer = packet.getlayer(Dot1Q)
+                self.verify(
+                    vlan_layer is not None
+                    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
+
+            tag_count = count_vlan_tags(packet)
+            self.verify(
+                tag_count == 1,
+                f"""Expected one 0x8100 VLAN tag but found {tag_count}: {packet.summary()}
+                sent packet: {context}""",
+            )
+            first_dot1q = packet.getlayer(Dot1Q)
+            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}""",
+            )
+            if check_id:
+                self.verify(
+                    packet[Dot1Q].vlan == 200,
+                    f"""VLAN ID 200 not found in received packet: {packet.summary()}
+                    sent packet: {context}""",
+                )
+
+    @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()
+            testpmd.set_vlan_strip(0, True)
+            testpmd.start_all_ports()
+            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."
+                )
+
+    @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()
+            testpmd.set_qinq_strip(0, True)
+            testpmd.start_all_ports()
+            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.")
-- 
2.50.1

Re: [PATCH v1 1/2] dts: add qinq strip and VLAN extend to testpmd shell

[Luca Vizzarro]

Looks good to me, just some minor comments.

On 17/07/2025 21:57, Dean Marx wrote:
> +    def set_vlan_extend(self, port: int, enable: bool, verify: bool = True) -> None:
> <snip>
> +            if enable ^ (vlan_settings is not None and VLANOffloadFlag.EXTEND in vlan_settings):
> +                self._logger.debug(
> +                    f"""Failed to {"enable" if enable else "disable"}
> +                                   extend on port {port}: \n{extend_cmd_output}"""
> +                )
> +                raise InteractiveCommandExecutionError(
> +                    f"""Failed to {"enable" if enable else "disable"} extend on port {port}"""
> +                )
Please don't use triple quotes if you need to break a line, or if you 
don't need a reason like for InteractiveCommandExecutionError.

Breaking a line with triple quotes will result in all the whitespace 
being considered as part of the string. You can just as easily split by 
writing multiple quoted text next to each other:

   self._logger.debug(
       f"Failed to {...} extend "
       f"on port {port}..."
   )

> +    def set_qinq_strip(self, port: int, enable: bool, verify: bool = True) -> None:
> <snip>
> +            if enable ^ (vlan_settings is not None and VLANOffloadFlag.QINQ_STRIP in vlan_settings):
> +                self._logger.debug(
> +                    f"""Failed to {"enable" if enable else "disable"}
> +                                   QinQ strip on port {port}: \n{qinq_cmd_output}"""
> +                )
> +                raise InteractiveCommandExecutionError(
> +                    f"""Failed to {"enable" if enable else "disable"} QinQ strip on port {port}"""
> +                )
Same problem as above.

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

[Luca Vizzarro]

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