@@ -187,7 +187,8 @@
"enum": [
"hello_world",
"os_udp",
- "pmd_buffer_scatter"
+ "pmd_buffer_scatter",
+ "rxtx_offload"
]
},
"test_target": {
new file mode 100644
@@ -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