[PATCH v1 0/2] dts: port over checksum offload suite

[PATCH v1 0/2] dts: port over checksum offload suite

[Dean Marx]

Port over checksum hardware offload testing suite from old DTS. The
suite verifies the ability of the PMD to recognize whether an incoming
packet has valid or invalid L4/IP checksum values.

v1: In the original test plan, there were two Tx checksum test cases. I
removed them due to the lack of consistency in testpmd with Tx checksum
flags, either not being displayed during packet transmission or showing
values that did not align with the original test plan. 

Dean Marx (2):
  dts: add csum HW offload to testpmd shell
  dts: checksum offload test suite

 dts/framework/remote_session/testpmd_shell.py | 124 +++++++++
 dts/tests/TestSuite_checksum_offload.py       | 255 ++++++++++++++++++
 2 files changed, 379 insertions(+)
 create mode 100644 dts/tests/TestSuite_checksum_offload.py

-- 
2.44.0

[PATCH v1 1/2] dts: add csum HW offload to testpmd shell

[Dean Marx]

add csum_set_hw method to testpmd shell class. Port over
set_verbose and port start/stop from queue start/stop suite.

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

diff --git a/dts/framework/remote_session/testpmd_shell.py b/dts/framework/remote_session/testpmd_shell.py
index 43e9f56517..be7cd16b96 100644
--- a/dts/framework/remote_session/testpmd_shell.py
+++ b/dts/framework/remote_session/testpmd_shell.py
@@ -334,6 +334,32 @@ def make_parser(cls) -> ParserFn:
         )
 
 
+class ChecksumOffloadOptions(Flag):
+    """Flag representing checksum hardware offload layer options."""
+
+    #:
+    ip = auto()
+    #:
+    udp = auto()
+    #:
+    tcp = auto()
+    #:
+    sctp = auto()
+    #:
+    outerip = auto()
+    #:
+    outerudp = auto()
+
+    def __str__(self):
+        """String method for use in csum_set_hw."""
+        if self == ChecksumOffloadOptions.outerip:
+            return "outer-ip"
+        elif self == ChecksumOffloadOptions.outerudp:
+            return "outer-udp"
+        else:
+            return f"{self.name}"
+
+
 class DeviceErrorHandlingMode(StrEnum):
     """Enum representing the device error handling mode."""
 
@@ -806,6 +832,104 @@ def show_port_stats(self, port_id: int) -> TestPmdPortStats:
 
         return TestPmdPortStats.parse(output)
 
+    def port_stop(self, port: int, verify: bool = True):
+        """Stop specified port.
+
+        Args:
+            port: Specifies the port number to use, must be between 0-32.
+            verify: If :data:`True`, the output of the command is scanned
+                to ensure specified port is stopped. If not, it is considered
+                an error.
+
+        Raises:
+            InteractiveCommandExecutionError: If `verify` is :data:`True` and the port
+                is not stopped.
+        """
+        port_output = self.send_command(f"port stop {port}")
+        if verify:
+            if "Done" not in port_output:
+                self._logger.debug(f"Failed to stop port {port}: \n{port_output}")
+                raise InteractiveCommandExecutionError(f"Testpmd failed to stop port {port}.")
+
+    def port_start(self, port: int, verify: bool = True):
+        """Start specified port.
+
+        Args:
+            port: Specifies the port number to use, must be between 0-32.
+            verify: If :data:`True`, the output of the command is scanned
+                to ensure specified port is started. If not, it is considered
+                an error.
+
+        Raises:
+            InteractiveCommandExecutionError: If `verify` is :data:`True` and the port
+                is not started.
+        """
+        port_output = self.send_command(f"port start {port}")
+        if verify:
+            if "Done" not in port_output:
+                self._logger.debug(f"Failed to start port {port}: \n{port_output}")
+                raise InteractiveCommandExecutionError(f"Testpmd failed to start port {port}.")
+
+    def csum_set_hw(self, layer: ChecksumOffloadOptions, port_id: int, verify: bool = True) -> None:
+        """Enables hardware checksum offloading on the specified layer.
+
+        Args:
+            layer: The layer that checksum offloading should be enabled on.
+                options: tcp, ip, udp, sctp, outer-ip, outer-udp.
+            port_id: The port number to enable checksum offloading on, should be within 0-32.
+            verify: If :data:`True` the output of the command will be scanned in an attempt to
+                verify that checksum offloading was enabled on the port.
+
+        Raises:
+            InteractiveCommandExecutionError: If checksum offload is not enabled successfully.
+        """
+        csum_output = self.send_command(f"csum set {str(layer)} hw {port_id}")
+        if verify:
+            if "Bad arguments" in csum_output or f"Please stop port {port_id} first" in csum_output:
+                self._logger.debug(f"Failed to set csum hw mode on port {port_id}:\n{csum_output}")
+                raise InteractiveCommandExecutionError(
+                    f"Failed to set csum hw mode on port {port_id}"
+                )
+        if verify and f"checksum offload is not supported by port {port_id}" in csum_output:
+            self._logger.debug(f"Checksum {layer} offload is not supported:\n{csum_output}")
+
+        success = False
+        if layer == ChecksumOffloadOptions.outerip:
+            if "Outer-Ip checksum offload is hw" in csum_output:
+                success = True
+        elif layer == ChecksumOffloadOptions.outerudp:
+            if "Outer-Udp checksum offload is hw" in csum_output:
+                success = True
+        else:
+            if f"{str(layer).upper} checksum offload is hw" in csum_output:
+                success = True
+        if not success and verify:
+            self._logger.debug(f"Failed to set csum hw mode on port {port_id}:\n{csum_output}")
+
+    def set_verbose(self, level: int, verify: bool = True) -> None:
+        """Set debug verbosity level.
+
+        Args:
+            level:
+                0: silent except for error.
+                1: fully verbose except for Tx packets.
+                2: fully verbose except for Rx packets.
+                >2: fully verbose.
+            verify: if :data:`True` an additional command will be sent to verify that verbose level
+                is properly set. Defaults to :data:`True`.
+
+        Raises:
+            InteractiveCommandExecutionError: If `verify` is :data:`True` and verbose level
+            is not correctly set.
+        """
+        verbose_output = self.send_command(f"set verbose {level}")
+        if verify:
+            if "Change verbose level" not in verbose_output:
+                self._logger.debug(f"Failed to set verbose level to {level}: \n{verbose_output}")
+                raise InteractiveCommandExecutionError(
+                    f"Testpmd failed to set verbose level to {level}."
+                )
+
     def _close(self) -> None:
         """Overrides :meth:`~.interactive_shell.close`."""
         self.stop()
-- 
2.44.0

[PATCH v1 2/2] dts: checksum offload test suite

[Dean Marx]

test suite for verifying layer 3/4 checksum offload
features on poll mode driver.

Depends-on: patch-143033
("dts: add text parser for testpmd verbose output")
Depends-on: patch-142691
("dts: add send_packets to test suites and rework packet addressing")

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

diff --git a/dts/tests/TestSuite_checksum_offload.py b/dts/tests/TestSuite_checksum_offload.py
new file mode 100644
index 0000000000..b327261ff3
--- /dev/null
+++ b/dts/tests/TestSuite_checksum_offload.py
@@ -0,0 +1,255 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2024 University of New Hampshire
+
+"""DPDK checksum offload testing suite.
+
+This suite verifies L3/L4 checksum offload features of the Poll Mode Driver.
+On the Rx side, IPv4 and UDP/TCP checksum by hardware is checked to ensure
+checksum flags match expected flags. On the Tx side, IPv4/UDP, IPv4/TCP,
+IPv6/UDP, and IPv6/TCP insertion by hardware is checked to checksum flags
+match expected flags.
+
+"""
+
+from typing import List
+
+from scapy.all import Packet  # type: ignore[import-untyped]
+from scapy.layers.inet import IP, TCP, UDP  # type: ignore[import-untyped]
+from scapy.layers.inet6 import IPv6  # type: ignore[import-untyped]
+from scapy.layers.sctp import SCTP  # type: ignore[import-untyped]
+from scapy.layers.l2 import Dot1Q  # type: ignore[import-untyped]
+from scapy.layers.l2 import Ether
+from scapy.packet import Raw  # type: ignore[import-untyped]
+
+from framework.remote_session.testpmd_shell import (
+    SimpleForwardingModes,
+    TestPmdShell,
+    VerboseOLFlag,
+    ChecksumOffloadOptions
+)
+from framework.test_suite import TestSuite
+
+
+class TestChecksumOffload(TestSuite):
+    """Checksum offload test suite.
+
+    This suite consists of 8 test cases:
+    1. Insert checksum on transmit packet
+    2. Do not insert checksum on transmit packet
+    3. Validate Rx checksum valid flags
+    4. Hardware checksum check L4 Rx
+    5. Hardware checksum check L3 Rx
+    6. Checksum offload with vlan
+
+    """
+
+    def set_up_suite(self) -> None:
+        """Set up the test suite.
+
+        Setup:
+            Verify that at least two port links are created when the
+            test run is initialized.
+        """
+        self.verify(len(self._port_links) > 1, "Not enough port links.")
+
+    def send_packet_and_verify(
+        self, packet_list: List[Packet], load: str, should_receive: bool
+    ) -> None:
+        """Send and verify packet is received on the traffic generator.
+
+        Args:
+            packet_list: list of Scapy packets to send and verify.
+            load: Raw layer load attribute in the sent packet.
+            should_receive: Indicates whether the packet should be received
+                by the traffic generator.
+        """
+        for i in range(0, len(packet_list)):
+            received_packets = self.send_packet_and_capture(packet=packet_list[i])
+            received = any(
+                packet.haslayer(Raw) and load in str(packet.load) for packet in received_packets
+            )
+            self.verify(
+                received == should_receive,
+                f"Packet was {'dropped' if should_receive else 'received'}",
+            )
+
+    def send_packet_and_verify_checksum(
+        self, packet: Packet, goodL4: bool, goodIP: bool, testpmd: TestPmdShell
+    ) -> None:
+        """Send packet and verify verbose output matches expected output.
+
+        Args:
+            packet: Scapy packet to send to DUT.
+            goodL4: Verifies RTE_MBUF_F_RX_L4_CKSUM_GOOD in verbose output
+                if :data:`True`, or RTE_MBUF_F_RX_L4_CKSUM_UNKNOWN if :data:`False`.
+            goodIP: Verifies RTE_MBUF_F_RX_IP_CKSUM_GOOD in verbose output
+                if :data:`True`, or RTE_MBUF_F_RX_IP_CKSUM_UNKNOWN if :data:`False`.
+            testpmd: Testpmd shell session to analyze verbose output of.
+        """
+        testpmd.start()
+        self.send_packet_and_capture(packet=packet)
+        verbose_output = testpmd.extract_verbose_output(testpmd.stop())
+        isL4 = any(
+            VerboseOLFlag.RTE_MBUF_F_RX_L4_CKSUM_GOOD in packet.ol_flags
+            for index in verbose_output
+            for packet in index.packets
+        )
+        isIP = any(
+            VerboseOLFlag.RTE_MBUF_F_RX_IP_CKSUM_GOOD in packet.ol_flags
+            for index in verbose_output
+            for packet in index.packets
+        )
+        self.verify(isL4 == goodL4, "Layer 4 checksum flag did not match expected checksum flag.")
+        self.verify(isIP == goodIP, "IP checksum flag did not match expected checksum flag.")
+
+    def setup_hw_offload(self, testpmd: TestPmdShell) -> None:
+        """Sets IP, UDP, TCP, and SCTP layers to hardware offload."""
+        testpmd.port_stop(port=0)
+        testpmd.csum_set_hw(layer=ChecksumOffloadOptions.ip, port_id=0)
+        testpmd.csum_set_hw(layer=ChecksumOffloadOptions.udp, port_id=0)
+        testpmd.csum_set_hw(layer=ChecksumOffloadOptions.tcp, port_id=0)
+        testpmd.csum_set_hw(layer=ChecksumOffloadOptions.sctp, port_id=0)
+        testpmd.port_start(port=0)
+
+    def test_insert_checksums(self) -> None:
+        """Enable checksum offload insertion and verify packet reception."""
+        payload = "xxxxx"
+        packet_list = [
+            Ether() / IP() / UDP() / Raw(payload),
+            Ether() / IP() / TCP() / Raw(payload),
+            Ether() / IP() / SCTP() / Raw(payload),
+            Ether() / IPv6(src="::1") / UDP() / Raw(payload),
+            Ether() / IPv6(src="::1") / TCP() / Raw(payload),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            testpmd.start()
+            self.send_packet_and_verify(packet_list=packet_list, load=payload, should_receive=True)
+            for i in range(0, len(packet_list)):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd
+                )
+
+    def test_no_insert_checksums(self) -> None:
+        """Enable checksum offload insertion and verify packet reception."""
+        payload = "xxxxx"
+        packet_list = [
+            Ether() / IP() / UDP() / Raw(payload),
+            Ether() / IP() / TCP() / Raw(payload),
+            Ether() / IP() / SCTP() / Raw(payload),
+            Ether() / IPv6(src="::1") / UDP() / Raw(payload),
+            Ether() / IPv6(src="::1") / TCP() / Raw(payload),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            testpmd.start()
+            self.send_packet_and_verify(packet_list=packet_list, load=payload, should_receive=True)
+            for i in range(0, len(packet_list)):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd
+                )
+
+    def test_validate_rx_checksum(self) -> None:
+        """Verify verbose output of Rx packets matches expected behavior."""
+        packet_list = [
+            Ether() / IP() / UDP(),
+            Ether() / IP() / TCP(),
+            Ether() / IP() / SCTP(),
+            Ether() / IPv6(src="::1") / UDP(),
+            Ether() / IPv6(src="::1") / TCP(),
+            Ether() / IP(chksum=0x0) / UDP(chksum=0xF),
+            Ether() / IP(chksum=0x0) / TCP(chksum=0xF),
+            Ether() / IP(chksum=0x0) / SCTP(chksum=0xf),
+            Ether() / IPv6(src="::1") / UDP(chksum=0xF),
+            Ether() / IPv6(src="::1") / TCP(chksum=0xF),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            for i in range(0, 5):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd
+                )
+            for i in range(5, 8):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=False, testpmd=testpmd
+                )
+            for i in range(8, 10):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=True, testpmd=testpmd
+                )
+
+    def test_l4_rx_checksum(self) -> None:
+        """Tests L4 Rx checksum in a variety of scenarios."""
+        packet_list = [
+            Ether() / IP() / UDP(),
+            Ether() / IP() / TCP(),
+            Ether() / IP() / SCTP(),
+            Ether() / IP() / UDP(chksum=0xF),
+            Ether() / IP() / TCP(chksum=0xF),
+            Ether() / IP() / SCTP(chksum=0xf)
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            for i in range(0, 3):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd
+                )
+            for i in range(3, 6):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=True, testpmd=testpmd
+                )
+
+    def test_l3_rx_checksum(self) -> None:
+        """Tests L3 Rx checksum hardware offload."""
+        packet_list = [
+            Ether() / IP() / UDP(),
+            Ether() / IP() / TCP(),
+            Ether() / IP() / SCTP(),
+            Ether() / IP(chksum=0xF) / UDP(),
+            Ether() / IP(chksum=0xF) / TCP(),
+            Ether() / IP(chksum=0xf) / SCTP()
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            for i in range(0, 3):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd
+                )
+            for i in range(3, 6):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=False, testpmd=testpmd
+                )
+
+    def test_vlan_checksum(self) -> None:
+        """Tests VLAN Rx checksum hardware offload and verify packet reception."""
+        payload = "xxxxx"
+        packet_list = [
+            Ether() / Dot1Q(vlan=1) / IP(chksum=0x0) / UDP(chksum=0xF) / Raw(payload),
+            Ether() / Dot1Q(vlan=1) / IP(chksum=0x0) / TCP(chksum=0xF) / Raw(payload),
+            Ether() / Dot1Q(vlan=1) / IP(chksum=0x0) / SCTP(chksum=0x0) / Raw(payload),
+            Ether() / Dot1Q(vlan=1) / IPv6(src="::1") / UDP(chksum=0xF) / Raw(payload),
+            Ether() / Dot1Q(vlan=1) / IPv6(src="::1") / TCP(chksum=0xF) / Raw(payload),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            testpmd.start()
+            self.send_packet_and_verify(packet_list=packet_list, load=payload, should_receive=True)
+            for i in range(0, 3):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=False, testpmd=testpmd
+                )
+            for i in range(3, 5):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=True, testpmd=testpmd
+                )
-- 
2.44.0

