|SUCCESS| [v1] dts: add ipgre test suite

[dpdklab at iol.unh.edu]

Test-Label: ai-code-review
Test-Status: SUCCESS
https://dpdk.org/patch/162761

_AI Code Review_

=== Patch Review: series.patch (via Claude) ===
Review date: 2026-03-29

I'll review this DPDK patch against the coding standards and contribution guidelines.

---

**Commit:** (from patch header, git sha not available in patch format)
**Author:** Andrew Bailey <abailey at iol.unh.edu>
**Subject:** dts: add ipgre test suite

This patch ports the IP GRE test suite from old DTS to next DTS,
covering GRE tunneling and checksum offload verification.

> diff --git a/dts/api/testpmd/__init__.py b/dts/api/testpmd/__init__.py
> index e9187440bb..33bf5e63de 100644
> --- a/dts/api/testpmd/__init__.py
> +++ b/dts/api/testpmd/__init__.py
> @@ -951,6 +951,29 @@ def set_flow_control(
>                      f"Testpmd failed to set the {flow_ctrl} in port {port}."
>                  )
>  
> +    def set_csum_parse_tunnel(self, port: int, on: bool, verify: bool = True) -> None:
> +        """Set parse tunnel on or of in testpmd for a given port.

Typo: Should be "on or off" not "on or of".

> +
> +        Args:
> +            port: The ID of the requested port
> +            on: set parse tunnel on if `on` is :data:`True`, otherwise off
> +            verify: if :data:`True`, the output of the command is scanned to verify that
> +                parse tunnel was set successfully
> +
> +        Raises:
> +            InteractiveCommandExecutionError: If `verify` is :data:`True` and the command
> +                fails to execute.
> +
> +        """
> +        output = self.send_command(f"csum parse-tunnel {"on" if on else "off"} {port}")
> +        if verify and f"Parse tunnel is {'on' if on else'off'}" not in output:

In the string verification, is there a missing space between "else" and
"'off'"? The format string appears to be "{'on' if on else'off'}" which
would produce "Parse tunnel isoff" when on is False. Should this be
"{'on' if on else 'off'}"?

> +            self._logger.debug(
> +                f"Testpmd failed to set csum parse-tunnel {'on' if on else 'off'} in port {port}"
> +            )
> +            raise InteractiveCommandExecutionError(
> +                f"Testpmd failed to set csum parse-tunnel {'on' if on else 'off'} in port {port}"
> +            )

[ ... ]

> diff --git a/dts/tests/TestSuite_ip_gre.py b/dts/tests/TestSuite_ip_gre.py
> new file mode 100644
> index 0000000000..fc51eef181
> --- /dev/null
> +++ b/dts/tests/TestSuite_ip_gre.py
> @@ -0,0 +1,301 @@

[ ... ]

> +    def _check_for_matching_packet(
> +        self, output: list[TestPmdVerbosePacket], flags: RtePTypes
> +    ) -> bool:
> +        """Returns :data:`True` if the packet in verbose output contains all specified flags."""
> +        for packet in output:
> +            if packet.src_mac == SRC_ID:
> +                if flags not in packet.hw_ptype and flags not in packet.sw_ptype:
> +                    return False
> +        return True

Is the logic in _check_for_matching_packet()->loop correct? When the
loop finds a packet matching SRC_ID but the flags don't match, it
returns False immediately. However, if no packet with SRC_ID is found,
or if we iterate through all non-matching packets, the function returns
True. Does this cause false positives when the test packet is dropped?

> +
> +    def _send_packet_and_verify_flags(
> +        self, expected_flag: RtePTypes, packet: Packet, testpmd: TestPmd
> +    ) -> None:
> +        """Sends a packet to the DUT and verifies the verbose ptype flags."""
> +        send_packet_and_capture(packet=packet)
> +        verbose_output = testpmd.extract_verbose_output(testpmd.stop(verify=True))
> +        valid = self._check_for_matching_packet(output=verbose_output, flags=expected_flag)
> +        verify(valid, f"Packet type flag did not match the expected flag: {expected_flag}.")
> +
> +    def _setup_session(
> +        self, testpmd: TestPmd, expected_flags: list[RtePTypes], packet_list=list[Packet]

Does the default argument packet_list=list[Packet] create a mutable
default that persists across calls? Should this be
packet_list: list[Packet] without a default value?

> +    ) -> None:
> +        """Sets the forwarding and verbose mode of each test case interactive shell session."""
> +        testpmd.set_forward_mode(SimpleForwardingModes.rxonly)
> +        testpmd.set_verbose(level=1)
> +        for i in range(0, len(packet_list)):
> +            testpmd.start(verify=True)
> +            self._send_packet_and_verify_flags(
> +                expected_flag=expected_flags[i], packet=packet_list[i], testpmd=testpmd
> +            )

