[PATCH v2 5/5] dts: add test suite for RX and TX offloads

[jspewock@iol.unh.edu]

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

This patch adds a test sutie that ports over and expands upon
functionality provided in the RxTx test sutie in Old DTS. This test
suite provides convenience methods and decorators in an attempt to
reduce code duplication when developers are tasked with testing the
same offloaded functionality through 3 different mediums (passed on the
command-line, configured on a port at runtime, and configured on queues
at runtime).

Signed-off-by: Jeremy Spewock <jspewock@iol.unh.edu>
---
 dts/framework/config/conf_yaml_schema.json |   3 +-
 dts/tests/TestSuite_rxtx_offload.py        | 627 +++++++++++++++++++++
 2 files changed, 629 insertions(+), 1 deletion(-)
 create mode 100644 dts/tests/TestSuite_rxtx_offload.py

diff --git a/dts/framework/config/conf_yaml_schema.json b/dts/framework/config/conf_yaml_schema.json
index f02a310bb5..c1243ea5d8 100644
--- a/dts/framework/config/conf_yaml_schema.json
+++ b/dts/framework/config/conf_yaml_schema.json
@@ -187,7 +187,8 @@
       "enum": [
         "hello_world",
         "os_udp",
-        "pmd_buffer_scatter"
+        "pmd_buffer_scatter",
+        "rxtx_offload"
       ]
     },
     "test_target": {
diff --git a/dts/tests/TestSuite_rxtx_offload.py b/dts/tests/TestSuite_rxtx_offload.py
new file mode 100644
index 0000000000..d994b44fc1
--- /dev/null
+++ b/dts/tests/TestSuite_rxtx_offload.py
@@ -0,0 +1,627 @@
+"""Rx/Tx offload configuration suite.
+
+The goal of this suite is to test the support for three different methods of offloading different
+capabilities using testpmd: On a queue, on a port, and on the command-line. Support for configuring
+the capability through different means is testing alongside verifying the functionality of the
+capability when offloaded. Each of the three methods of setting the capability should be tested
+using the same criteria to monitor for differences between the methods.
+
+Testing consists of enabling the capability if it wasn't passed in through the command-line,
+verifying that the capability performs it's expected task, then, in the general case, disabling the
+capability and verifying that the same result is not achieved in a default state. Some cases do not
+check the base-case since their functionality is enabled by default without offloading the
+capability (like invalid checksum verification, for example).
+
+There should be test cases for each of the 3 configuration strategies for every offload that is
+tested. Additionally, there are two additional test cases that validates the ability to enable
+every offload that a device supports on its port without actually testing the functionality of the
+offload for both Rx and Tx.
+"""
+
+import random
+from typing import Callable, ClassVar, Protocol, TypeVar
+
+from scapy.layers.inet import IP, TCP, UDP  # type: ignore[import-untyped]
+from scapy.layers.l2 import Dot1Q, Ether  # type: ignore[import-untyped]
+from scapy.packet import Packet, Raw  # type: ignore[import-untyped]
+from typing_extensions import Unpack
+
+from framework.exception import InteractiveCommandExecutionError, TestCaseVerifyError
+from framework.params.types import TestPmdParamsDict
+from framework.remote_session.testpmd_shell import (
+    FlowRule,
+    NicCapability,
+    OffloadCapability,
+    OLFlag,
+    RxOffloadCapability,
+    SimpleForwardingModes,
+    TestPmdShell,
+    TestPmdVerbosePacket,
+    TxOffloadCapability,
+)
+from framework.test_suite import TestSuite, func_test
+from framework.testbed_model.capability import requires
+
+T = TypeVar("T")
+
+
+class DecoratedFuncType(Protocol):
+    """Protocol used to provide a useful typehint for methods that are decorated.
+
+    Methods decorated by :meth:`TestRxtxOffload.setup_testpmd` are able to pass kwargs into the
+    a :class:`TestPmdShell` upon creation and therefore have many non-obvious arguments. This type
+    allows static type checkers the ability to unpack and expose all of those non-obvious
+    arguments.
+    """
+
+    def __call__(
+        self: T,
+        port_id: int,
+        no_set: bool = False,
+        modify_queues: bool = False,
+        no_invert: bool = False,
+        **kwargs: Unpack[TestPmdParamsDict],
+    ) -> None:
+        """Function stub to create callable type.
+
+        Args:
+            test: Instance of the test suite class that the methods belong to.
+            port_id: ID of the port to set the offloads on.
+            no_set: Whether to enable the offload before testing or not. When :data:`True`,
+                the method will validate that the offload is already configured rather than
+                enabling it. This is used in test cases that enable the offload using the
+                command-line. Defaults to :data:`False`.
+            modify_queues: Whether to add offloads to individual queues or the entire port.
+                If :data:`True`, individual queues will be modified, otherwise the whole
+                port will. Defaults to :data:`False`.
+            no_invert: If :data:`True` skip testing behavior of testpmd without the offload
+                enabled. Defaults to :data:`False`.
+        """
+        ...
+
+
+def set_offload(
+    offload: OffloadCapability,
+    testpmd: TestPmdShell,
+    is_rx: bool,
+    port_id: int,
+    queues_to_modify: set[int],
+    on: bool,
+) -> None:
+    """Set offload on a port/queue using testpmd.
+
+    This helper method allows you to reach both the :meth:`TestPmdShell.set_queue_offload` and
+    :meth:`TestPmdShell.set_port_offload` through a single method, and update the offloads on
+    multiple queues at once.
+
+    Args:
+        offload: Offload to configure.
+        testpmd: Testpmd shell to use for modifying the offload configuration. This shell is
+            expected to be running before being passed into this method.
+        is_rx: If :data:`True` the offload will be configured on an Rx port/queue, otherwise it
+            will be configured on a Tx port/queue.
+        port_id: ID of the port to make configuration changes on.
+        queues_to_modify: IDs of queues to make configuration changes on. If this set is empty then
+            changes will only be made to the port configuration.
+        on: If :data:`True`, then enable the offload, otherwise disable it.
+    """
+    if len(queues_to_modify) > 0:
+        for queue_id in queues_to_modify:
+            testpmd.set_queue_offload(port_id, queue_id, is_rx, offload, on, verify=True)
+    else:
+        testpmd.set_port_offload(port_id, is_rx, offload, on, verify=True)
+
+
+class TestRxtxOffload(TestSuite):
+    """Rx/Tx offload testing suite.
+
+    There are three ways to configure offloads in testpmd: Through the command line, at runtime on
+    a port, or at runtime on individual queues. Each of these configuration methods should be
+    tested for each of the offloads that are tested in this suite. Methods for testing each of the
+    offloads functionalities should be designed to be decorated by :func:`setup_testpmd`, meaning
+    their first parameter must be a :class:`TestPmdShell` (which will be running once the method
+    receives it) and the second parameter must be a bool signifying whether testing is to be done
+    on an entire port, or individual queues (:data:`True` for testing on a port, otherwise testing
+    on a queue). This decorator allows for testpmd to be configured with the proper offloads before
+    being passed to the method for testing.
+    """
+
+    #:
+    rx_port_for_testing: ClassVar[int] = 0
+    #:
+    tx_port_for_testing: ClassVar[int] = 1
+    #: Specify the number of queues to use for test cases the set offloads on queues. This should
+    #: generally be an odd integer since the test cases will apply offloads to the majority of
+    #: queues.
+    number_of_queues: ClassVar[int] = 5
+    #: Common testpmd parameters for all tests to use.
+    common_testpmd_params: ClassVar[dict] = {
+        "forward_mode": SimpleForwardingModes.mac,
+        "rx_queues": number_of_queues,
+        "tx_queues": number_of_queues,
+    }
+    #:
+    relevant_checksum_errors: ClassVar[OLFlag] = (
+        OLFlag.RTE_MBUF_F_RX_IP_CKSUM_BAD
+        | OLFlag.RTE_MBUF_F_RX_L4_CKSUM_BAD
+        | OLFlag.RTE_MBUF_F_RX_OUTER_L4_CKSUM_BAD
+    )
+
+    @staticmethod
+    def setup_testpmd(
+        offloads: OffloadCapability, is_rx: bool
+    ) -> Callable[[Callable[["TestRxtxOffload", TestPmdShell, bool], bool]], DecoratedFuncType]:
+        """Decorator function that initializes testpmd before running a test.
+
+        All offloads in `offloads` are either enabled if they weren't passed in on the
+        command-line, or verified to be enabled on the port before running the decorated method.
+        The decorated method must return a boolean that represents whether or not the expected
+        behavior took place while the offload was enabled which will be used for verification that
+        the test functioned as expected. By default, this decorator will also disable all of the
+        offloads in `offloads` and verify that the decorated method instead returns :data:`False`
+        where it was expected to return :data:`True` with them enabled. This functionality can be
+        overridden, however, for cases where the opposite functionality cannot be easily tested
+        (like invalid checksums, for example).
+
+        The decorated method additionally can be used for both setting offloads across the entire
+        port or on individual queues using the `modify_queues` parameter. If individual queues are
+        being modified then the majority of queues ((x / 2) + 1 where x is the number of queues)
+        will have the offload enabled rather than all of them. All decorated functions must be a
+        method of :class:`TestRxtxOffload` and take a testpmd shell for testing along with a
+        boolean that signifies whether it is a port being tested or individual queues.
+
+        Args:
+            offloads: Offloads to set in testpmd prior to running decorated function.
+            is_rx: Whether to set Rx or Tx offloads when testing.
+
+        Returns:
+            Decorator function that enables all offloads, runs the decorated method, then optionally
+            disables offloads and runs the test method again.
+        """
+
+        def wrap(
+            func: Callable[["TestRxtxOffload", TestPmdShell, bool], bool]
+        ) -> DecoratedFuncType:
+            def wrapper(
+                test: "TestRxtxOffload",
+                port_id: int,
+                no_set: bool = False,
+                modify_queues: bool = False,
+                no_invert: bool = False,
+                **kwargs: Unpack[TestPmdParamsDict],
+            ) -> None:
+                """Function that wraps the decorated method.
+
+                Refer to :class:`DecoratedFuncType` for information about parameters of this
+                method.
+                """
+                queues_to_modify: set[int] = set()
+                if modify_queues:
+                    while len(queues_to_modify) < int(test.number_of_queues / 2 + 1):
+                        queues_to_modify.add(random.randint(0, test.number_of_queues - 1))
+                with TestPmdShell(test.sut_node, **test.common_testpmd_params, **kwargs) as testpmd:
+                    # If no_set then it should have been passed on the command-line and already
+                    # be configured, else we need to set them ourselves.
+                    if no_set:
+                        testpmd.is_port_offload_configured(port_id, is_rx, offloads)
+                    else:
+                        set_offload(offloads, testpmd, is_rx, port_id, queues_to_modify, on=True)
+
+                    test.verify(
+                        func(test, testpmd, not modify_queues),
+                        f"Offloaded capabilities ({offloads}) failed.",
+                    )
+
+                    set_offload(offloads, testpmd, is_rx, port_id, queues_to_modify, on=False)
+                    if not no_invert:
+                        test.verify(
+                            not func(test, testpmd, not modify_queues),
+                            f"After disabling capabilities ({offloads}) the "
+                            "result was the same.",
+                        )
+
+            # This type ignore is required to ignore the mismatch in types between the protocol
+            # and the wrapper function. Mypy complains that these two are not the same type since
+            # the type produced by the protocol does not include a parameter for "TestRxtxOffload".
+            # However, if you add this parameter, it makes it so that you have to explicitly pass
+            # the value of `self` into the method when you call it, so you cannot simply call
+            # methods with the standard syntax of "self.<meth_name>" without mypy complaining.
+            return wrapper  # type: ignore[return-value]
+
+        return wrap
+
+    """ ========== Verify methods ==========
+        All verify methods must match the type
+        Callable[[list[Packet], list[TestPmdVerbosePacket]], bool] where their
+        first parameter is a list of packets that were forwarded back to the TG in the test and the
+        second parameter is the verbose information gathered by testpmd while packets were being
+        forwarded. These methods are meant to return a boolean that represents whether or not a
+        packet that matches their expected behavior can be found among any of their parameters in
+        order to be used by :meth:`send_packets_and_verify` for testing functionality.
+    """
+
+    @staticmethod
+    def _packet_is_relevant(pakt: Packet) -> bool:
+        """Helper method to test whether or not a packet was sent by a method in this test suite.
+
+        All packets in this test suite are sent with a payload of 20 "X" characters, so this method
+        checks to see if `pakt` has a matching payload.
+
+        Args:
+            pakt: The packet to validate.
+
+        Returns:
+            :data:`True` if the packet has a valid payload, :data:`False` otherwise.
+        """
+        return hasattr(pakt, "load") and pakt.load == bytes("X" * 20, "utf-8")
+
+    @staticmethod
+    def verify_insertion(packets: list[Packet], *_) -> bool:
+        """Method to verify VLAN headers were inserted into sent packets.
+
+        Checks to make sure that there is at least one packet in `packets` that is relevant to the
+        test suite with a VLAN header in it. This method consumes all arguments after `packets` to
+        conform to the type
+        Callable[[list[Packet], list[TestPmdVerbosePacket]], bool] even though
+        it does not need the verbose packet information.
+
+        Args:
+            packets: Packets to scan for a valid result.
+
+        Returns:
+            :data:`True` if there is at least one packet in `packets` that is relevant to the test
+            suite with a VLAN header in it, :data:`False` otherwise.
+        """
+        return any(TestRxtxOffload._packet_is_relevant(pakt) and Dot1Q in pakt for pakt in packets)
+
+    @staticmethod
+    def verify_stripping(packets: list[Packet], *_) -> bool:
+        """Method to verify VLAN headers were stripped from sent packets.
+
+        Checks to make sure there is at least one packet in `packets` that is relevant to the suite
+        which does not contain a VLAN header. This method consumes all arguments after `packets` to
+        conform to the type
+        Callable[[list[Packet], list[TestPmdVerbosePacket]], bool] even though
+        it does not need the verbose packets.
+
+        Args:
+            packets: The list of packets to scan for a valid result.
+
+        Returns:
+            :data:`True` if there is at least one packet that is relevant to the suite in `packets`
+            without a VLAN header, :data:`False` otherwise.
+        """
+        return any(
+            TestRxtxOffload._packet_is_relevant(pakt) and Dot1Q not in pakt for pakt in packets
+        )
+
+    @classmethod
+    def verify_chksum(
+        cls, expected_errors: OLFlag | None = None
+    ) -> Callable[[list[Packet], list[TestPmdVerbosePacket]], bool]:
+        """Verify that the expected checksum errors appear in the forwarding statistics.
+
+        Provides a closure for the wrapped function that stores the errors to expect in the
+        output. The wrapped function then accepts a list of packets received from forwarding as
+        well as forwarding statistics to validate with. The list of received packets are unused.
+
+        Args:
+            expected_error: Errors to search for in the forwarding statistics. If flag is
+                :data:`None`, verify that none of the :attr:`relevant_checksum_errors` appear in
+                the forwarding status. Defaults to :data:`None`.
+
+        Returns:
+            A function that searches for the expected errors in the forwarding statistics and
+            returns :data:`True` if they are all present or if `expected_errors` is :data:`None`
+            and there are no errors present. The function will return :data:`False` otherwise.
+        """
+
+        def wrap(_, fwd_stats: list[TestPmdVerbosePacket]) -> bool:
+            """Method that fits the expected type of verify methods."""
+            if expected_errors is None:
+                return all(
+                    cls.relevant_checksum_errors & pakt.ol_flags == OLFlag(0) for pakt in fwd_stats
+                )
+            else:
+                return any(expected_errors in pakt.ol_flags for pakt in fwd_stats)
+
+        return wrap
+
+    """ ========== End verify methods ========= """
+
+    def send_packets_and_verify(
+        self,
+        testpmd: TestPmdShell,
+        packets: list[Packet],
+        verify_meth: Callable[[list[Packet], list[TestPmdVerbosePacket]], bool],
+        is_port: bool,
+    ) -> bool:
+        """Send packets and verify the result using `verify_meth`.
+
+        If testing is being done on queues, this method makes flow rules that make it so that
+        anything that has the source address 192.168.1.X will be handled by queue X in testpmd.
+        Then, before sending packets in all cases, this method will adjust the source IP addresses
+        of the packets in `packets` to ensure they are all different and processed by different,
+        start testpmd, send all of the packets in `packets`, and captures the resulting received
+        traffic and forwarding stats. The received traffic and forwarding stats are then passed
+        into `verify_meth` to test if the expected result took place.
+
+        In the case of testing the results of queues, it is expected that the majority of queues
+        (but not all) have the offload enabled and, therefore, only the majority of packets in
+        `packets` need to return :data:`True` from `verify_meth` to get a positive
+        result.
+
+        Args:
+            testpmd: Testpmd shell that is handling the packet forwarding. This shell is expected
+                to already be started before being passed into this method.
+            packets: List of packets to send to the SUT during testing.
+            verify_meth: Method used to verify that the packet matches the expected results.
+            is_port: Flag to differentiate testing results from an offload set on a port from an
+                offload set on a queue.
+
+        Returns:
+            :data:`True` if `verify_meth` returned :data:`True` for every packet in `packets` if
+            testing offloads on a port, or the majority of packets if testing offloads on queues.
+            :data:`False` otherwise.
+        """
+        majority = int(self.number_of_queues / 2 + 1)
+        dst_ip_addr = "192.168.1.{}"
+        verify_results: list[bool] = []
+        base_pattern = f"eth / {'vlan /' if packets[0].haslayer(Dot1Q) else ''} " "ipv4 dst is {}"
+        base_action = "queue index {}"
+        if not is_port:
+            rule = FlowRule(self.rx_port_for_testing, True, "", "")
+            for queue_id in range(self.number_of_queues):
+                rule.pattern = base_pattern.format(dst_ip_addr.format(queue_id))
+                rule.actions = base_action.format(queue_id)
+                testpmd.flow_create(rule)
+
+        for ind, pakt in enumerate(packets):
+            testpmd.start()
+            pakt.getlayer(IP).dst = dst_ip_addr.format(ind % self.number_of_queues)
+            recv = self.send_packet_and_capture(pakt)
+            stats = TestPmdShell.extract_verbose_output(testpmd.stop())
+            verify_results.append(verify_meth(recv, stats))
+        return all(verify_results) if is_port else verify_results.count(True) == majority
+
+    """ ========== Functionality testing methods ==========
+        Functionality testing methods are the methods that test, assuming the offloads are already
+        configured in the environment, Rx/Tx offloads perform their expected behavior. These
+        methods must conform to the type Callable[["TestRxtxOffload", TestPmdShell, bool], bool]
+        so that they can be decorated by :func:`setup_testpmd`.
+    """
+
+    @setup_testpmd(TxOffloadCapability.VLAN_INSERT, False)
+    def tx_vlan_insertion_test(self, testpmd: TestPmdShell, is_port: bool) -> bool:
+        """Testing method for VLAN insertion on a Tx port/queue.
+
+        Testing is done by sending one packet without a VLAN header for as many queues as there are
+        configured and verifying with :meth:`verify_insertion`.
+
+        Args:
+            testpmd: Testpmd shell to use for testing. It is expected that this shell is already
+                running.
+            is_port: If :data:`True`, do testing on a port, otherwise do testing on individual
+                queues.
+
+        Returns:
+            :data:`True` if the received traffic contained a VLAN header when expected with the
+            offload enabled.
+        """
+        return self.send_packets_and_verify(
+            testpmd,
+            [Ether() / IP() / Raw("X" * 20)] * self.number_of_queues,
+            TestRxtxOffload.verify_insertion,
+            is_port,
+        )
+
+    @setup_testpmd(RxOffloadCapability.VLAN_STRIP, True)
+    def rx_vlan_stripping(self, testpmd: TestPmdShell, is_port: bool) -> bool:
+        """Testing method for VLAN stripping on an Rx port/queue.
+
+        Testing is done by sending one packet with a VLAN header for every configured queue and
+        verifying using :func:`verify_stripping` that the received packets had their VLAN header
+        stripped.
+
+        Args:
+            testpmd: Testpmd shell to use for testing. This shell is expected to already be
+                running when passed into this method.
+            is_port: If :data:`True`, do testing on a port, otherwise do testing on individual
+                queues.
+
+        Returns:
+            :data:`True` if the expected amount of received packets had their VLAN headers stripped
+            when the offload is enabled, :data:`False` otherwise.
+        """
+        return self.send_packets_and_verify(
+            testpmd,
+            [Ether() / Dot1Q() / IP() / Raw("X" * 20)] * self.number_of_queues,
+            TestRxtxOffload.verify_stripping,
+            is_port,
+        )
+
+    @setup_testpmd(
+        RxOffloadCapability.UDP_CKSUM
+        | RxOffloadCapability.TCP_CKSUM
+        | RxOffloadCapability.IPV4_CKSUM,
+        True,
+    )
+    def rx_cksum_test(self, testpmd: TestPmdShell, is_port: bool) -> bool:
+        """Test for verifying invalid checksum reporting.
+
+        Testing is done in multiple stages for checksum test cases. Testpmd must first be set to
+        Rx-only verbose mode to capture the checksum errors reported by testpmd when receiving
+        packets with a bad checksum. There are then 4 main cases to test:
+
+        * Valid packets do not display any checksum errors.
+        * Packets with a bad IP checksum and a bad layer 4 checksum report both errors properly.
+        * Packets with only a bad layer 4 checksum report only that error properly.
+        * Packets with only a bad IP checksum report only that error properly.
+
+        All of these cases must pass for this method to return :data:`True`.
+
+        Args:
+            testpmd: Testpmd shell to use for testing. It is expected that this shell is already
+                running when it is passed into this method.
+            is_port: If :data:`True`, do testing on a port, otherwise do testing on individual
+                queues.
+
+        Returns:
+            :data:`True` if all 4 cases pass, :data:`False` otherwise.
+        """
+        testpmd.set_verbose(1)
+        results = [
+            self.send_packets_and_verify(
+                testpmd, [Ether() / IP() / TCP() / ("X" * 20)], self.verify_chksum(), is_port
+            ),
+            self.send_packets_and_verify(
+                testpmd,
+                [Ether() / IP(chksum=0x0) / TCP(chksum=0xF) / ("X" * 20)],
+                TestRxtxOffload.verify_chksum(
+                    OLFlag.RTE_MBUF_F_RX_IP_CKSUM_BAD | OLFlag.RTE_MBUF_F_RX_L4_CKSUM_BAD
+                ),
+                is_port,
+            ),
+            self.send_packets_and_verify(
+                testpmd,
+                [Ether() / IP() / UDP(chksum=0xF) / ("X" * 20)],
+                TestRxtxOffload.verify_chksum(OLFlag.RTE_MBUF_F_RX_L4_CKSUM_BAD),
+                is_port,
+            ),
+            self.send_packets_and_verify(
+                testpmd,
+                [Ether() / IP(chksum=0x0) / UDP() / ("X" * 20)],
+                TestRxtxOffload.verify_chksum(OLFlag.RTE_MBUF_F_RX_IP_CKSUM_BAD),
+                is_port,
+            ),
+        ]
+        return all(results)
+
+    """ ========== End functionality testing methods ========== """
+
+    @requires(NicCapability.PORT_TX_OFFLOAD_VLAN_INSERT)
+    @func_test
+    def test_tx_port_vlan_insertion(self) -> None:
+        """Run :meth:`tx_vlan_insertion_test` with common testpmd parameters."""
+        self.tx_vlan_insertion_test(self.tx_port_for_testing)
+
+    @requires(NicCapability.PORT_TX_OFFLOAD_VLAN_INSERT)
+    @func_test
+    def test_tx_cmdline_vlan_insertion(self) -> None:
+        """Run :meth:`tx_vlan_insertion_test` with insertion offload passed on the command-line.
+
+        This requires specifying that the offload should not be set at runtime when calling the
+        method.
+        """
+        self.tx_vlan_insertion_test(self.tx_port_for_testing, no_set=True, tx_offloads=0x0001)
+
+    @requires(NicCapability.QUEUE_TX_OFFLOAD_VLAN_INSERT)
+    @func_test
+    def test_tx_queue_vlan_insertion(self) -> None:
+        """Run :meth:`tx_vlan_insertion_test` specifying queues should be modified."""
+        self.tx_vlan_insertion_test(self.tx_port_for_testing, modify_queues=True)
+
+    @requires(NicCapability.PORT_RX_OFFLOAD_VLAN_STRIP)
+    @func_test
+    def test_rx_port_vlan_strip(self) -> None:
+        """Run :meth:`rx_vlan_stripping` with common testpmd parameters."""
+        self.rx_vlan_stripping(self.rx_port_for_testing)
+
+    @requires(NicCapability.PORT_RX_OFFLOAD_VLAN_STRIP)
+    @func_test
+    def test_rx_cmdline_vlan_strip(self) -> None:
+        """Run :meth:`rx_vlan_stripping` with stripping offload passed on the command-line.
+
+        This requires specifying that the offload should not be set at runtime when calling the
+        method.
+        """
+        self.rx_vlan_stripping(self.rx_port_for_testing, no_set=True, enable_hw_vlan_strip=True)
+
+    @requires(NicCapability.QUEUE_RX_OFFLOAD_VLAN_STRIP)
+    @func_test
+    def test_rx_queue_vlan_strip(self) -> None:
+        """Run :meth:`rx_vlan_stripping` specifying queues should be modified."""
+        self.rx_vlan_stripping(self.rx_port_for_testing, modify_queues=True)
+
+    @requires(
+        NicCapability.PORT_RX_OFFLOAD_UDP_CKSUM,
+        NicCapability.PORT_RX_OFFLOAD_TCP_CKSUM,
+        NicCapability.PORT_RX_OFFLOAD_IPV4_CKSUM,
+    )
+    @func_test
+    def test_rx_port_chksum(self) -> None:
+        """Run :meth:`rx_cksum_test` with common testpmd parameters.
+
+        Since checksum errors will be thrown even when the offload is disabled, specify not to
+        invert testing.
+        """
+        self.rx_cksum_test(self.rx_port_for_testing, no_invert=True)
+
+    @requires(
+        NicCapability.PORT_RX_OFFLOAD_UDP_CKSUM,
+        NicCapability.PORT_RX_OFFLOAD_TCP_CKSUM,
+        NicCapability.PORT_RX_OFFLOAD_IPV4_CKSUM,
+    )
+    @func_test
+    def test_rx_cmdline_chksum(self) -> None:
+        """Run :meth:`rx_cksum_test` with checksum offloads enabled through the command-line.
+
+        Since checksum errors will be thrown even when the offload is disabled, specify not to
+        invert testing. Additionally, specify that the offload should not be set at runtime.
+        """
+        self.rx_cksum_test(
+            self.rx_port_for_testing,
+            no_set=True,
+            no_invert=True,
+            enable_rx_cksum=True,
+        )
+
+    @requires(
+        NicCapability.QUEUE_RX_OFFLOAD_UDP_CKSUM,
+        NicCapability.QUEUE_RX_OFFLOAD_TCP_CKSUM,
+        NicCapability.QUEUE_RX_OFFLOAD_IPV4_CKSUM,
+    )
+    @func_test
+    def test_rx_queue_chksum(self) -> None:
+        """Run :meth:`rx_cksum_test` specifying testing should be run on queues.
+
+        Since checksum errors will be thrown even when the offload is disabled, specify not to
+        invert testing.
+        """
+        self.rx_cksum_test(
+            self.rx_port_for_testing,
+            modify_queues=True,
+            no_invert=True,
+        )
+
+    @func_test
+    def test_rx_all_port_offloads(self) -> None:
+        """Verify that testpmd is able to set all Rx offloads the port is capable of at runtime."""
+        with TestPmdShell(self.sut_node) as testpmd:
+            supported_capabilities = testpmd.show_port_rx_offload_capabilities(
+                self.rx_port_for_testing
+            ).per_port
+            try:
+                testpmd.set_port_offload(
+                    self.rx_port_for_testing, True, supported_capabilities, True, verify=True
+                )
+            except InteractiveCommandExecutionError as e:
+                raise TestCaseVerifyError(
+                    f"Failed to set all Rx offloads on port {self.rx_port_for_testing}"
+                ) from e
+
+    @func_test
+    def test_tx_all_port_offloads(self) -> None:
+        """Verify that testpmd is able to set all Tx offloads the port is capable of at runtime."""
+        with TestPmdShell(self.sut_node) as testpmd:
+            supported_capabilities = testpmd.show_port_tx_offload_capabilities(
+                self.tx_port_for_testing
+            ).per_port
+            try:
+                testpmd.set_port_offload(
+                    self.tx_port_for_testing, False, supported_capabilities, True, verify=True
+                )
+            except InteractiveCommandExecutionError as e:
+                raise TestCaseVerifyError(
+                    f"Failed to set all Tx offloads on port {self.tx_port_for_testing}"
+                ) from e
-- 
2.46.0