Re: [PATCH v1 0/2] dts: port over checksum offload suite

[Alex Chapman]

Hi Dean,

I went over the test suite and it looks really solid, but noticed you 
didn't update the `conf_yaml_schema.json`, is there any reason for this?

On 8/16/24 15:20, Dean Marx wrote:
> Port over checksum hardware offload testing suite from old DTS. The
> suite verifies the ability of the PMD to recognize whether an incoming
> packet has valid or invalid L4/IP checksum values.
> 
> v1: In the original test plan, there were two Tx checksum test cases. I
> removed them due to the lack of consistency in testpmd with Tx checksum
> flags, either not being displayed during packet transmission or showing
> values that did not align with the original test plan.
> 
> Dean Marx (2):
>    dts: add csum HW offload to testpmd shell
>    dts: checksum offload test suite
> 
>   dts/framework/remote_session/testpmd_shell.py | 124 +++++++++
>   dts/tests/TestSuite_checksum_offload.py       | 255 ++++++++++++++++++
>   2 files changed, 379 insertions(+)
>   create mode 100644 dts/tests/TestSuite_checksum_offload.py
> 

Best regards,
Alex

Re: [PATCH v1 2/2] dts: checksum offload test suite

[Alex Chapman]

Hi Dean,

Just though I would point out a few things.

On 8/16/24 15:20, Dean Marx wrote:
<snip>
> +    def send_packet_and_verify(

Should this not be `send_packets_and_verify(` as the argument is
`packet_list`.

<snip>
> +        isL4 = any(
> +            VerboseOLFlag.RTE_MBUF_F_RX_L4_CKSUM_GOOD in packet.ol_flags
> +            for index in verbose_output
> +            for packet in index.packets
> +        )

How does this filter out noise packets with valid checksums?

> +        isIP = any(
> +            VerboseOLFlag.RTE_MBUF_F_RX_IP_CKSUM_GOOD in packet.ol_flags
> +            for index in verbose_output
> +            for packet in index.packets
> +        )

As above

I also noticed there was no implementation of IPv6/SCTP. Although the
old test suite did not include IPv6/SCTP, would it be worth adding it to
the new test suite?
It was included in the test plan, but never implemented for some reason.

Best regards,
Alex

Re: [PATCH v1 2/2] dts: checksum offload test suite

[Dean Marx]

[-- Attachment #1: Type: text/plain, Size: 1633 bytes --]

>
> <snip>
> > +    def send_packet_and_verify(
>
> Should this not be `send_packets_and_verify(` as the argument is
> `packet_list`.
>

Yeah that would definitely make more sense, I'll change that in the next
version.


> <snip>
> > +        isL4 = any(
> > +            VerboseOLFlag.RTE_MBUF_F_RX_L4_CKSUM_GOOD in packet.ol_flags
> > +            for index in verbose_output
> > +            for packet in index.packets
> > +        )
>
> How does this filter out noise packets with valid checksums?
>

It really doesn't right now, I was considering using another verbose output
parameter to verify the packet matches the one that is sent, but I hadn't
included it yet since I wasn't sure which would be best. I feel as though
either length or src_mac would be the easiest to use, but if you or anyone
else has a better idea let me know and I'll incorporate it.


>
> > +        isIP = any(
> > +            VerboseOLFlag.RTE_MBUF_F_RX_IP_CKSUM_GOOD in packet.ol_flags
> > +            for index in verbose_output
> > +            for packet in index.packets
> > +        )
>
>
> I also noticed there was no implementation of IPv6/SCTP. Although the
> old test suite did not include IPv6/SCTP, would it be worth adding it to
> the new test suite?
> It was included in the test plan, but never implemented for some reason.
>
>
I'm assuming you mean SCTP/IPv6 checksums, which I didn't see in any
verbose output on any NIC I had access to so I just left it out. I was
assuming that was the same reason they left it out of the old suite, as
this included the i40e driver, and testpmd only ever displayed IP, L4, and
outer checksums.

[-- Attachment #2: Type: text/html, Size: 2381 bytes --]

Re: [PATCH v1 0/2] dts: port over checksum offload suite

[Dean Marx]

[-- Attachment #1: Type: text/plain, Size: 314 bytes --]

On Mon, Aug 19, 2024 at 10:28 AM Alex Chapman <alex.chapman@arm.com> wrote:

> Hi Dean,
>
> I went over the test suite and it looks really solid, but noticed you
> didn't update the `conf_yaml_schema.json`, is there any reason for this?
>

Good catch, I must've messed up my git commits earlier. Thanks

[-- Attachment #2: Type: text/html, Size: 654 bytes --]

[PATCH v2 0/2] dts: port over checksum offload suite

[Dean Marx]

-------
v2:
* added filter for verbose output using dst mac address

Dean Marx (2):
  dts: add csum HW offload to testpmd shell
  dts: checksum offload test suite

 dts/framework/config/conf_yaml_schema.json    |   3 +-
 dts/framework/remote_session/testpmd_shell.py | 124 ++++++++
 dts/tests/TestSuite_checksum_offload.py       | 264 ++++++++++++++++++
 3 files changed, 390 insertions(+), 1 deletion(-)
 create mode 100644 dts/tests/TestSuite_checksum_offload.py

-- 
2.44.0

[PATCH v2 1/2] dts: add csum HW offload to testpmd shell

[Dean Marx]

add csum_set_hw method to testpmd shell class. Port over
set_verbose and port start/stop from queue start/stop suite.

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

diff --git a/dts/framework/remote_session/testpmd_shell.py b/dts/framework/remote_session/testpmd_shell.py
index 43e9f56517..be7cd16b96 100644
--- a/dts/framework/remote_session/testpmd_shell.py
+++ b/dts/framework/remote_session/testpmd_shell.py
@@ -334,6 +334,32 @@ def make_parser(cls) -> ParserFn:
         )
 
 
+class ChecksumOffloadOptions(Flag):
+    """Flag representing checksum hardware offload layer options."""
+
+    #:
+    ip = auto()
+    #:
+    udp = auto()
+    #:
+    tcp = auto()
+    #:
+    sctp = auto()
+    #:
+    outerip = auto()
+    #:
+    outerudp = auto()
+
+    def __str__(self):
+        """String method for use in csum_set_hw."""
+        if self == ChecksumOffloadOptions.outerip:
+            return "outer-ip"
+        elif self == ChecksumOffloadOptions.outerudp:
+            return "outer-udp"
+        else:
+            return f"{self.name}"
+
+
 class DeviceErrorHandlingMode(StrEnum):
     """Enum representing the device error handling mode."""
 
@@ -806,6 +832,104 @@ def show_port_stats(self, port_id: int) -> TestPmdPortStats:
 
         return TestPmdPortStats.parse(output)
 
+    def port_stop(self, port: int, verify: bool = True):
+        """Stop specified port.
+
+        Args:
+            port: Specifies the port number to use, must be between 0-32.
+            verify: If :data:`True`, the output of the command is scanned
+                to ensure specified port is stopped. If not, it is considered
+                an error.
+
+        Raises:
+            InteractiveCommandExecutionError: If `verify` is :data:`True` and the port
+                is not stopped.
+        """
+        port_output = self.send_command(f"port stop {port}")
+        if verify:
+            if "Done" not in port_output:
+                self._logger.debug(f"Failed to stop port {port}: \n{port_output}")
+                raise InteractiveCommandExecutionError(f"Testpmd failed to stop port {port}.")
+
+    def port_start(self, port: int, verify: bool = True):
+        """Start specified port.
+
+        Args:
+            port: Specifies the port number to use, must be between 0-32.
+            verify: If :data:`True`, the output of the command is scanned
+                to ensure specified port is started. If not, it is considered
+                an error.
+
+        Raises:
+            InteractiveCommandExecutionError: If `verify` is :data:`True` and the port
+                is not started.
+        """
+        port_output = self.send_command(f"port start {port}")
+        if verify:
+            if "Done" not in port_output:
+                self._logger.debug(f"Failed to start port {port}: \n{port_output}")
+                raise InteractiveCommandExecutionError(f"Testpmd failed to start port {port}.")
+
+    def csum_set_hw(self, layer: ChecksumOffloadOptions, port_id: int, verify: bool = True) -> None:
+        """Enables hardware checksum offloading on the specified layer.
+
+        Args:
+            layer: The layer that checksum offloading should be enabled on.
+                options: tcp, ip, udp, sctp, outer-ip, outer-udp.
+            port_id: The port number to enable checksum offloading on, should be within 0-32.
+            verify: If :data:`True` the output of the command will be scanned in an attempt to
+                verify that checksum offloading was enabled on the port.
+
+        Raises:
+            InteractiveCommandExecutionError: If checksum offload is not enabled successfully.
+        """
+        csum_output = self.send_command(f"csum set {str(layer)} hw {port_id}")
+        if verify:
+            if "Bad arguments" in csum_output or f"Please stop port {port_id} first" in csum_output:
+                self._logger.debug(f"Failed to set csum hw mode on port {port_id}:\n{csum_output}")
+                raise InteractiveCommandExecutionError(
+                    f"Failed to set csum hw mode on port {port_id}"
+                )
+        if verify and f"checksum offload is not supported by port {port_id}" in csum_output:
+            self._logger.debug(f"Checksum {layer} offload is not supported:\n{csum_output}")
+
+        success = False
+        if layer == ChecksumOffloadOptions.outerip:
+            if "Outer-Ip checksum offload is hw" in csum_output:
+                success = True
+        elif layer == ChecksumOffloadOptions.outerudp:
+            if "Outer-Udp checksum offload is hw" in csum_output:
+                success = True
+        else:
+            if f"{str(layer).upper} checksum offload is hw" in csum_output:
+                success = True
+        if not success and verify:
+            self._logger.debug(f"Failed to set csum hw mode on port {port_id}:\n{csum_output}")
+
+    def set_verbose(self, level: int, verify: bool = True) -> None:
+        """Set debug verbosity level.
+
+        Args:
+            level:
+                0: silent except for error.
+                1: fully verbose except for Tx packets.
+                2: fully verbose except for Rx packets.
+                >2: fully verbose.
+            verify: if :data:`True` an additional command will be sent to verify that verbose level
+                is properly set. Defaults to :data:`True`.
+
+        Raises:
+            InteractiveCommandExecutionError: If `verify` is :data:`True` and verbose level
+            is not correctly set.
+        """
+        verbose_output = self.send_command(f"set verbose {level}")
+        if verify:
+            if "Change verbose level" not in verbose_output:
+                self._logger.debug(f"Failed to set verbose level to {level}: \n{verbose_output}")
+                raise InteractiveCommandExecutionError(
+                    f"Testpmd failed to set verbose level to {level}."
+                )
+
     def _close(self) -> None:
         """Overrides :meth:`~.interactive_shell.close`."""
         self.stop()
-- 
2.44.0

[PATCH v2 2/2] dts: checksum offload test suite

[Dean Marx]

test suite for verifying layer 3/4 checksum offload
features on poll mode driver.

Depends-on: patch-143033
("dts: add text parser for testpmd verbose output")
Depends-on: patch-142691
("dts: add send_packets to test suites and rework packet addressing")

Signed-off-by: Dean Marx <dmarx@iol.unh.edu>
---
 dts/framework/config/conf_yaml_schema.json |   3 +-
 dts/tests/TestSuite_checksum_offload.py    | 264 +++++++++++++++++++++
 2 files changed, 266 insertions(+), 1 deletion(-)
 create mode 100644 dts/tests/TestSuite_checksum_offload.py

diff --git a/dts/framework/config/conf_yaml_schema.json b/dts/framework/config/conf_yaml_schema.json
index f02a310bb5..a83a6786df 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",
+        "checksum_offload"
       ]
     },
     "test_target": {
diff --git a/dts/tests/TestSuite_checksum_offload.py b/dts/tests/TestSuite_checksum_offload.py
new file mode 100644
index 0000000000..406c9077dc
--- /dev/null
+++ b/dts/tests/TestSuite_checksum_offload.py
@@ -0,0 +1,264 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2024 University of New Hampshire
+
+"""DPDK checksum offload testing suite.
+
+This suite verifies L3/L4 checksum offload features of the Poll Mode Driver.
+On the Rx side, IPv4 and UDP/TCP checksum by hardware is checked to ensure
+checksum flags match expected flags. On the Tx side, IPv4/UDP, IPv4/TCP,
+IPv6/UDP, and IPv6/TCP insertion by hardware is checked to checksum flags
+match expected flags.
+
+"""
+
+from typing import List
+
+from scapy.all import Packet  # type: ignore[import-untyped]
+from scapy.layers.inet import IP, TCP, UDP  # type: ignore[import-untyped]
+from scapy.layers.inet6 import IPv6  # type: ignore[import-untyped]
+from scapy.layers.sctp import SCTP  # type: ignore[import-untyped]
+from scapy.layers.l2 import Dot1Q  # type: ignore[import-untyped]
+from scapy.layers.l2 import Ether
+from scapy.packet import Raw  # type: ignore[import-untyped]
+
+from framework.remote_session.testpmd_shell import (
+    SimpleForwardingModes,
+    TestPmdShell,
+    OLFlag,
+    ChecksumOffloadOptions
+)
+from framework.test_suite import TestSuite
+
+
+class TestChecksumOffload(TestSuite):
+    """Checksum offload test suite.
+
+    This suite consists of 6 test cases:
+    1. Insert checksum on transmit packet
+    2. Do not insert checksum on transmit packet
+    3. Validate Rx checksum valid flags
+    4. Hardware checksum check L4 Rx
+    5. Hardware checksum check L3 Rx
+    6. Checksum offload with vlan
+
+    """
+
+    def set_up_suite(self) -> None:
+        """Set up the test suite.
+
+        Setup:
+            Verify that at least two port links are created when the
+            test run is initialized.
+        """
+        self.verify(len(self._port_links) > 1, "Not enough port links.")
+
+    def send_packets_and_verify(
+        self, packet_list: List[Packet], load: str, should_receive: bool
+    ) -> None:
+        """Send and verify packet is received on the traffic generator.
+
+        Args:
+            packet_list: list of Scapy packets to send and verify.
+            load: Raw layer load attribute in the sent packet.
+            should_receive: Indicates whether the packet should be received
+                by the traffic generator.
+        """
+        for i in range(0, len(packet_list)):
+            received_packets = self.send_packet_and_capture(packet=packet_list[i])
+            received = any(
+                packet.haslayer(Raw) and load in str(packet.load) for packet in received_packets
+            )
+            self.verify(
+                received == should_receive,
+                f"Packet was {'dropped' if should_receive else 'received'}",
+            )
+
+    def send_packet_and_verify_checksum(
+        self, packet: Packet, goodL4: bool, goodIP: bool, testpmd: TestPmdShell
+    ) -> None:
+        """Send packet and verify verbose output matches expected output.
+
+        Args:
+            packet: Scapy packet to send to DUT.
+            goodL4: Verifies RTE_MBUF_F_RX_L4_CKSUM_GOOD in verbose output
+                if :data:`True`, or RTE_MBUF_F_RX_L4_CKSUM_UNKNOWN if :data:`False`.
+            goodIP: Verifies RTE_MBUF_F_RX_IP_CKSUM_GOOD in verbose output
+                if :data:`True`, or RTE_MBUF_F_RX_IP_CKSUM_UNKNOWN if :data:`False`.
+            testpmd: Testpmd shell session to analyze verbose output of.
+        """
+        testpmd.start()
+        self.send_packet_and_capture(packet=packet)
+        verbose_output = testpmd.extract_verbose_output(testpmd.stop())
+        for packet in verbose_output:
+            if packet.dst_mac == "00:00:00:00:00:01":
+                if OLFlag.RTE_MBUF_F_RX_L4_CKSUM_GOOD in packet.ol_flags:
+                    isIP = True
+                else:
+                    isIP = False
+                if OLFlag.RTE_MBUF_F_RX_L4_CKSUM_GOOD in packet.ol_flags:
+                    isL4 = True
+                else:
+                    isL4 = False
+            else:
+                isIP = False
+                isL4 = False
+        self.verify(isL4 == goodL4, "Layer 4 checksum flag did not match expected checksum flag.")
+        self.verify(isIP == goodIP, "IP checksum flag did not match expected checksum flag.")
+
+    def setup_hw_offload(self, testpmd: TestPmdShell) -> None:
+        """Sets IP, UDP, TCP, and SCTP layers to hardware offload."""
+        testpmd.port_stop(port=0)
+        testpmd.csum_set_hw(layer=ChecksumOffloadOptions.ip, port_id=0)
+        testpmd.csum_set_hw(layer=ChecksumOffloadOptions.udp, port_id=0)
+        testpmd.csum_set_hw(layer=ChecksumOffloadOptions.tcp, port_id=0)
+        testpmd.csum_set_hw(layer=ChecksumOffloadOptions.sctp, port_id=0)
+        testpmd.port_start(port=0)
+
+    def test_insert_checksums(self) -> None:
+        """Enable checksum offload insertion and verify packet reception."""
+        payload = "xxxxx"
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / UDP() / Raw(payload),
+            Ether(dst=mac_id) / IP() / TCP() / Raw(payload),
+            Ether(dst=mac_id) / IP() / SCTP() / Raw(payload),
+            Ether(dst=mac_id) / IPv6(src="::1") / UDP() / Raw(payload),
+            Ether(dst=mac_id) / IPv6(src="::1") / TCP() / Raw(payload),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            testpmd.start()
+            self.send_packet_and_verify(packet_list=packet_list, load=payload, should_receive=True)
+            for i in range(0, len(packet_list)):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd
+                )
+
+    def test_no_insert_checksums(self) -> None:
+        """Enable checksum offload insertion and verify packet reception."""
+        payload = "xxxxx"
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / UDP() / Raw(payload),
+            Ether(dst=mac_id) / IP() / TCP() / Raw(payload),
+            Ether(dst=mac_id) / IP() / SCTP() / Raw(payload),
+            Ether(dst=mac_id) / IPv6(src="::1") / UDP() / Raw(payload),
+            Ether(dst=mac_id) / IPv6(src="::1") / TCP() / Raw(payload),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            testpmd.start()
+            self.send_packet_and_verify(packet_list=packet_list, load=payload, should_receive=True)
+            for i in range(0, len(packet_list)):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd
+                )
+
+    def test_validate_rx_checksum(self) -> None:
+        """Verify verbose output of Rx packets matches expected behavior."""
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / UDP(),
+            Ether(dst=mac_id) / IP() / TCP(),
+            Ether(dst=mac_id) / IP() / SCTP(),
+            Ether(dst=mac_id) / IPv6(src="::1") / UDP(),
+            Ether(dst=mac_id) / IPv6(src="::1") / TCP(),
+            Ether(dst=mac_id) / IP(chksum=0x0) / UDP(chksum=0xF),
+            Ether(dst=mac_id) / IP(chksum=0x0) / TCP(chksum=0xF),
+            Ether(dst=mac_id) / IP(chksum=0x0) / SCTP(chksum=0xf),
+            Ether(dst=mac_id) / IPv6(src="::1") / UDP(chksum=0xF),
+            Ether(dst=mac_id) / IPv6(src="::1") / TCP(chksum=0xF),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            for i in range(0, 5):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd
+                )
+            for i in range(5, 8):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=False, testpmd=testpmd
+                )
+            for i in range(8, 10):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=True, testpmd=testpmd
+                )
+
+    def test_l4_rx_checksum(self) -> None:
+        """Tests L4 Rx checksum in a variety of scenarios."""
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / UDP(),
+            Ether(dst=mac_id) / IP() / TCP(),
+            Ether(dst=mac_id) / IP() / SCTP(),
+            Ether(dst=mac_id) / IP() / UDP(chksum=0xF),
+            Ether(dst=mac_id) / IP() / TCP(chksum=0xF),
+            Ether(dst=mac_id) / IP() / SCTP(chksum=0xf)
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            for i in range(0, 3):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd
+                )
+            for i in range(3, 6):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=True, testpmd=testpmd
+                )
+
+    def test_l3_rx_checksum(self) -> None:
+        """Tests L3 Rx checksum hardware offload."""
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / UDP(),
+            Ether(dst=mac_id) / IP() / TCP(),
+            Ether(dst=mac_id) / IP() / SCTP(),
+            Ether(dst=mac_id) / IP(chksum=0xF) / UDP(),
+            Ether(dst=mac_id) / IP(chksum=0xF) / TCP(),
+            Ether(dst=mac_id) / IP(chksum=0xf) / SCTP()
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            for i in range(0, 3):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd
+                )
+            for i in range(3, 6):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=False, testpmd=testpmd
+                )
+
+    def test_vlan_checksum(self) -> None:
+        """Tests VLAN Rx checksum hardware offload and verify packet reception."""
+        payload = "xxxxx"
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id) / Dot1Q(vlan=1) / IP(chksum=0x0) / UDP(chksum=0xF) / Raw(payload),
+            Ether(dst=mac_id) / Dot1Q(vlan=1) / IP(chksum=0x0) / TCP(chksum=0xF) / Raw(payload),
+            Ether(dst=mac_id) / Dot1Q(vlan=1) / IP(chksum=0x0) / SCTP(chksum=0x0) / Raw(payload),
+            Ether(dst=mac_id) / Dot1Q(vlan=1) / IPv6(src="::1") / UDP(chksum=0xF) / Raw(payload),
+            Ether(dst=mac_id) / Dot1Q(vlan=1) / IPv6(src="::1") / TCP(chksum=0xF) / Raw(payload),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            testpmd.start()
+            self.send_packet_and_verify(packet_list=packet_list, load=payload, should_receive=True)
+            for i in range(0, 3):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=False, testpmd=testpmd
+                )
+            for i in range(3, 5):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=True, testpmd=testpmd
+                )
-- 
2.44.0