Does this code handle mismatched list lengths between expected_flags and
packet_list? Would an IndexError occur if len(expected_flags) <
len(packet_list)?

> +
> +    def _send_packet_and_verify_checksum(
> +        self, packet: Packet, good_L4: bool, good_IP: bool, testpmd: TestPmd
> +    ) -> None:
> +        """Send packet and verify verbose output matches expected output."""
> +        testpmd.start()
> +        send_packet_and_capture(packet=packet)
> +        verbose_output = testpmd.extract_verbose_output(testpmd.stop())
> +        is_IP = is_L4 = None
> +        for testpmd_packet in verbose_output:
> +            if testpmd_packet.src_mac == SRC_ID:
> +                is_IP = PacketOffloadFlag.RTE_MBUF_F_RX_IP_CKSUM_GOOD in testpmd_packet.ol_flags
> +                is_L4 = PacketOffloadFlag.RTE_MBUF_F_RX_L4_CKSUM_GOOD in testpmd_packet.ol_flags

Does this loop continue after finding the matching packet? Would
subsequent packets with SRC_ID overwrite is_IP and is_L4?

> +        verify(
> +            is_IP is not None and is_L4 is not None,
> +            "Test packet was dropped when it should have been received.",
> +        )

[ ... ]

> +    @requires_nic_capability(NicCapability.PORT_TX_OFFLOAD_GRE_TNL_TSO)
> +    @func_test
> +    def gre_ip4_pkt_detect(self) -> None:

[ ... ]

> +        with TestPmd() as testpmd:
> +            self._setup_session(testpmd=testpmd, expected_flags=flags, packet_list=packets)

[ ... ]

> +    @requires_nic_capability(NicCapability.PORT_TX_OFFLOAD_OUTER_IPV4_CKSUM)
> +    @requires_nic_capability(NicCapability.PORT_TX_OFFLOAD_IPV4_CKSUM)
> +    @requires_nic_capability(NicCapability.PORT_TX_OFFLOAD_TCP_CKSUM)
> +    @requires_nic_capability(NicCapability.PORT_TX_OFFLOAD_UDP_CKSUM)
> +    @requires_nic_capability(NicCapability.PORT_TX_OFFLOAD_SCTP_CKSUM)
> +    @requires_nic_capability(NicCapability.PORT_TX_OFFLOAD_GRE_TNL_TSO)
> +    @func_test
> +    def gre_checksum_offload(self) -> None:
> +        """GRE checksum offload test.
> +
> +        Steps:
> +            * Craft packets using GRE tunneling.
> +            * Alter checksum of each packet.
> +            * Send packets to the testpmd application.
> +
> +        Verify:
> +            * All packets were received with the expected checksum flags.
> +        """
> +        packets = [
> +            Ether(src=SRC_ID) / IP(chksum=0x0) / GRE() / IP() / TCP(),
> +            Ether(src=SRC_ID) / IP() / GRE() / IP(chksum=0x0) / TCP(),
> +            Ether(src=SRC_ID) / IP() / GRE() / IP() / TCP(chksum=0x0),
> +            Ether(src=SRC_ID) / IP() / GRE() / IP() / UDP(chksum=0xFFFF),
> +            Ether(src=SRC_ID) / IP() / GRE() / IP() / SCTP(chksum=0x0),
> +        ]
> +        good_l4_ip = [
> +            (True, True),
> +            (True, False),
> +            (False, True),
> +            (False, True),
> +            (False, True),
> +        ]
> +        with TestPmd() as testpmd:
> +            testpmd.set_forward_mode(SimpleForwardingModes.csum)
> +            testpmd.csum_set_hw(
> +                layers=ChecksumOffloadOptions.ip
> +                | ChecksumOffloadOptions.udp
> +                | ChecksumOffloadOptions.outer_ip
> +                | ChecksumOffloadOptions.sctp
> +                | ChecksumOffloadOptions.tcp,
> +                port_id=0,
> +            )
> +            testpmd.set_csum_parse_tunnel(port=0, on=True)
> +            testpmd.set_verbose(1)
> +            testpmd.start_all_ports()
> +            testpmd.start()
> +            for i in range(len(packets)):
> +                self._send_packet_and_verify_checksum(
> +                    packets[i],
> +                    good_l4_ip[i][0],
> +                    good_l4_ip[i][1],
> +                    testpmd,
> +                )

Does the loop in gre_checksum_offload() handle the case where
len(good_l4_ip) < len(packets)? Would this cause an IndexError?