[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