Re: [PATCH v2 0/2] dts: port over checksum offload suite

[Jeremy Spewock]

Hey Dean,

Thanks for the series! Looks good to me in general, I just left some
comments with a few thoughts I had about some ways to potentially move
some stuff around or swap implementations, but the functionality all
looks good to me. Let me know what you think!

Thanks,
Jeremy


On Wed, Aug 21, 2024 at 12:25 PM Dean Marx <dmarx@iol.unh.edu> wrote:
>
> -------
> v2:
> * added filter for verbose output using dst mac address
>
> Dean Marx (2):
>   dts: add csum HW offload to testpmd shell
>   dts: checksum offload test suite
>
>  dts/framework/config/conf_yaml_schema.json    |   3 +-
>  dts/framework/remote_session/testpmd_shell.py | 124 ++++++++
>  dts/tests/TestSuite_checksum_offload.py       | 264 ++++++++++++++++++
>  3 files changed, 390 insertions(+), 1 deletion(-)
>  create mode 100644 dts/tests/TestSuite_checksum_offload.py
>
> --
> 2.44.0
>

Re: [PATCH v2 1/2] dts: add csum HW offload to testpmd shell

[Jeremy Spewock]

On Wed, Aug 21, 2024 at 12:25 PM Dean Marx <dmarx@iol.unh.edu> wrote:
>
> add csum_set_hw method to testpmd shell class. Port over
> set_verbose and port start/stop from queue start/stop suite.

Since we had that discussion in a DTS meeting about there not really
being a rule against multiple dependencies or anything like that, I
think it might be better if we start moving to just always depending
on other patches rather than duplicating code in between multiple
series'. Not a call out to you at all because I think I have multiple
patches open right now where I also borrow from other suites because I
didn't want long dependency lists, but I think the lists of
dependencies might end up being easier to track than where the code is
from. It also makes for more targeted commit messages.

Let me know what you think though. This might be something worth
talking about with the larger development group as well to get more
opinions on it.

>
> Signed-off-by: Dean Marx <dmarx@iol.unh.edu>
> ---
>  dts/framework/remote_session/testpmd_shell.py | 124 ++++++++++++++++++
>  1 file changed, 124 insertions(+)
>
> diff --git a/dts/framework/remote_session/testpmd_shell.py b/dts/framework/remote_session/testpmd_shell.py
> index 43e9f56517..be7cd16b96 100644
> --- a/dts/framework/remote_session/testpmd_shell.py
> +++ b/dts/framework/remote_session/testpmd_shell.py
> @@ -334,6 +334,32 @@ def make_parser(cls) -> ParserFn:
>          )
>
>
> +class ChecksumOffloadOptions(Flag):
> +    """Flag representing checksum hardware offload layer options."""
> +
> +    #:
> +    ip = auto()
> +    #:
> +    udp = auto()
> +    #:
> +    tcp = auto()
> +    #:
> +    sctp = auto()
> +    #:
> +    outerip = auto()
> +    #:
> +    outerudp = auto()
> +
> +    def __str__(self):
> +        """String method for use in csum_set_hw."""
> +        if self == ChecksumOffloadOptions.outerip:
> +            return "outer-ip"
> +        elif self == ChecksumOffloadOptions.outerudp:
> +            return "outer-udp"

It might be easier to name these values outer_ip and outer_udp and
then just do a str.replace("_", "-") on them to get the same result.

> +        else:
> +            return f"{self.name}"

How does this behave if you have multiple flags combined? Like, for
example, if I had `ChecksumOffloadOptions.ip |
ChecksumOffloadOptions.tcp | ChecksumOffloadOptions.outerip`? I think,
since it is not exactly equal to ChecksumOffloadOptions.outerip, it
wouldn't hit those cases, and I'm not sure exactly what the name
attribute becomes with multiple combined (I know the default string
method does something like "ChecksumOffloadOptions.ip|tcp|outerip").

If you want it to only be one value instead of combinations of values
it is probably better to make this an Enum, but I actually like it
being a Flag for a reason I'll expand on more in the method where this
is used.

> +
> +
>  class DeviceErrorHandlingMode(StrEnum):
>      """Enum representing the device error handling mode."""
>
> @@ -806,6 +832,104 @@ def show_port_stats(self, port_id: int) -> TestPmdPortStats:
>
>          return TestPmdPortStats.parse(output)
>
> +    def port_stop(self, port: int, verify: bool = True):

This method is missing the return-type.

> +        """Stop specified port.
> +
> +        Args:
> +            port: Specifies the port number to use, must be between 0-32.
> +            verify: If :data:`True`, the output of the command is scanned
> +                to ensure specified port is stopped. If not, it is considered
> +                an error.
> +
> +        Raises:
> +            InteractiveCommandExecutionError: If `verify` is :data:`True` and the port
> +                is not stopped.
> +        """
> +        port_output = self.send_command(f"port stop {port}")
> +        if verify:
> +            if "Done" not in port_output:
> +                self._logger.debug(f"Failed to stop port {port}: \n{port_output}")
> +                raise InteractiveCommandExecutionError(f"Testpmd failed to stop port {port}.")
> +
> +    def port_start(self, port: int, verify: bool = True):

This method is also missing the return type annotation.

> +        """Start specified port.
> +
> +        Args:
> +            port: Specifies the port number to use, must be between 0-32.
> +            verify: If :data:`True`, the output of the command is scanned
> +                to ensure specified port is started. If not, it is considered
> +                an error.
> +
> +        Raises:
> +            InteractiveCommandExecutionError: If `verify` is :data:`True` and the port
> +                is not started.
> +        """
> +        port_output = self.send_command(f"port start {port}")
> +        if verify:
> +            if "Done" not in port_output:
> +                self._logger.debug(f"Failed to start port {port}: \n{port_output}")
> +                raise InteractiveCommandExecutionError(f"Testpmd failed to start port {port}.")
> +
> +    def csum_set_hw(self, layer: ChecksumOffloadOptions, port_id: int, verify: bool = True) -> None:
> +        """Enables hardware checksum offloading on the specified layer.
> +
> +        Args:
> +            layer: The layer that checksum offloading should be enabled on.
> +                options: tcp, ip, udp, sctp, outer-ip, outer-udp.

Now that the options are contained in a Flag, it might be better to
not specify the options here and just let the Flag class be the one
that documents it. That way this doc-string doesn't have to be
modified if the class is.

> +            port_id: The port number to enable checksum offloading on, should be within 0-32.
> +            verify: If :data:`True` the output of the command will be scanned in an attempt to
> +                verify that checksum offloading was enabled on the port.
> +
> +        Raises:
> +            InteractiveCommandExecutionError: If checksum offload is not enabled successfully.

I think you may have forgotten to raise the exception in the `if
verify` statement , it looks like it only prints a debug message right
now.

> +        """
> +        csum_output = self.send_command(f"csum set {str(layer)} hw {port_id}")

I think this might also run into some issues if you had a combination
like the example I mentioned above (`ChecksumOffloadOptions.ip |
ChecksumOffloadOptions.tcp | ChecksumOffloadOptions.outerip`).

The reason I think it is a good idea to leave it as a Flag instead of
an Enum however is because it would make it so that if a user wanted
to enable multiple offloads at once they would only have to call this
method one time rather than once for each offload they want to enable.
You wouldn't be able to do that directly using the string version of
the flag like this however, I think the easiest way to do it would be
a for loop like this:

for offload in ChecksumOffloadOptions.__members__:
    if offload in layer:
        csum_output = self.send_command(f"csum set
{offload.name.replace("_", "-")} hw {port_id}")
        if verify:
            ...
            ...
            ...

So basically the same as what you have already, but instead you call
the command for each layer that is specified in the options instead of
trying to convert the Flag to a string. Then someone can call this
method with the example combination I gave above and enable all 3
offloads in one call. I'm not sure if the flags have a way to extract
all of the values that are within an instance of the flag, but that
would also make it easier since you wouldn't have to loop through all
of the ChecksumOffloadOptions.__members__.

> +        if verify:
> +            if "Bad arguments" in csum_output or f"Please stop port {port_id} first" in csum_output:
> +                self._logger.debug(f"Failed to set csum hw mode on port {port_id}:\n{csum_output}")
> +                raise InteractiveCommandExecutionError(
> +                    f"Failed to set csum hw mode on port {port_id}"
> +                )
> +        if verify and f"checksum offload is not supported by port {port_id}" in csum_output:
> +            self._logger.debug(f"Checksum {layer} offload is not supported:\n{csum_output}")
> +
> +        success = False
> +        if layer == ChecksumOffloadOptions.outerip:
> +            if "Outer-Ip checksum offload is hw" in csum_output:
> +                success = True
> +        elif layer == ChecksumOffloadOptions.outerudp:
> +            if "Outer-Udp checksum offload is hw" in csum_output:
> +                success = True
> +        else:
> +            if f"{str(layer).upper} checksum offload is hw" in csum_output:

The call to `upper` here is missing the parenthesis so this might not
be producing the right string.

> +                success = True
> +        if not success and verify:
> +            self._logger.debug(f"Failed to set csum hw mode on port {port_id}:\n{csum_output}")

I think these if-statements would also be simplified if you did it in
a for-loop since all you would have to do is:

if "-" in offload.name and f"{offload.name.title()} offload is hw" not
in csum_output:
   ...
elif f"{offload.name.upper()} offload is hw" not in csum_output:
   ...

I honestly didn't know the `title()` method of a string existed in
python until I just did a little searching and it seems strange to me,
but it would be helpful for this use case. It also is weird to me that
they would have everything other than outer-ip and outer-udp be all
upper case.

> +
> +    def set_verbose(self, level: int, verify: bool = True) -> None:
<snip>


> --
> 2.44.0
>

Re: [PATCH v2 2/2] dts: checksum offload test suite

[Jeremy Spewock]

On Wed, Aug 21, 2024 at 12:25 PM Dean Marx <dmarx@iol.unh.edu> wrote:
>
> test suite for verifying layer 3/4 checksum offload
> features on poll mode driver.
>
> Depends-on: patch-143033
> ("dts: add text parser for testpmd verbose output")
> Depends-on: patch-142691
> ("dts: add send_packets to test suites and rework packet addressing")
>
> Signed-off-by: Dean Marx <dmarx@iol.unh.edu>
> ---
<snip>
> +
> +from typing import List
> +
> +from scapy.all import Packet  # type: ignore[import-untyped]
> +from scapy.layers.inet import IP, TCP, UDP  # type: ignore[import-untyped]
> +from scapy.layers.inet6 import IPv6  # type: ignore[import-untyped]
> +from scapy.layers.sctp import SCTP  # type: ignore[import-untyped]
> +from scapy.layers.l2 import Dot1Q  # type: ignore[import-untyped]
> +from scapy.layers.l2 import Ether

You could probably combine this line with the previous since they are
from the same module.

> +from scapy.packet import Raw  # type: ignore[import-untyped]

I think you can also import `Packet` from this module if you wanted to
combine another two lines as well.

> +
> +from framework.remote_session.testpmd_shell import (
> +    SimpleForwardingModes,

This reminds me of a question I've had for a little while now which
is, should this be imported from the module that it originates from
(params) or is it fine to just grab it from the testpmd shell where it
is also imported? I guess I don't really see this causing a problem at
all since there isn't really a chance of any circular imports in this
case or things that would be breaking, but I just don't know if there
is any kind of guideline regarding these scenarios.

> +    TestPmdShell,
> +    OLFlag,
> +    ChecksumOffloadOptions
> +)
> +from framework.test_suite import TestSuite
> +
> +
> +class TestChecksumOffload(TestSuite):
> +    """Checksum offload test suite.
> +
> +    This suite consists of 6 test cases:
> +    1. Insert checksum on transmit packet
> +    2. Do not insert checksum on transmit packet
> +    3. Validate Rx checksum valid flags
> +    4. Hardware checksum check L4 Rx
> +    5. Hardware checksum check L3 Rx
> +    6. Checksum offload with vlan
> +
> +    """
> +
> +    def set_up_suite(self) -> None:
> +        """Set up the test suite.
> +
> +        Setup:
> +            Verify that at least two port links are created when the
> +            test run is initialized.
> +        """
> +        self.verify(len(self._port_links) > 1, "Not enough port links.")
> +
> +    def send_packets_and_verify(
> +        self, packet_list: List[Packet], load: str, should_receive: bool
> +    ) -> None:
> +        """Send and verify packet is received on the traffic generator.

Now that the name of the method has changed we probably should also
change this subject line of the doc-string to something like "verifies
packets are received..."

> +
<snip>
> +    def send_packet_and_verify_checksum(
> +        self, packet: Packet, goodL4: bool, goodIP: bool, testpmd: TestPmdShell
> +    ) -> None:
> +        """Send packet and verify verbose output matches expected output.
> +
> +        Args:
> +            packet: Scapy packet to send to DUT.
> +            goodL4: Verifies RTE_MBUF_F_RX_L4_CKSUM_GOOD in verbose output
> +                if :data:`True`, or RTE_MBUF_F_RX_L4_CKSUM_UNKNOWN if :data:`False`.
> +            goodIP: Verifies RTE_MBUF_F_RX_IP_CKSUM_GOOD in verbose output
> +                if :data:`True`, or RTE_MBUF_F_RX_IP_CKSUM_UNKNOWN if :data:`False`.
> +            testpmd: Testpmd shell session to analyze verbose output of.
> +        """
> +        testpmd.start()
> +        self.send_packet_and_capture(packet=packet)
> +        verbose_output = testpmd.extract_verbose_output(testpmd.stop())
> +        for packet in verbose_output:
> +            if packet.dst_mac == "00:00:00:00:00:01":

Since this method is the one that relies on this MAC address being set
on the packet, it might be helpful to set that MAC on the packet
before sending it in the same method. Why this address is the one you
were searching for would then be clear at just a glance at the send
method which could be useful. That or you could note in the doc-string
what you expect the MAC to be.

> +                if OLFlag.RTE_MBUF_F_RX_L4_CKSUM_GOOD in packet.ol_flags:
> +                    isIP = True
> +                else:
> +                    isIP = False
> +                if OLFlag.RTE_MBUF_F_RX_L4_CKSUM_GOOD in packet.ol_flags:
> +                    isL4 = True
> +                else:
> +                    isL4 = False
> +            else:
> +                isIP = False
> +                isL4 = False

Would having this else statement break the booleans if there was
another noise packet after the one that you sent in the verbose
output? I think that would make both of these booleans false by the
end of the loop even if one was supposed to be true. You might be able
to fix this however with either a break statement after you find the
first packet with the right MAC address, or you could use the built-in
`any()` function that python offers.

> +        self.verify(isL4 == goodL4, "Layer 4 checksum flag did not match expected checksum flag.")
> +        self.verify(isIP == goodIP, "IP checksum flag did not match expected checksum flag.")
> +
> +    def setup_hw_offload(self, testpmd: TestPmdShell) -> None:
> +        """Sets IP, UDP, TCP, and SCTP layers to hardware offload."""
> +        testpmd.port_stop(port=0)
> +        testpmd.csum_set_hw(layer=ChecksumOffloadOptions.ip, port_id=0)
> +        testpmd.csum_set_hw(layer=ChecksumOffloadOptions.udp, port_id=0)
> +        testpmd.csum_set_hw(layer=ChecksumOffloadOptions.tcp, port_id=0)
> +        testpmd.csum_set_hw(layer=ChecksumOffloadOptions.sctp, port_id=0)
> +        testpmd.port_start(port=0)
> +
> +    def test_insert_checksums(self) -> None:
> +        """Enable checksum offload insertion and verify packet reception."""
> +        payload = "xxxxx"
> +        mac_id = "00:00:00:00:00:01"
> +        packet_list = [
> +            Ether(dst=mac_id) / IP() / UDP() / Raw(payload),
> +            Ether(dst=mac_id) / IP() / TCP() / Raw(payload),
> +            Ether(dst=mac_id) / IP() / SCTP() / Raw(payload),
> +            Ether(dst=mac_id) / IPv6(src="::1") / UDP() / Raw(payload),
> +            Ether(dst=mac_id) / IPv6(src="::1") / TCP() / Raw(payload),
> +        ]
> +        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
> +            testpmd.set_forward_mode(SimpleForwardingModes.csum)
> +            testpmd.set_verbose(level=1)
> +            self.setup_hw_offload(testpmd=testpmd)
> +            testpmd.start()
> +            self.send_packet_and_verify(packet_list=packet_list, load=payload, should_receive=True)

I think the name of this method should also be updated now since
`send_packet_and_verify` was renamed in this version.

> +            for i in range(0, len(packet_list)):
> +                self.send_packet_and_verify_checksum(
> +                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd
> +                )
> +
> +    def test_no_insert_checksums(self) -> None:
> +        """Enable checksum offload insertion and verify packet reception."""

This doc-string is a little confusing to me since the name of the
method implies that there won't be any insertion.

> +        payload = "xxxxx"
> +        mac_id = "00:00:00:00:00:01"
> +        packet_list = [
> +            Ether(dst=mac_id) / IP() / UDP() / Raw(payload),
> +            Ether(dst=mac_id) / IP() / TCP() / Raw(payload),
> +            Ether(dst=mac_id) / IP() / SCTP() / Raw(payload),
> +            Ether(dst=mac_id) / IPv6(src="::1") / UDP() / Raw(payload),
> +            Ether(dst=mac_id) / IPv6(src="::1") / TCP() / Raw(payload),
> +        ]
<snip>
> +
> +    def test_l4_rx_checksum(self) -> None:
> +        """Tests L4 Rx checksum in a variety of scenarios."""
> +        mac_id = "00:00:00:00:00:01"
> +        packet_list = [
> +            Ether(dst=mac_id) / IP() / UDP(),
> +            Ether(dst=mac_id) / IP() / TCP(),
> +            Ether(dst=mac_id) / IP() / SCTP(),
> +            Ether(dst=mac_id) / IP() / UDP(chksum=0xF),
> +            Ether(dst=mac_id) / IP() / TCP(chksum=0xF),
> +            Ether(dst=mac_id) / IP() / SCTP(chksum=0xf)
> +        ]
> +        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
> +            testpmd.set_forward_mode(SimpleForwardingModes.csum)
> +            testpmd.set_verbose(level=1)
> +            self.setup_hw_offload(testpmd=testpmd)
> +            for i in range(0, 3):
> +                self.send_packet_and_verify_checksum(
> +                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd
> +                )
> +            for i in range(3, 6):
> +                self.send_packet_and_verify_checksum(
> +                    packet=packet_list[i], goodL4=False, goodIP=True, testpmd=testpmd
> +                )
> +
> +    def test_l3_rx_checksum(self) -> None:
> +        """Tests L3 Rx checksum hardware offload."""
> +        mac_id = "00:00:00:00:00:01"
> +        packet_list = [
> +            Ether(dst=mac_id) / IP() / UDP(),
> +            Ether(dst=mac_id) / IP() / TCP(),
> +            Ether(dst=mac_id) / IP() / SCTP(),
> +            Ether(dst=mac_id) / IP(chksum=0xF) / UDP(),
> +            Ether(dst=mac_id) / IP(chksum=0xF) / TCP(),
> +            Ether(dst=mac_id) / IP(chksum=0xf) / SCTP()
> +        ]
> +        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
> +            testpmd.set_forward_mode(SimpleForwardingModes.csum)
> +            testpmd.set_verbose(level=1)
> +            self.setup_hw_offload(testpmd=testpmd)
> +            for i in range(0, 3):
> +                self.send_packet_and_verify_checksum(
> +                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd
> +                )
> +            for i in range(3, 6):
> +                self.send_packet_and_verify_checksum(
> +                    packet=packet_list[i], goodL4=True, goodIP=False, testpmd=testpmd
> +                )

Not really important at all (especially because it won't affect the
order that the cases are run in) but it might make sense to put the
two individual tests of l3_rx and l4_rx before the test of both of
them combined. Again, super nit-picky, but it makes sense in my head
to see the individual parts tested and then see them tested together.
I'll leave it up to you if you think it makes sense to just leave it
as is :).

> +
> +    def test_vlan_checksum(self) -> None:
> +        """Tests VLAN Rx checksum hardware offload and verify packet reception."""

What is this testing? Just that the checksums work with VLANs also
set? That's fine if so I just wasn't sure initially since it looks
like the method is checking to see if you can receive the packets and
then if the checksums are right.

> +        payload = "xxxxx"
> +        mac_id = "00:00:00:00:00:01"
> +        packet_list = [
> +            Ether(dst=mac_id) / Dot1Q(vlan=1) / IP(chksum=0x0) / UDP(chksum=0xF) / Raw(payload),
> +            Ether(dst=mac_id) / Dot1Q(vlan=1) / IP(chksum=0x0) / TCP(chksum=0xF) / Raw(payload),
> +            Ether(dst=mac_id) / Dot1Q(vlan=1) / IP(chksum=0x0) / SCTP(chksum=0x0) / Raw(payload),
> +            Ether(dst=mac_id) / Dot1Q(vlan=1) / IPv6(src="::1") / UDP(chksum=0xF) / Raw(payload),
> +            Ether(dst=mac_id) / Dot1Q(vlan=1) / IPv6(src="::1") / TCP(chksum=0xF) / Raw(payload),
> +        ]
> +        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
> +            testpmd.set_forward_mode(SimpleForwardingModes.csum)
> +            testpmd.set_verbose(level=1)
> +            self.setup_hw_offload(testpmd=testpmd)
> +            testpmd.start()
> +            self.send_packet_and_verify(packet_list=packet_list, load=payload, should_receive=True)

I think this call also has to get renamed.



> +            for i in range(0, 3):
> +                self.send_packet_and_verify_checksum(
> +                    packet=packet_list[i], goodL4=False, goodIP=False, testpmd=testpmd
> +                )
> +            for i in range(3, 5):
> +                self.send_packet_and_verify_checksum(
> +                    packet=packet_list[i], goodL4=False, goodIP=True, testpmd=testpmd
> +                )
> --
> 2.44.0
>

Re: [PATCH v2 1/2] dts: add csum HW offload to testpmd shell

[Dean Marx]

[-- Attachment #1: Type: text/plain, Size: 3425 bytes --]

On Fri, Aug 23, 2024 at 10:54 AM Jeremy Spewock <jspewock@iol.unh.edu>
wrote:

> On Wed, Aug 21, 2024 at 12:25 PM Dean Marx <dmarx@iol.unh.edu> wrote:
> >
> > add csum_set_hw method to testpmd shell class. Port over
> > set_verbose and port start/stop from queue start/stop suite.
>
> Since we had that discussion in a DTS meeting about there not really
> being a rule against multiple dependencies or anything like that, I
> think it might be better if we start moving to just always depending
> on other patches rather than duplicating code in between multiple
> series'. Not a call out to you at all because I think I have multiple
> patches open right now where I also borrow from other suites because I
> didn't want long dependency lists, but I think the lists of
> dependencies might end up being easier to track than where the code is
> from. It also makes for more targeted commit messages.
>
> Let me know what you think though. This might be something worth
> talking about with the larger development group as well to get more
> opinions on it.
>
<snip>

I actually like that idea a lot, I'm going to add the dependency and remove
the corresponding methods, especially since it probably makes the
maintainer's jobs easier when there's less code duplication. I can also
send a message in the slack chat about this to see what other people think.


> > +class ChecksumOffloadOptions(Flag):
> > +    """Flag representing checksum hardware offload layer options."""
> > +
> > +    #:
> > +    ip = auto()
> > +    #:
> > +    udp = auto()
> > +    #:
> > +    tcp = auto()
> > +    #:
> > +    sctp = auto()
> > +    #:
> > +    outerip = auto()
> > +    #:
> > +    outerudp = auto()
> > +
> > +    def __str__(self):
> > +        """String method for use in csum_set_hw."""
> > +        if self == ChecksumOffloadOptions.outerip:
> > +            return "outer-ip"
> > +        elif self == ChecksumOffloadOptions.outerudp:
> > +            return "outer-udp"
>
> It might be easier to name these values outer_ip and outer_udp and
> then just do a str.replace("_", "-") on them to get the same result.
>

Makes sense, I ended up just getting rid of the __str__ method entirely and
iterating through the options within the csum set hw method with the
__members__ method you mentioned later in this review. I was able to create
a for loop that looks like this:

for name, offload in ChecksumOffloadOptions.__members__.items():
        if offload in layer:
               (action)

where .items() returns all the flags in a dictionary, where the key is a
string of the flag name and the offload value is the actual flag instance
from the class. This way I could just call name = name.replace("_", "-")
within the loop and use name for send_command and offload for comparing
flags.

<snip>

> I honestly didn't know the `title()` method of a string existed in
> python until I just did a little searching and it seems strange to me,
> but it would be helpful for this use case. It also is weird to me that
> they would have everything other than outer-ip and outer-udp be all
> upper case.


 Yeah it is really odd, I'm not sure if they had consistency in mind while
developing this part of testpmd. The title command is a great idea though,
I added that to the second part of the csum method and it really simplified
everything.

[-- Attachment #2: Type: text/html, Size: 4548 bytes --]

Re: [PATCH v2 2/2] dts: checksum offload test suite

[Dean Marx]

[-- Attachment #1: Type: text/plain, Size: 4794 bytes --]

<snip>

> You could probably combine this line with the previous since they are
> from the same module.
>
> > +from scapy.packet import Raw  # type: ignore[import-untyped]
>
> I think you can also import `Packet` from this module if you wanted to
> combine another two lines as well.
>
>
Wow I didn't even notice that good catch


> > +
> > +from framework.remote_session.testpmd_shell import (
> > +    SimpleForwardingModes,
>
> This reminds me of a question I've had for a little while now which
> is, should this be imported from the module that it originates from
> (params) or is it fine to just grab it from the testpmd shell where it
> is also imported? I guess I don't really see this causing a problem at
> all since there isn't really a chance of any circular imports in this
> case or things that would be breaking, but I just don't know if there
> is any kind of guideline regarding these scenarios.
>

I briefly looked for some best practice guidelines about this kind of thing
but I couldn't find anything explicit. However, I'm going to assume it's
probably preferred to do a direct import like you mentioned so I'm going to
change that.

<snip>

> > +        testpmd.start()
> > +        self.send_packet_and_capture(packet=packet)
> > +        verbose_output = testpmd.extract_verbose_output(testpmd.stop())
> > +        for packet in verbose_output:
> > +            if packet.dst_mac == "00:00:00:00:00:01":
>
> Since this method is the one that relies on this MAC address being set
> on the packet, it might be helpful to set that MAC on the packet
> before sending it in the same method. Why this address is the one you
> were searching for would then be clear at just a glance at the send
> method which could be useful. That or you could note in the doc-string
> what you expect the MAC to be.
>

Funny that you mentioned this because I actually tried to set the
destination mac address within the send_packet_and_verify_checksums method
and I couldn't get it to work no matter what I tried. For some reason even
though the mac address was the one I set after send_packet_and_capture,
none of the packets in verbose_output would have that mac address. I tried
debugging for a while but I just couldn't get it to work, even with the
adjust_addresses patch applied it was breaking so I just removed it
entirely and set them all in the test cases, which worked fine. I did add
an extra arg to the send_and_verify_checksums method called ID, which I
passed the mac_id variable to so that it's cleaner.


>
> > +                if OLFlag.RTE_MBUF_F_RX_L4_CKSUM_GOOD in
> packet.ol_flags:
> > +                    isIP = True
> > +                else:
> > +                    isIP = False
> > +                if OLFlag.RTE_MBUF_F_RX_L4_CKSUM_GOOD in
> packet.ol_flags:
> > +                    isL4 = True
> > +                else:
> > +                    isL4 = False
> > +            else:
> > +                isIP = False
> > +                isL4 = False
>
> Would having this else statement break the booleans if there was
> another noise packet after the one that you sent in the verbose
> output? I think that would make both of these booleans false by the
> end of the loop even if one was supposed to be true. You might be able
> to fix this however with either a break statement after you find the
> first packet with the right MAC address, or you could use the built-in
> `any()` function that python offers.
>

Yeah you're right, not sure what I was thinking there haha. I simplified it
back down to two lines in the new version.

<snip>

>
> Not really important at all (especially because it won't affect the
> order that the cases are run in) but it might make sense to put the
> two individual tests of l3_rx and l4_rx before the test of both of
> them combined. Again, super nit-picky, but it makes sense in my head
> to see the individual parts tested and then see them tested together.
> I'll leave it up to you if you think it makes sense to just leave it
> as is :).
>

Makes sense to me, swapped them


>
> > +
> > +    def test_vlan_checksum(self) -> None:
> > +        """Tests VLAN Rx checksum hardware offload and verify packet
> reception."""
>
> What is this testing? Just that the checksums work with VLANs also
> set? That's fine if so I just wasn't sure initially since it looks
> like the method is checking to see if you can receive the packets and
> then if the checksums are right.
>

Yes it's just testing to ensure checksums work with VLAN packets according
to the test plan. I'm not entirely sure why they also check to make sure
they're received, but it was in the test plan so I just left it in there.

<snip>

Other than that, all the docstring errors and function names were fixed in
both patches. Thanks for the review Jeremy!

[-- Attachment #2: Type: text/html, Size: 6464 bytes --]

[PATCH v3 0/2] dts: port over checksum offload suite

[Dean Marx]

Port over checksum hardware offload testing suite from old DTS. The
suite verifies the ability of the PMD to recognize whether an incoming
packet has valid or invalid L4/IP checksum values.

---------
v1: 
* In the original test plan, there were two Tx checksum test cases. I
removed them due to the lack of consistency in testpmd with Tx checksum
flags, either not being displayed during packet transmission or showing
values that did not align with the original test plan.

v2:
* Added filter for verbose output using dst mac address

v3:
* Refactored csum set hw method to iterate over an instance with
multiple flags
* Fixed docstring errors and method names to be match functionality

Dean Marx (2):
  dts: add csum HW offload to testpmd shell
  dts: checksum offload test suite

 dts/framework/config/conf_yaml_schema.json    |   3 +-
 dts/framework/remote_session/testpmd_shell.py |  51 ++++
 dts/tests/TestSuite_checksum_offload.py       | 257 ++++++++++++++++++
 3 files changed, 310 insertions(+), 1 deletion(-)
 create mode 100644 dts/tests/TestSuite_checksum_offload.py

-- 
2.44.0

[PATCH v3 1/2] dts: add csum HW offload to testpmd shell

[Dean Marx]

add csum_set_hw method to testpmd shell class. Port over
set_verbose and port start/stop from queue start/stop suite.

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

diff --git a/dts/framework/remote_session/testpmd_shell.py b/dts/framework/remote_session/testpmd_shell.py
index 43e9f56517..f0074be9ef 100644
--- a/dts/framework/remote_session/testpmd_shell.py
+++ b/dts/framework/remote_session/testpmd_shell.py
@@ -334,6 +334,23 @@ def make_parser(cls) -> ParserFn:
         )
 
 
+class ChecksumOffloadOptions(Flag):
+    """Flag representing checksum hardware offload layer options."""
+
+    #:
+    ip = auto()
+    #:
+    udp = auto()
+    #:
+    tcp = auto()
+    #:
+    sctp = auto()
+    #:
+    outer_ip = auto()
+    #:
+    outer_udp = auto()
+
+
 class DeviceErrorHandlingMode(StrEnum):
     """Enum representing the device error handling mode."""
 
@@ -806,6 +823,40 @@ def show_port_stats(self, port_id: int) -> TestPmdPortStats:
 
         return TestPmdPortStats.parse(output)
 
+    def csum_set_hw(self, layer: ChecksumOffloadOptions, port_id: int, verify: bool = True) -> None:
+        """Enables hardware checksum offloading on the specified layer.
+
+        Args:
+            layer: The layer that checksum offloading should be enabled on.
+            port_id: The port number to enable checksum offloading on, should be within 0-32.
+            verify: If :data:`True` the output of the command will be scanned in an attempt to
+                verify that checksum offloading was enabled on the port.
+
+        Raises:
+            InteractiveCommandExecutionError: If checksum offload is not enabled successfully.
+        """
+        for name, offload in ChecksumOffloadOptions.__members__.items():
+            if offload in layer:
+                name = name.replace("_", "-")
+                csum_output = self.send_command(f"csum set {name} hw {port_id}")
+            if verify:
+                if ("Bad arguments" in csum_output
+                    or f"Please stop port {port_id} first" in csum_output
+                        or f"checksum offload is not supported by port {port_id}" in csum_output):
+                    self._logger.debug(f"Csum set hw error:\n{csum_output}")
+                    raise InteractiveCommandExecutionError(
+                        f"Failed to set csum hw {name} mode on port {port_id}"
+                    )
+            success = False
+            if "-" in name:
+                name.title()
+            else:
+                name.upper()
+            if f"{name} checksum offload is hw" in csum_output:
+                success = True
+            if not success and verify:
+                self._logger.debug(f"Failed to set csum hw mode on port {port_id}:\n{csum_output}")
+
     def _close(self) -> None:
         """Overrides :meth:`~.interactive_shell.close`."""
         self.stop()
-- 
2.44.0

[PATCH v3 2/2] dts: checksum offload test suite

[Dean Marx]

test suite for verifying layer 3/4 checksum offload
features on poll mode driver.

Depends-on: patch-143033
("dts: add text parser for testpmd verbose output")
Depends-on: patch-142691
("dts: add send_packets to test suites and rework packet addressing")
Depends-on: patch-143005
("dts: add functions to testpmd shell")

Signed-off-by: Dean Marx <dmarx@iol.unh.edu>
---
 dts/framework/config/conf_yaml_schema.json |   3 +-
 dts/tests/TestSuite_checksum_offload.py    | 257 +++++++++++++++++++++
 2 files changed, 259 insertions(+), 1 deletion(-)
 create mode 100644 dts/tests/TestSuite_checksum_offload.py

diff --git a/dts/framework/config/conf_yaml_schema.json b/dts/framework/config/conf_yaml_schema.json
index f02a310bb5..a83a6786df 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",
+        "checksum_offload"
       ]
     },
     "test_target": {
diff --git a/dts/tests/TestSuite_checksum_offload.py b/dts/tests/TestSuite_checksum_offload.py
new file mode 100644
index 0000000000..7467eb5242
--- /dev/null
+++ b/dts/tests/TestSuite_checksum_offload.py
@@ -0,0 +1,257 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2024 University of New Hampshire
+
+"""DPDK checksum offload testing suite.
+
+This suite verifies L3/L4 checksum offload features of the Poll Mode Driver.
+On the Rx side, IPv4 and UDP/TCP checksum by hardware is checked to ensure
+checksum flags match expected flags. On the Tx side, IPv4/UDP, IPv4/TCP,
+IPv6/UDP, and IPv6/TCP insertion by hardware is checked to checksum flags
+match expected flags.
+
+"""
+
+from typing import List
+
+from scapy.layers.inet import IP, TCP, UDP  # type: ignore[import-untyped]
+from scapy.layers.inet6 import IPv6  # type: ignore[import-untyped]
+from scapy.layers.sctp import SCTP  # 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 framework.params.testpmd import SimpleForwardingModes
+from framework.remote_session.testpmd_shell import (
+    TestPmdShell,
+    OLFlag,
+    ChecksumOffloadOptions
+)
+from framework.test_suite import TestSuite
+
+
+class TestChecksumOffload(TestSuite):
+    """Checksum offload test suite.
+
+    This suite consists of 6 test cases:
+    1. Insert checksum on transmit packet
+    2. Do not insert checksum on transmit packet
+    3. Hardware checksum check L4 Rx
+    4. Hardware checksum check L3 Rx
+    5. Validate Rx checksum valid flags
+    6. Checksum offload with vlan
+
+    """
+
+    def set_up_suite(self) -> None:
+        """Set up the test suite.
+
+        Setup:
+            Verify that at least two port links are created when the
+            test run is initialized.
+        """
+        self.verify(len(self._port_links) > 1, "Not enough port links.")
+
+    def send_packets_and_verify(
+        self, packet_list: List[Packet], load: str, should_receive: bool
+    ) -> None:
+        """Iterates through a list of packets and verifies they are received.
+
+        Args:
+            packet_list: List of Scapy packets to send and verify.
+            load: Raw layer load attribute in the sent packet.
+            should_receive: Indicates whether the packet should be received
+                by the traffic generator.
+        """
+        for i in range(0, len(packet_list)):
+            received_packets = self.send_packet_and_capture(packet=packet_list[i])
+            received = any(
+                packet.haslayer(Raw) and load in str(packet.load) for packet in received_packets
+            )
+            self.verify(
+                received == should_receive,
+                f"Packet was {'dropped' if should_receive else 'received'}"
+            )
+
+    def send_packet_and_verify_checksum(
+        self, packet: Packet, goodL4: bool, goodIP: bool, testpmd: TestPmdShell, id: str
+    ) -> None:
+        """Send packet and verify verbose output matches expected output.
+
+        Args:
+            packet: Scapy packet to send to DUT.
+            goodL4: Verifies RTE_MBUF_F_RX_L4_CKSUM_GOOD in verbose output
+                if :data:`True`, or RTE_MBUF_F_RX_L4_CKSUM_UNKNOWN if :data:`False`.
+            goodIP: Verifies RTE_MBUF_F_RX_IP_CKSUM_GOOD in verbose output
+                if :data:`True`, or RTE_MBUF_F_RX_IP_CKSUM_UNKNOWN if :data:`False`.
+            testpmd: Testpmd shell session to analyze verbose output of.
+            id: The destination mac address that matches the sent packet in verbose output.
+        """
+        testpmd.start()
+        self.send_packet_and_capture(packet=packet)
+        verbose_output = testpmd.extract_verbose_output(testpmd.stop())
+        for packet in verbose_output:
+            if packet.dst_mac == id:
+                isIP = OLFlag.RTE_MBUF_F_RX_IP_CKSUM_GOOD in packet.ol_flags
+                isL4 = OLFlag.RTE_MBUF_F_RX_L4_CKSUM_GOOD in packet.ol_flags
+        self.verify(isL4 == goodL4, "Layer 4 checksum flag did not match expected checksum flag.")
+        self.verify(isIP == goodIP, "IP checksum flag did not match expected checksum flag.")
+
+    def setup_hw_offload(self, testpmd: TestPmdShell) -> None:
+        """Sets IP, UDP, TCP, and SCTP layers to hardware offload."""
+        testpmd.port_stop(port=0)
+        offloads = (
+            ChecksumOffloadOptions.ip
+            | ChecksumOffloadOptions.udp
+            | ChecksumOffloadOptions.tcp
+            | ChecksumOffloadOptions.sctp
+        )
+        testpmd.csum_set_hw(layer=offloads, port_id=0)
+        testpmd.port_start(port=0)
+
+    def test_insert_checksums(self) -> None:
+        """Enable checksum offload insertion and verify packet reception."""
+        mac_id = "00:00:00:00:00:01"
+        payload = "xxxxx"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / UDP() / Raw(payload),
+            Ether(dst=mac_id) / IP() / TCP() / Raw(payload),
+            Ether(dst=mac_id) / IP() / SCTP() / Raw(payload),
+            Ether(dst=mac_id) / IPv6(src="::1") / UDP() / Raw(payload),
+            Ether(dst=mac_id) / IPv6(src="::1") / TCP() / Raw(payload),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            testpmd.start()
+            self.send_packets_and_verify(packet_list=packet_list, load=payload, should_receive=True)
+            for i in range(0, len(packet_list)):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd, id=mac_id
+                )
+
+    def test_no_insert_checksums(self) -> None:
+        """Disable checksum offload insertion and verify packet reception."""
+        mac_id = "00:00:00:00:00:01"
+        payload = "xxxxx"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / UDP() / Raw(payload),
+            Ether(dst=mac_id) / IP() / TCP() / Raw(payload),
+            Ether(dst=mac_id) / IP() / SCTP() / Raw(payload),
+            Ether(dst=mac_id) / IPv6(src="::1") / UDP() / Raw(payload),
+            Ether(dst=mac_id) / IPv6(src="::1") / TCP() / Raw(payload),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            testpmd.start()
+            self.send_packets_and_verify(packet_list=packet_list, load=payload, should_receive=True)
+            for i in range(0, len(packet_list)):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd, id=mac_id
+                )
+
+    def test_l4_rx_checksum(self) -> None:
+        """Tests L4 Rx checksum in a variety of scenarios."""
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / UDP(),
+            Ether(dst=mac_id) / IP() / TCP(),
+            Ether(dst=mac_id) / IP() / SCTP(),
+            Ether(dst=mac_id) / IP() / UDP(chksum=0xF),
+            Ether(dst=mac_id) / IP() / TCP(chksum=0xF),
+            Ether(dst=mac_id) / IP() / SCTP(chksum=0xf)
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            for i in range(0, 3):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd, id=mac_id
+                )
+            for i in range(3, 6):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=True, testpmd=testpmd, id=mac_id
+                )
+
+    def test_l3_rx_checksum(self) -> None:
+        """Tests L3 Rx checksum hardware offload."""
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / UDP(),
+            Ether(dst=mac_id) / IP() / TCP(),
+            Ether(dst=mac_id) / IP() / SCTP(),
+            Ether(dst=mac_id) / IP(chksum=0xF) / UDP(),
+            Ether(dst=mac_id) / IP(chksum=0xF) / TCP(),
+            Ether(dst=mac_id) / IP(chksum=0xf) / SCTP()
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            for i in range(0, 3):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd, id=mac_id
+                )
+            for i in range(3, 6):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=False, testpmd=testpmd, id=mac_id
+                )
+
+    def test_validate_rx_checksum(self) -> None:
+        """Verify verbose output of Rx packets matches expected behavior."""
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / UDP(),
+            Ether(dst=mac_id) / IP() / TCP(),
+            Ether(dst=mac_id) / IP() / SCTP(),
+            Ether(dst=mac_id) / IPv6(src="::1") / UDP(),
+            Ether(dst=mac_id) / IPv6(src="::1") / TCP(),
+            Ether(dst=mac_id) / IP(chksum=0x0) / UDP(chksum=0xF),
+            Ether(dst=mac_id) / IP(chksum=0x0) / TCP(chksum=0xF),
+            Ether(dst=mac_id) / IP(chksum=0x0) / SCTP(chksum=0xf),
+            Ether(dst=mac_id) / IPv6(src="::1") / UDP(chksum=0xF),
+            Ether(dst=mac_id) / IPv6(src="::1") / TCP(chksum=0xF)
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            for i in range(0, 5):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd, id=mac_id
+                )
+            for i in range(5, 8):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=False, testpmd=testpmd, id=mac_id
+                )
+            for i in range(8, 10):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=True, testpmd=testpmd, id=mac_id
+                )
+
+    def test_vlan_checksum(self) -> None:
+        """Tests VLAN Rx checksum hardware offload and verify packet reception."""
+        mac_id = "00:00:00:00:00:01"
+        payload = "xxxxx"
+        packet_list = [
+            Ether(dst=mac_id) / Dot1Q(vlan=1) / IP(chksum=0x0) / UDP(chksum=0xF) / Raw(payload),
+            Ether(dst=mac_id) / Dot1Q(vlan=1) / IP(chksum=0x0) / TCP(chksum=0xF) / Raw(payload),
+            Ether(dst=mac_id) / Dot1Q(vlan=1) / IP(chksum=0x0) / SCTP(chksum=0x0) / Raw(payload),
+            Ether(dst=mac_id) / Dot1Q(vlan=1) / IPv6(src="::1") / UDP(chksum=0xF) / Raw(payload),
+            Ether(dst=mac_id) / Dot1Q(vlan=1) / IPv6(src="::1") / TCP(chksum=0xF) / Raw(payload)
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            testpmd.start()
+            self.send_packets_and_verify(packet_list=packet_list, load=payload, should_receive=True)
+            for i in range(0, 3):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=False, testpmd=testpmd, id=mac_id
+                )
+            for i in range(3, 5):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=True, testpmd=testpmd, id=mac_id
+                )
-- 
2.44.0

Re: [PATCH v3 1/2] dts: add csum HW offload to testpmd shell

[Jeremy Spewock]

Hey Dean,

This patch is looking good, I like the changes from the previous
version, there were just a few comments that I left but they should be
pretty easy fixes.

On Mon, Aug 26, 2024 at 5:22 PM Dean Marx <dmarx@iol.unh.edu> wrote:
<snip>
> +    def csum_set_hw(self, layer: ChecksumOffloadOptions, port_id: int, verify: bool = True) -> None:
> +        """Enables hardware checksum offloading on the specified layer.
> +
> +        Args:
> +            layer: The layer that checksum offloading should be enabled on.
> +            port_id: The port number to enable checksum offloading on, should be within 0-32.
> +            verify: If :data:`True` the output of the command will be scanned in an attempt to
> +                verify that checksum offloading was enabled on the port.
> +
> +        Raises:
> +            InteractiveCommandExecutionError: If checksum offload is not enabled successfully.
> +        """
> +        for name, offload in ChecksumOffloadOptions.__members__.items():
> +            if offload in layer:
> +                name = name.replace("_", "-")
> +                csum_output = self.send_command(f"csum set {name} hw {port_id}")
> +            if verify:

I think this line and all of the ones below it in this method need to
be indented one more time so that they are only done if the offload is
one that the user is trying to set. csum_output might not exist if the
previous statement is false, so I don't think we would be able to even
use it in that case, but even if we could I don't think we really need
to verify if the offload is set or not if it isn't in `layer`.

> +                if ("Bad arguments" in csum_output
> +                    or f"Please stop port {port_id} first" in csum_output
> +                        or f"checksum offload is not supported by port {port_id}" in csum_output):
> +                    self._logger.debug(f"Csum set hw error:\n{csum_output}")
> +                    raise InteractiveCommandExecutionError(
> +                        f"Failed to set csum hw {name} mode on port {port_id}"
> +                    )
> +            success = False
> +            if "-" in name:
> +                name.title()

For both this and the call to upper() below, they return a new string
that has the results applied, so `name` here would have to get
re-assigned for the changes to take place. So I think this line would
need to be `name = name.title()`.

> +            else:
> +                name.upper()
> +            if f"{name} checksum offload is hw" in csum_output:

Another thing you could do which I should have thought about sooner is
you could just change this if-statement to be:

if f"{name} checksum offload is hw" in csum_output.lower():

and then you wouldn't need to worry at all about which letters are
uppercase or lowercase in the output and it would all be the same so
you don't have to do this either upper() or title() call. It might be
technically less efficient, but we've mentioned before that we're fine
with that as long as it isn't grossly inefficient, so if it is easier
to just make it all lowercase then I think it is fine :).

> +                success = True
> +            if not success and verify:
> +                self._logger.debug(f"Failed to set csum hw mode on port {port_id}:\n{csum_output}")

Should we also raise the exception here as a failure?


> +
>      def _close(self) -> None:
>          """Overrides :meth:`~.interactive_shell.close`."""
>          self.stop()
> --
> 2.44.0
>

Re: [PATCH v3 2/2] dts: checksum offload test suite

[Jeremy Spewock]

Just one comment here but it is a super small doc-string thing that
still passes the formatting script and everything, but whenever we add
Ruff for checking format in the future, it would get caught so it
might be better to just add it now. Otherwise:

Reviewed-by: Jeremy Spewock <jspewock@iol.unh.edu>

On Mon, Aug 26, 2024 at 5:22 PM Dean Marx <dmarx@iol.unh.edu> wrote:
<snip>
> +    def setup_hw_offload(self, testpmd: TestPmdShell) -> None:
> +        """Sets IP, UDP, TCP, and SCTP layers to hardware offload."""

Since this is a public method and it also has an argument, I think
that it should follow the Google doc-string format and have an Args:
section for it.

> +        testpmd.port_stop(port=0)
> +        offloads = (
> +            ChecksumOffloadOptions.ip
> +            | ChecksumOffloadOptions.udp
> +            | ChecksumOffloadOptions.tcp
> +            | ChecksumOffloadOptions.sctp
> +        )
> +        testpmd.csum_set_hw(layer=offloads, port_id=0)
> +        testpmd.port_start(port=0)
> +
<snip>
>

[PATCH v4 0/2] dts: port over checksum offload suite

[Dean Marx]

Port over checksum hardware offload testing suite from old DTS. The
suite verifies the ability of the PMD to recognize whether an incoming
packet has valid or invalid L4/IP checksum values.

---------
v1: 
* In the original test plan, there were two Tx checksum test cases. I
removed them due to the lack of consistency in testpmd with Tx checksum
flags, either not being displayed during packet transmission or showing
values that did not align with the original test plan.

v2:
* Added filter for verbose output using dst mac address

v3:
* Refactored csum set hw method to iterate over an instance with
multiple flags
* Fixed docstring errors and method names to be match functionality

v4:
* Rebased off next-dts

Dean Marx (2):
  dts: add csum HW offload to testpmd shell
  dts: checksum offload test suite

 dts/framework/config/conf_yaml_schema.json    |   3 +-
 dts/framework/remote_session/testpmd_shell.py |  58 ++++
 dts/tests/TestSuite_checksum_offload.py       | 266 ++++++++++++++++++
 3 files changed, 326 insertions(+), 1 deletion(-)
 create mode 100644 dts/tests/TestSuite_checksum_offload.py

-- 
2.44.0

[PATCH v4 1/2] dts: add csum HW offload to testpmd shell

[Dean Marx]

Add csum_set_hw method to testpmd shell class.

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

diff --git a/dts/framework/remote_session/testpmd_shell.py b/dts/framework/remote_session/testpmd_shell.py
index 16b41a7814..aa1f2a58ca 100644
--- a/dts/framework/remote_session/testpmd_shell.py
+++ b/dts/framework/remote_session/testpmd_shell.py
@@ -126,6 +126,23 @@ def make_parser(cls) -> ParserFn:
         )
 
 
+class ChecksumOffloadOptions(Flag):
+    """Flag representing checksum hardware offload layer options."""
+
+    #:
+    ip = auto()
+    #:
+    udp = auto()
+    #:
+    tcp = auto()
+    #:
+    sctp = auto()
+    #:
+    outer_ip = auto()
+    #:
+    outer_udp = auto()
+
+
 class RSSOffloadTypesFlag(Flag):
     """Flag representing the RSS offload flow types supported by the NIC port."""
 
@@ -1702,6 +1719,47 @@ def show_port_stats(self, port_id: int) -> TestPmdPortStats:
 
         return TestPmdPortStats.parse(output)
 
+    @requires_stopped_ports
+    def csum_set_hw(
+        self, layers: ChecksumOffloadOptions, port_id: int, verify: bool = True
+    ) -> None:
+        """Enables hardware checksum offloading on the specified layer.
+
+        Args:
+            layers: The layer/layers that checksum offloading should be enabled on.
+            port_id: The port number to enable checksum offloading on, should be within 0-32.
+            verify: If :data:`True` the output of the command will be scanned in an attempt to
+                verify that checksum offloading was enabled on the port.
+
+        Raises:
+            InteractiveCommandExecutionError: If checksum offload is not enabled successfully.
+        """
+        for name, offload in ChecksumOffloadOptions.__members__.items():
+            if offload in layers:
+                name = name.replace("_", "-")
+                csum_output = self.send_command(f"csum set {name} hw {port_id}")
+                if verify:
+                    if (
+                        "Bad arguments" in csum_output
+                        or f"Please stop port {port_id} first" in csum_output
+                        or f"checksum offload is not supported by port {port_id}" in csum_output
+                    ):
+                        self._logger.debug(f"Csum set hw error:\n{csum_output}")
+                        raise InteractiveCommandExecutionError(
+                            f"Failed to set csum hw {name} mode on port {port_id}"
+                        )
+                success = False
+                if f"{name} checksum offload is hw" in csum_output.lower():
+                    success = True
+                if not success and verify:
+                    self._logger.debug(
+                        f"Failed to set csum hw mode on port {port_id}:\n{csum_output}"
+                    )
+                    raise InteractiveCommandExecutionError(
+                        f"""Failed to set csum hw mode on port
+                                                           {port_id}:\n{csum_output}"""
+                    )
+
     @requires_stopped_ports
     def set_port_mtu(self, port_id: int, mtu: int, verify: bool = True) -> None:
         """Change the MTU of a port using testpmd.
-- 
2.44.0

[PATCH v4 2/2] dts: checksum offload test suite

[Dean Marx]

test suite for verifying layer 3/4 checksum offload
features on poll mode driver.

Signed-off-by: Dean Marx <dmarx@iol.unh.edu>
---
 dts/framework/config/conf_yaml_schema.json |   3 +-
 dts/tests/TestSuite_checksum_offload.py    | 266 +++++++++++++++++++++
 2 files changed, 268 insertions(+), 1 deletion(-)
 create mode 100644 dts/tests/TestSuite_checksum_offload.py

diff --git a/dts/framework/config/conf_yaml_schema.json b/dts/framework/config/conf_yaml_schema.json
index df390e8ae2..99ae7afe9b 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",
+        "checksum_offload"
       ]
     },
     "test_target": {
diff --git a/dts/tests/TestSuite_checksum_offload.py b/dts/tests/TestSuite_checksum_offload.py
new file mode 100644
index 0000000000..93c931abf8
--- /dev/null
+++ b/dts/tests/TestSuite_checksum_offload.py
@@ -0,0 +1,266 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2024 University of New Hampshire
+
+"""DPDK checksum offload testing suite.
+
+This suite verifies L3/L4 checksum offload features of the Poll Mode Driver.
+On the Rx side, IPv4 and UDP/TCP checksum by hardware is checked to ensure
+checksum flags match expected flags. On the Tx side, IPv4/UDP, IPv4/TCP,
+IPv6/UDP, and IPv6/TCP insertion by hardware is checked to checksum flags
+match expected flags.
+
+"""
+
+from typing import List
+
+from scapy.layers.inet import IP, TCP, UDP  # type: ignore[import-untyped]
+from scapy.layers.inet6 import IPv6  # type: ignore[import-untyped]
+from scapy.layers.l2 import Dot1Q, Ether  # type: ignore[import-untyped]
+from scapy.layers.sctp import SCTP  # type: ignore[import-untyped]
+from scapy.packet import Packet, Raw  # type: ignore[import-untyped]
+
+from framework.params.testpmd import SimpleForwardingModes
+from framework.remote_session.testpmd_shell import (
+    ChecksumOffloadOptions,
+    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_CHECKSUM)
+@requires(NicCapability.RX_OFFLOAD_IPV4_CKSUM)
+@requires(NicCapability.RX_OFFLOAD_UDP_CKSUM)
+@requires(NicCapability.RX_OFFLOAD_TCP_CKSUM)
+@requires(NicCapability.RX_OFFLOAD_SCTP_CKSUM)
+class TestChecksumOffload(TestSuite):
+    """Checksum offload test suite.
+
+    This suite consists of 6 test cases:
+    1. Insert checksum on transmit packet
+    2. Do not insert checksum on transmit packet
+    3. Hardware checksum check L4 Rx
+    4. Hardware checksum check L3 Rx
+    5. Validate Rx checksum valid flags
+    6. Checksum offload with vlan
+
+    """
+
+    def send_packets_and_verify(
+        self, packet_list: List[Packet], load: str, should_receive: bool
+    ) -> None:
+        """Iterates through a list of packets and verifies they are received.
+
+        Args:
+            packet_list: List of Scapy packets to send and verify.
+            load: Raw layer load attribute in the sent packet.
+            should_receive: Indicates whether the packet should be received
+                by the traffic generator.
+        """
+        for i in range(0, len(packet_list)):
+            received_packets = self.send_packet_and_capture(packet=packet_list[i])
+            received = any(
+                packet.haslayer(Raw) and load in str(packet.load) for packet in received_packets
+            )
+            self.verify(
+                received == should_receive,
+                f"Packet was {'dropped' if should_receive else 'received'}",
+            )
+
+    def send_packet_and_verify_checksum(
+        self, packet: Packet, goodL4: bool, goodIP: bool, testpmd: TestPmdShell, id: str
+    ) -> None:
+        """Send packet and verify verbose output matches expected output.
+
+        Args:
+            packet: Scapy packet to send to DUT.
+            goodL4: Verifies RTE_MBUF_F_RX_L4_CKSUM_GOOD in verbose output
+                if :data:`True`, or RTE_MBUF_F_RX_L4_CKSUM_UNKNOWN if :data:`False`.
+            goodIP: Verifies RTE_MBUF_F_RX_IP_CKSUM_GOOD in verbose output
+                if :data:`True`, or RTE_MBUF_F_RX_IP_CKSUM_UNKNOWN if :data:`False`.
+            testpmd: Testpmd shell session to analyze verbose output of.
+            id: The destination mac address that matches the sent packet in verbose output.
+        """
+        testpmd.start()
+        self.send_packet_and_capture(packet=packet)
+        verbose_output = testpmd.extract_verbose_output(testpmd.stop())
+        for packet in verbose_output:
+            if packet.dst_mac == id:
+                isIP = PacketOffloadFlag.RTE_MBUF_F_RX_IP_CKSUM_GOOD in packet.ol_flags
+                isL4 = PacketOffloadFlag.RTE_MBUF_F_RX_L4_CKSUM_GOOD in packet.ol_flags
+        self.verify(isL4 == goodL4, "Layer 4 checksum flag did not match expected checksum flag.")
+        self.verify(isIP == goodIP, "IP checksum flag did not match expected checksum flag.")
+
+    def setup_hw_offload(self, testpmd: TestPmdShell) -> None:
+        """Sets IP, UDP, TCP, and SCTP layers to hardware offload.
+
+        Args:
+            testpmd: Testpmd shell to configure.
+        """
+        testpmd.stop_all_ports()
+        offloads = (
+            ChecksumOffloadOptions.ip
+            | ChecksumOffloadOptions.udp
+            | ChecksumOffloadOptions.tcp
+            | ChecksumOffloadOptions.sctp
+        )
+        testpmd.csum_set_hw(layers=offloads, port_id=0)
+        testpmd.start_all_ports()
+
+    @func_test
+    def test_insert_checksums(self) -> None:
+        """Enable checksum offload insertion and verify packet reception."""
+        mac_id = "00:00:00:00:00:01"
+        payload = "xxxxx"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / UDP() / Raw(payload),
+            Ether(dst=mac_id) / IP() / TCP() / Raw(payload),
+            Ether(dst=mac_id) / IP() / SCTP() / Raw(payload),
+            Ether(dst=mac_id) / IPv6(src="::1") / UDP() / Raw(payload),
+            Ether(dst=mac_id) / IPv6(src="::1") / TCP() / Raw(payload),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            testpmd.start()
+            self.send_packets_and_verify(packet_list=packet_list, load=payload, should_receive=True)
+            for i in range(0, len(packet_list)):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd, id=mac_id
+                )
+
+    @func_test
+    def test_no_insert_checksums(self) -> None:
+        """Disable checksum offload insertion and verify packet reception."""
+        mac_id = "00:00:00:00:00:01"
+        payload = "xxxxx"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / UDP() / Raw(payload),
+            Ether(dst=mac_id) / IP() / TCP() / Raw(payload),
+            Ether(dst=mac_id) / IP() / SCTP() / Raw(payload),
+            Ether(dst=mac_id) / IPv6(src="::1") / UDP() / Raw(payload),
+            Ether(dst=mac_id) / IPv6(src="::1") / TCP() / Raw(payload),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            testpmd.start()
+            self.send_packets_and_verify(packet_list=packet_list, load=payload, should_receive=True)
+            for i in range(0, len(packet_list)):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd, id=mac_id
+                )
+
+    @func_test
+    def test_l4_rx_checksum(self) -> None:
+        """Tests L4 Rx checksum in a variety of scenarios."""
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / UDP(),
+            Ether(dst=mac_id) / IP() / TCP(),
+            Ether(dst=mac_id) / IP() / SCTP(),
+            Ether(dst=mac_id) / IP() / UDP(chksum=0xF),
+            Ether(dst=mac_id) / IP() / TCP(chksum=0xF),
+            Ether(dst=mac_id) / IP() / SCTP(chksum=0xF),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            for i in range(0, 3):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd, id=mac_id
+                )
+            for i in range(3, 6):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=True, testpmd=testpmd, id=mac_id
+                )
+
+    @func_test
+    def test_l3_rx_checksum(self) -> None:
+        """Tests L3 Rx checksum hardware offload."""
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / UDP(),
+            Ether(dst=mac_id) / IP() / TCP(),
+            Ether(dst=mac_id) / IP() / SCTP(),
+            Ether(dst=mac_id) / IP(chksum=0xF) / UDP(),
+            Ether(dst=mac_id) / IP(chksum=0xF) / TCP(),
+            Ether(dst=mac_id) / IP(chksum=0xF) / SCTP(),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            for i in range(0, 3):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd, id=mac_id
+                )
+            for i in range(3, 6):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=False, testpmd=testpmd, id=mac_id
+                )
+
+    @func_test
+    def test_validate_rx_checksum(self) -> None:
+        """Verify verbose output of Rx packets matches expected behavior."""
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / UDP(),
+            Ether(dst=mac_id) / IP() / TCP(),
+            Ether(dst=mac_id) / IP() / SCTP(),
+            Ether(dst=mac_id) / IPv6(src="::1") / UDP(),
+            Ether(dst=mac_id) / IPv6(src="::1") / TCP(),
+            Ether(dst=mac_id) / IP(chksum=0x0) / UDP(chksum=0xF),
+            Ether(dst=mac_id) / IP(chksum=0x0) / TCP(chksum=0xF),
+            Ether(dst=mac_id) / IP(chksum=0x0) / SCTP(chksum=0xF),
+            Ether(dst=mac_id) / IPv6(src="::1") / UDP(chksum=0xF),
+            Ether(dst=mac_id) / IPv6(src="::1") / TCP(chksum=0xF),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            for i in range(0, 5):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd, id=mac_id
+                )
+            for i in range(5, 8):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=False, testpmd=testpmd, id=mac_id
+                )
+            for i in range(8, 10):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=True, testpmd=testpmd, id=mac_id
+                )
+
+    @requires(NicCapability.RX_OFFLOAD_VLAN)
+    @func_test
+    def test_vlan_checksum(self) -> None:
+        """Tests VLAN Rx checksum hardware offload and verify packet reception."""
+        mac_id = "00:00:00:00:00:01"
+        payload = "xxxxx"
+        packet_list = [
+            Ether(dst=mac_id) / Dot1Q(vlan=1) / IP(chksum=0x0) / UDP(chksum=0xF) / Raw(payload),
+            Ether(dst=mac_id) / Dot1Q(vlan=1) / IP(chksum=0x0) / TCP(chksum=0xF) / Raw(payload),
+            Ether(dst=mac_id) / Dot1Q(vlan=1) / IP(chksum=0x0) / SCTP(chksum=0x0) / Raw(payload),
+            Ether(dst=mac_id) / Dot1Q(vlan=1) / IPv6(src="::1") / UDP(chksum=0xF) / Raw(payload),
+            Ether(dst=mac_id) / Dot1Q(vlan=1) / IPv6(src="::1") / TCP(chksum=0xF) / Raw(payload),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            testpmd.start()
+            self.send_packets_and_verify(packet_list=packet_list, load=payload, should_receive=True)
+            for i in range(0, 3):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=False, testpmd=testpmd, id=mac_id
+                )
+            for i in range(3, 5):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=True, testpmd=testpmd, id=mac_id
+                )
-- 
2.44.0

[PATCH v5 0/2] dts: port over checksum offload suite

[Dean Marx]

Port over checksum hardware offload testing suite from old DTS. The
suite verifies the ability of the PMD to recognize whether an incoming
packet has valid or invalid L4/IP checksum values.

---------
v1: 
* In the original test plan, there were two Tx checksum test cases. I
removed them due to the lack of consistency in testpmd with Tx checksum
flags, either not being displayed during packet transmission or showing
values that did not align with the original test plan.

v2:
* Added filter for verbose output using dst mac address

v3:
* Refactored csum set hw method to iterate over an instance with
multiple flags
* Fixed docstring errors and method names to be match functionality

v4:
* Rebased off next-dts

v5:
* Added an SCTP-only test case and removed SCTP packets from other
cases to include a capability check

Dean Marx (2):
  dts: add csum HW offload to testpmd shell
  dts: checksum offload test suite

 dts/framework/config/conf_yaml_schema.json    |   3 +-
 dts/framework/remote_session/testpmd_shell.py |  58 ++++
 dts/tests/TestSuite_checksum_offload.py       | 274 ++++++++++++++++++
 3 files changed, 334 insertions(+), 1 deletion(-)
 create mode 100644 dts/tests/TestSuite_checksum_offload.py

-- 
2.44.0

[PATCH v5 1/2] dts: add csum HW offload to testpmd shell

[Dean Marx]

Add csum_set_hw method to testpmd shell class.

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

diff --git a/dts/framework/remote_session/testpmd_shell.py b/dts/framework/remote_session/testpmd_shell.py
index 16b41a7814..aa1f2a58ca 100644
--- a/dts/framework/remote_session/testpmd_shell.py
+++ b/dts/framework/remote_session/testpmd_shell.py
@@ -126,6 +126,23 @@ def make_parser(cls) -> ParserFn:
         )
 
 
+class ChecksumOffloadOptions(Flag):
+    """Flag representing checksum hardware offload layer options."""
+
+    #:
+    ip = auto()
+    #:
+    udp = auto()
+    #:
+    tcp = auto()
+    #:
+    sctp = auto()
+    #:
+    outer_ip = auto()
+    #:
+    outer_udp = auto()
+
+
 class RSSOffloadTypesFlag(Flag):
     """Flag representing the RSS offload flow types supported by the NIC port."""
 
@@ -1702,6 +1719,47 @@ def show_port_stats(self, port_id: int) -> TestPmdPortStats:
 
         return TestPmdPortStats.parse(output)
 
+    @requires_stopped_ports
+    def csum_set_hw(
+        self, layers: ChecksumOffloadOptions, port_id: int, verify: bool = True
+    ) -> None:
+        """Enables hardware checksum offloading on the specified layer.
+
+        Args:
+            layers: The layer/layers that checksum offloading should be enabled on.
+            port_id: The port number to enable checksum offloading on, should be within 0-32.
+            verify: If :data:`True` the output of the command will be scanned in an attempt to
+                verify that checksum offloading was enabled on the port.
+
+        Raises:
+            InteractiveCommandExecutionError: If checksum offload is not enabled successfully.
+        """
+        for name, offload in ChecksumOffloadOptions.__members__.items():
+            if offload in layers:
+                name = name.replace("_", "-")
+                csum_output = self.send_command(f"csum set {name} hw {port_id}")
+                if verify:
+                    if (
+                        "Bad arguments" in csum_output
+                        or f"Please stop port {port_id} first" in csum_output
+                        or f"checksum offload is not supported by port {port_id}" in csum_output
+                    ):
+                        self._logger.debug(f"Csum set hw error:\n{csum_output}")
+                        raise InteractiveCommandExecutionError(
+                            f"Failed to set csum hw {name} mode on port {port_id}"
+                        )
+                success = False
+                if f"{name} checksum offload is hw" in csum_output.lower():
+                    success = True
+                if not success and verify:
+                    self._logger.debug(
+                        f"Failed to set csum hw mode on port {port_id}:\n{csum_output}"
+                    )
+                    raise InteractiveCommandExecutionError(
+                        f"""Failed to set csum hw mode on port
+                                                           {port_id}:\n{csum_output}"""
+                    )
+
     @requires_stopped_ports
     def set_port_mtu(self, port_id: int, mtu: int, verify: bool = True) -> None:
         """Change the MTU of a port using testpmd.
-- 
2.44.0

[PATCH v5 2/2] dts: checksum offload test suite

[Dean Marx]

test suite for verifying layer 3/4 checksum offload
features on poll mode driver.

Signed-off-by: Dean Marx <dmarx@iol.unh.edu>
---
 dts/framework/config/conf_yaml_schema.json |   3 +-
 dts/tests/TestSuite_checksum_offload.py    | 274 +++++++++++++++++++++
 2 files changed, 276 insertions(+), 1 deletion(-)
 create mode 100644 dts/tests/TestSuite_checksum_offload.py

diff --git a/dts/framework/config/conf_yaml_schema.json b/dts/framework/config/conf_yaml_schema.json
index df390e8ae2..99ae7afe9b 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",
+        "checksum_offload"
       ]
     },
     "test_target": {
diff --git a/dts/tests/TestSuite_checksum_offload.py b/dts/tests/TestSuite_checksum_offload.py
new file mode 100644
index 0000000000..f9aea1655e
--- /dev/null
+++ b/dts/tests/TestSuite_checksum_offload.py
@@ -0,0 +1,274 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2024 University of New Hampshire
+
+"""DPDK checksum offload testing suite.
+
+This suite verifies L3/L4 checksum offload features of the Poll Mode Driver.
+On the Rx side, IPv4 and UDP/TCP checksum by hardware is checked to ensure
+checksum flags match expected flags. On the Tx side, IPv4/UDP, IPv4/TCP,
+IPv6/UDP, and IPv6/TCP insertion by hardware is checked to checksum flags
+match expected flags.
+
+"""
+
+from typing import List
+
+from scapy.layers.inet import IP, TCP, UDP  # type: ignore[import-untyped]
+from scapy.layers.inet6 import IPv6  # type: ignore[import-untyped]
+from scapy.layers.l2 import Dot1Q, Ether  # type: ignore[import-untyped]
+from scapy.layers.sctp import SCTP  # type: ignore[import-untyped]
+from scapy.packet import Packet, Raw  # type: ignore[import-untyped]
+
+from framework.params.testpmd import SimpleForwardingModes
+from framework.remote_session.testpmd_shell import (
+    ChecksumOffloadOptions,
+    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_CHECKSUM)
+@requires(NicCapability.RX_OFFLOAD_IPV4_CKSUM)
+@requires(NicCapability.RX_OFFLOAD_UDP_CKSUM)
+@requires(NicCapability.RX_OFFLOAD_TCP_CKSUM)
+class TestChecksumOffload(TestSuite):
+    """Checksum offload test suite.
+
+    This suite consists of 6 test cases:
+    1. Insert checksum on transmit packet
+    2. Do not insert checksum on transmit packet
+    3. Hardware checksum check L4 Rx
+    4. Hardware checksum check L3 Rx
+    5. Validate Rx checksum valid flags
+    6. Checksum offload with vlan
+
+    """
+
+    def send_packets_and_verify(
+        self, packet_list: List[Packet], load: str, should_receive: bool
+    ) -> None:
+        """Iterates through a list of packets and verifies they are received.
+
+        Args:
+            packet_list: List of Scapy packets to send and verify.
+            load: Raw layer load attribute in the sent packet.
+            should_receive: Indicates whether the packet should be received
+                by the traffic generator.
+        """
+        for i in range(0, len(packet_list)):
+            received_packets = self.send_packet_and_capture(packet=packet_list[i])
+            received = any(
+                packet.haslayer(Raw) and load in str(packet.load) for packet in received_packets
+            )
+            self.verify(
+                received == should_receive,
+                f"Packet was {'dropped' if should_receive else 'received'}",
+            )
+
+    def send_packet_and_verify_checksum(
+        self, packet: Packet, goodL4: bool, goodIP: bool, testpmd: TestPmdShell, id: str
+    ) -> None:
+        """Send packet and verify verbose output matches expected output.
+
+        Args:
+            packet: Scapy packet to send to DUT.
+            goodL4: Verifies RTE_MBUF_F_RX_L4_CKSUM_GOOD in verbose output
+                if :data:`True`, or RTE_MBUF_F_RX_L4_CKSUM_UNKNOWN if :data:`False`.
+            goodIP: Verifies RTE_MBUF_F_RX_IP_CKSUM_GOOD in verbose output
+                if :data:`True`, or RTE_MBUF_F_RX_IP_CKSUM_UNKNOWN if :data:`False`.
+            testpmd: Testpmd shell session to analyze verbose output of.
+            id: The destination mac address that matches the sent packet in verbose output.
+        """
+        testpmd.start()
+        self.send_packet_and_capture(packet=packet)
+        verbose_output = testpmd.extract_verbose_output(testpmd.stop())
+        for packet in verbose_output:
+            if packet.dst_mac == id:
+                isIP = PacketOffloadFlag.RTE_MBUF_F_RX_IP_CKSUM_GOOD in packet.ol_flags
+                isL4 = PacketOffloadFlag.RTE_MBUF_F_RX_L4_CKSUM_GOOD in packet.ol_flags
+        self.verify(isL4 == goodL4, "Layer 4 checksum flag did not match expected checksum flag.")
+        self.verify(isIP == goodIP, "IP checksum flag did not match expected checksum flag.")
+
+    def setup_hw_offload(self, testpmd: TestPmdShell) -> None:
+        """Sets IP, UDP, and TCP layers to hardware offload.
+
+        Args:
+            testpmd: Testpmd shell to configure.
+        """
+        testpmd.stop_all_ports()
+        offloads = (
+            ChecksumOffloadOptions.ip | ChecksumOffloadOptions.udp | ChecksumOffloadOptions.tcp
+        )
+        testpmd.csum_set_hw(layers=offloads, port_id=0)
+        testpmd.start_all_ports()
+
+    @func_test
+    def test_insert_checksums(self) -> None:
+        """Enable checksum offload insertion and verify packet reception."""
+        mac_id = "00:00:00:00:00:01"
+        payload = "xxxxx"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / UDP() / Raw(payload),
+            Ether(dst=mac_id) / IP() / TCP() / Raw(payload),
+            Ether(dst=mac_id) / IPv6(src="::1") / UDP() / Raw(payload),
+            Ether(dst=mac_id) / IPv6(src="::1") / TCP() / Raw(payload),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            testpmd.start()
+            self.send_packets_and_verify(packet_list=packet_list, load=payload, should_receive=True)
+            for i in range(0, len(packet_list)):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd, id=mac_id
+                )
+
+    @func_test
+    def test_no_insert_checksums(self) -> None:
+        """Disable checksum offload insertion and verify packet reception."""
+        mac_id = "00:00:00:00:00:01"
+        payload = "xxxxx"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / UDP() / Raw(payload),
+            Ether(dst=mac_id) / IP() / TCP() / Raw(payload),
+            Ether(dst=mac_id) / IPv6(src="::1") / UDP() / Raw(payload),
+            Ether(dst=mac_id) / IPv6(src="::1") / TCP() / Raw(payload),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            testpmd.start()
+            self.send_packets_and_verify(packet_list=packet_list, load=payload, should_receive=True)
+            for i in range(0, len(packet_list)):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd, id=mac_id
+                )
+
+    @func_test
+    def test_l4_rx_checksum(self) -> None:
+        """Tests L4 Rx checksum in a variety of scenarios."""
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / UDP(),
+            Ether(dst=mac_id) / IP() / TCP(),
+            Ether(dst=mac_id) / IP() / UDP(chksum=0xF),
+            Ether(dst=mac_id) / IP() / TCP(chksum=0xF),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            for i in range(0, 2):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd, id=mac_id
+                )
+            for i in range(2, 4):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=True, testpmd=testpmd, id=mac_id
+                )
+
+    @func_test
+    def test_l3_rx_checksum(self) -> None:
+        """Tests L3 Rx checksum hardware offload."""
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / UDP(),
+            Ether(dst=mac_id) / IP() / TCP(),
+            Ether(dst=mac_id) / IP(chksum=0xF) / UDP(),
+            Ether(dst=mac_id) / IP(chksum=0xF) / TCP(),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            for i in range(0, 2):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd, id=mac_id
+                )
+            for i in range(2, 4):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=False, testpmd=testpmd, id=mac_id
+                )
+
+    @func_test
+    def test_validate_rx_checksum(self) -> None:
+        """Verify verbose output of Rx packets matches expected behavior."""
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / UDP(),
+            Ether(dst=mac_id) / IP() / TCP(),
+            Ether(dst=mac_id) / IPv6(src="::1") / UDP(),
+            Ether(dst=mac_id) / IPv6(src="::1") / TCP(),
+            Ether(dst=mac_id) / IP(chksum=0x0) / UDP(chksum=0xF),
+            Ether(dst=mac_id) / IP(chksum=0x0) / TCP(chksum=0xF),
+            Ether(dst=mac_id) / IPv6(src="::1") / UDP(chksum=0xF),
+            Ether(dst=mac_id) / IPv6(src="::1") / TCP(chksum=0xF),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            for i in range(0, 4):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=True, goodIP=True, testpmd=testpmd, id=mac_id
+                )
+            for i in range(4, 6):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=False, testpmd=testpmd, id=mac_id
+                )
+            for i in range(6, 8):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=True, testpmd=testpmd, id=mac_id
+                )
+
+    @requires(NicCapability.RX_OFFLOAD_VLAN)
+    @func_test
+    def test_vlan_checksum(self) -> None:
+        """Test VLAN Rx checksum hardware offload and verify packet reception."""
+        mac_id = "00:00:00:00:00:01"
+        payload = "xxxxx"
+        packet_list = [
+            Ether(dst=mac_id) / Dot1Q(vlan=1) / IP(chksum=0x0) / UDP(chksum=0xF) / Raw(payload),
+            Ether(dst=mac_id) / Dot1Q(vlan=1) / IP(chksum=0x0) / TCP(chksum=0xF) / Raw(payload),
+            Ether(dst=mac_id) / Dot1Q(vlan=1) / IPv6(src="::1") / UDP(chksum=0xF) / Raw(payload),
+            Ether(dst=mac_id) / Dot1Q(vlan=1) / IPv6(src="::1") / TCP(chksum=0xF) / Raw(payload),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            self.setup_hw_offload(testpmd=testpmd)
+            testpmd.start()
+            self.send_packets_and_verify(packet_list=packet_list, load=payload, should_receive=True)
+            for i in range(0, 2):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=False, testpmd=testpmd, id=mac_id
+                )
+            for i in range(2, 4):
+                self.send_packet_and_verify_checksum(
+                    packet=packet_list[i], goodL4=False, goodIP=True, testpmd=testpmd, id=mac_id
+                )
+
+    @requires(NicCapability.RX_OFFLOAD_SCTP_CKSUM)
+    @func_test
+    def test_validate_sctp_checksum(self) -> None:
+        """Test SCTP Rx checksum hardware offload and verify packet reception."""
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / SCTP(),
+            Ether(dst=mac_id) / IP() / SCTP(chksum=0xF),
+        ]
+        with TestPmdShell(node=self.sut_node, enable_rx_cksum=True) as testpmd:
+            testpmd.set_forward_mode(SimpleForwardingModes.csum)
+            testpmd.set_verbose(level=1)
+            testpmd.csum_set_hw(layers=ChecksumOffloadOptions.sctp)
+            testpmd.start()
+            self.send_packet_and_verify_checksum(
+                packet=packet_list[0], goodL4=True, goodIP=True, testpmd=testpmd, id=mac_id
+            )
+            self.send_packet_and_verify_checksum(
+                packet=packet_list[1], goodL4=False, goodIP=True, testpmd=testpmd, id=mac_id
+            )
-- 
2.44.0

Re: [PATCH v5 1/2] dts: add csum HW offload to testpmd shell

[Patrick Robb]

[-- Attachment #1: Type: text/plain, Size: 46 bytes --]

Reviewed-by: Patrick Robb <probb@iol.unh.edu>

[-- Attachment #2: Type: text/html, Size: 112 bytes --]

Re: [PATCH v5 2/2] dts: checksum offload test suite

[Patrick Robb]

[-- Attachment #1: Type: text/plain, Size: 574 bytes --]

On Tue, Oct 15, 2024 at 3:13 PM Dean Marx <dmarx@iol.unh.edu> wrote:

>
> +
> +    This suite consists of 6 test cases:
> +    1. Insert checksum on transmit packet
> +    2. Do not insert checksum on transmit packet
> +    3. Hardware checksum check L4 Rx
> +    4. Hardware checksum check L3 Rx
> +    5. Validate Rx checksum valid flags
> +    6. Checksum offload with vlan
>

In your V5 you forgot to add the SCTP offload testcase to the list above. I
will update the commit before applying to next-dts.

Reviewed-by: Patrick Robb <probb@iol.unh.edu>

[-- Attachment #2: Type: text/html, Size: 1005 bytes --]

Re: [PATCH v5 2/2] dts: checksum offload test suite

[Patrick Robb]

[-- Attachment #1: Type: text/plain, Size: 29 bytes --]

Applied to next-dts, thanks.

[-- Attachment #2: Type: text/html, Size: 50 bytes --]

Re: [PATCH v5 1/2] dts: add csum HW offload to testpmd shell

[Patrick Robb]

[-- Attachment #1: Type: text/plain, Size: 29 bytes --]

Applied to next-dts, thanks.

[-- Attachment #2: Type: text/html, Size: 50 bytes --]

Re: [PATCH v1 1/2] dts: add csum HW offload to testpmd shell

[Nicholas Pratte]

On Mon, Aug 12, 2024 at 9:39 AM Dean Marx <dmarx@iol.unh.edu> wrote:
>
> add csum_set_hw method to testpmd shell class. Port over
> set_verbose and port start/stop from queue start/stop suite.
>
> Signed-off-by: Dean Marx <dmarx@iol.unh.edu>
> ---
>  dts/framework/remote_session/testpmd_shell.py | 94 +++++++++++++++++++
>  1 file changed, 94 insertions(+)
>
> diff --git a/dts/framework/remote_session/testpmd_shell.py b/dts/framework/remote_session/testpmd_shell.py
> index 43e9f56517..be8fbdc295 100644
> --- a/dts/framework/remote_session/testpmd_shell.py
> +++ b/dts/framework/remote_session/testpmd_shell.py
> @@ -806,6 +806,100 @@ def show_port_stats(self, port_id: int) -> TestPmdPortStats:
>
>          return TestPmdPortStats.parse(output)
>
> +    def port_stop(self, port: int, verify: bool = True):
> +        """Stop specified port.
> +
> +        Args:
> +            port: Specifies the port number to use, must be between 0-32.
> +            verify: If :data:`True`, the output of the command is scanned
> +                to ensure specified port is stopped. If not, it is considered
> +                an error.
> +
> +        Raises:
> +            InteractiveCommandExecutionError: If `verify` is :data:`True` and the port
> +                is not stopped."""
> +        port_output = self.send_command(f"port stop {port}")
> +        if verify:
> +            if "Done" not in port_output:
> +                self._logger.debug(f"Failed to stop port {port}: \n{port_output}")
> +                raise InteractiveCommandExecutionError(f"Testpmd failed to stop port {port}.")
> +
> +    def port_start(self, port: int, verify: bool = True):
> +        """Start specified port.
> +
> +        Args:
> +            port: Specifies the port number to use, must be between 0-32.
> +            verify: If :data:`True`, the output of the command is scanned
> +                to ensure specified port is started. If not, it is considered
> +                an error.
> +
> +        Raises:
> +            InteractiveCommandExecutionError: If `verify` is :data:`True` and the port
> +                is not started."""
> +        port_output = self.send_command(f"port start {port}")
> +        if verify:
> +            if "Done" not in port_output:
> +                self._logger.debug(f"Failed to start port {port}: \n{port_output}")
> +                raise InteractiveCommandExecutionError(f"Testpmd failed to start port {port}.")
> +
> +    def csum_set_hw(self, layer: str, port_id: int, verify: bool = True) -> None:
> +        """Enables hardware checksum offloading on the specified layer.
> +
> +        Args:
> +            layer: The layer that checksum offloading should be enabled on.
> +                options: tcp, ip, udp, sctp, outer-ip, outer-udp.
> +            port_id: The port number to enable checksum offloading on, should be within 0-32.
> +            verify: If :data:`True` the output of the command will be scanned in an attempt to
> +                verify that checksum offloading was enabled on the port.

I ran into a similar situation with the VLAN offload set command which
requires a string input much like that of the 'layer' parameter here.
While this would technically work, it might be best to create some
kind of type class (or even Flags if at all possible) so that users
can select from a more rigorous list of options when writing test
suites.

For example, instead of passing a string into the method, you have the
user pass in something like csum_set_hw(LayerTypes.ip, port_id = 0)
where 'LayerTypes' is some kind of Flag or user-defined type class.
It's really just a preference, but I do think it makes for more
concise code.
> +
> +        Raises:
> +            InteractiveCommandExecutionError: If checksum offload is not enabled successfully.
> +        """
> +        csum_output = self.send_command(f"csum set {layer} hw {port_id}")
> +        if (verify and ("Bad arguments" in csum_output or f"Please stop port {port_id} first")):
> +            self._logger.debug(f"Failed to set csum hw mode on port {port_id}:\n{csum_output}")
> +            raise InteractiveCommandExecutionError(
> +                f"Failed to set csum hw mode on port {port_id}"
> +                )
> +        if verify and f"checksum offload is not supported by port {port_id}" in csum_output:
> +            self._logger.debug(f"Checksum {layer} offload is not supported by port {port_id}:\n{csum_output}")
> +
> +        success = False
> +        if "outer-ip" in layer:
> +            if "Outer-Ip checksum offload is hw" in csum_output:
> +                success = True
> +        if "outer-udp" in layer:
> +            if "Outer-Udp checksum offload is hw" in csum_output:
> +                success = True
> +        else:
> +            if f"{layer.upper} checksum offload is hw" in csum_output:
> +                success = True
> +        if not success and verify:
> +            self._logger.debug(f"Failed to set csum hw mode on port {port_id}:\n{csum_output}")
> +
> +    def set_verbose(self, level: int, verify: bool = True) -> None:
> +        """Set debug verbosity level.
> +
> +        Args:
> +            level: 0 - silent except for error
> +                1 - fully verbose except for Tx packets
> +                2 - fully verbose except for Rx packets
> +                >2 - fully verbose
> +            verify: if :data:`True` an additional command will be sent to verify that verbose level
> +                is properly set. Defaults to :data:`True`.
> +
> +        Raises:
> +            InteractiveCommandExecutionError: If `verify` is :data:`True` and verbose level
> +            is not correctly set.
> +        """
> +        verbose_output = self.send_command(f"set verbose {level}")
> +        if verify:
> +            if "Change verbose level" not in verbose_output:
> +                self._logger.debug(f"Failed to set verbose level to {level}: \n{verbose_output}")
> +                raise InteractiveCommandExecutionError(
> +                    f"Testpmd failed to set verbose level to {level}."
> +                )
> +
>      def _close(self) -> None:
>          """Overrides :meth:`~.interactive_shell.close`."""
>          self.stop()
> --
> 2.44.0
>

[PATCH v1 1/2] dts: add csum HW offload to testpmd shell

[Dean Marx]

add csum_set_hw method to testpmd shell class. Port over
set_verbose and port start/stop from queue start/stop suite.

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

diff --git a/dts/framework/remote_session/testpmd_shell.py b/dts/framework/remote_session/testpmd_shell.py
index 43e9f56517..be8fbdc295 100644
--- a/dts/framework/remote_session/testpmd_shell.py
+++ b/dts/framework/remote_session/testpmd_shell.py
@@ -806,6 +806,100 @@ def show_port_stats(self, port_id: int) -> TestPmdPortStats:
 
         return TestPmdPortStats.parse(output)
 
+    def port_stop(self, port: int, verify: bool = True):
+        """Stop specified port.
+
+        Args:
+            port: Specifies the port number to use, must be between 0-32.
+            verify: If :data:`True`, the output of the command is scanned
+                to ensure specified port is stopped. If not, it is considered
+                an error.
+
+        Raises:
+            InteractiveCommandExecutionError: If `verify` is :data:`True` and the port
+                is not stopped."""
+        port_output = self.send_command(f"port stop {port}")
+        if verify:
+            if "Done" not in port_output:
+                self._logger.debug(f"Failed to stop port {port}: \n{port_output}")
+                raise InteractiveCommandExecutionError(f"Testpmd failed to stop port {port}.")
+
+    def port_start(self, port: int, verify: bool = True):
+        """Start specified port.
+
+        Args:
+            port: Specifies the port number to use, must be between 0-32.
+            verify: If :data:`True`, the output of the command is scanned
+                to ensure specified port is started. If not, it is considered
+                an error.
+
+        Raises:
+            InteractiveCommandExecutionError: If `verify` is :data:`True` and the port
+                is not started."""
+        port_output = self.send_command(f"port start {port}")
+        if verify:
+            if "Done" not in port_output:
+                self._logger.debug(f"Failed to start port {port}: \n{port_output}")
+                raise InteractiveCommandExecutionError(f"Testpmd failed to start port {port}.")
+
+    def csum_set_hw(self, layer: str, port_id: int, verify: bool = True) -> None:
+        """Enables hardware checksum offloading on the specified layer.
+
+        Args:
+            layer: The layer that checksum offloading should be enabled on.
+                options: tcp, ip, udp, sctp, outer-ip, outer-udp.
+            port_id: The port number to enable checksum offloading on, should be within 0-32.
+            verify: If :data:`True` the output of the command will be scanned in an attempt to
+                verify that checksum offloading was enabled on the port.
+
+        Raises:
+            InteractiveCommandExecutionError: If checksum offload is not enabled successfully.
+        """
+        csum_output = self.send_command(f"csum set {layer} hw {port_id}")
+        if (verify and ("Bad arguments" in csum_output or f"Please stop port {port_id} first")):
+            self._logger.debug(f"Failed to set csum hw mode on port {port_id}:\n{csum_output}")
+            raise InteractiveCommandExecutionError(
+                f"Failed to set csum hw mode on port {port_id}"
+                )
+        if verify and f"checksum offload is not supported by port {port_id}" in csum_output:
+            self._logger.debug(f"Checksum {layer} offload is not supported by port {port_id}:\n{csum_output}")
+
+        success = False
+        if "outer-ip" in layer:
+            if "Outer-Ip checksum offload is hw" in csum_output:
+                success = True
+        if "outer-udp" in layer:
+            if "Outer-Udp checksum offload is hw" in csum_output:
+                success = True
+        else:
+            if f"{layer.upper} checksum offload is hw" in csum_output:
+                success = True
+        if not success and verify:
+            self._logger.debug(f"Failed to set csum hw mode on port {port_id}:\n{csum_output}")
+
+    def set_verbose(self, level: int, verify: bool = True) -> None:
+        """Set debug verbosity level.
+
+        Args:
+            level: 0 - silent except for error
+                1 - fully verbose except for Tx packets
+                2 - fully verbose except for Rx packets
+                >2 - fully verbose
+            verify: if :data:`True` an additional command will be sent to verify that verbose level
+                is properly set. Defaults to :data:`True`.
+
+        Raises:
+            InteractiveCommandExecutionError: If `verify` is :data:`True` and verbose level
+            is not correctly set.
+        """
+        verbose_output = self.send_command(f"set verbose {level}")
+        if verify:
+            if "Change verbose level" not in verbose_output:
+                self._logger.debug(f"Failed to set verbose level to {level}: \n{verbose_output}")
+                raise InteractiveCommandExecutionError(
+                    f"Testpmd failed to set verbose level to {level}."
+                )
+
     def _close(self) -> None:
         """Overrides :meth:`~.interactive_shell.close`."""
         self.stop()
-- 
2.44.0