[dts] [PATCH] checksum checks: Add hardware offload l3 and l4 cases

[Owen Hilyard <ohilyard@iol.unh.edu>]

add rx and tx l3 test cases
add tx l4 test case
add documentation for new test cases

Signed-off-by: Owen Hilyard <ohilyard@iol.unh.edu>
---
 test_plans/checksum_offload_test_plan.rst | 122 +++++++-
 tests/TestSuite_checksum_offload.py       | 331 +++++++++++++++++++---
 2 files changed, 412 insertions(+), 41 deletions(-)

diff --git a/test_plans/checksum_offload_test_plan.rst b/test_plans/checksum_offload_test_plan.rst
index 4fd8c5a..de2e1a9 100644
--- a/test_plans/checksum_offload_test_plan.rst
+++ b/test_plans/checksum_offload_test_plan.rst
@@ -221,7 +221,7 @@ combination: good/bad ip checksum + good/bad udp/tcp checksum.
 
 Check the Rx checksum flags consistent with expected flags.
 
-Test Case: Hardware Checksum Check L4
+Test Case: Hardware Checksum Check L4 RX
 ===========================================
 This test involves testing many different scenarios with a L4 checksum.
 A variety of tunneling protocols, L3 protocols and L4 protocols are combined
@@ -256,4 +256,122 @@ Send a packet with a bad checksum::
    rx: l2_len=18 ethertype=800 l3_len=20 l4_proto=17 l4_len=8 flags=PKT_RX_L4_CKSUM_BAD PKT_RX_IP_CKSUM_BAD PKT_RX_OUTER_L4_CKSUM_UNKNOWN
    tx: flags=PKT_TX_L4_NO_CKSUM PKT_TX_IPV4
 
-Verify flags are as expected.
\ No newline at end of file
+Verify flags are as expected.
+
+Test Case: Hardware Checksum Check L3 RX
+===========================================
+This test involves testing L3 checksum hardware offload.
+Due to the relative dominance of IPv4 and IPv6 as L3 protocols, and IPv6's
+lack of a checksum, only IPv4's checksum is tested.
+
+Setup the ``csum`` forwarding mode::
+
+  testpmd> set fwd csum
+  Set csum packet forwarding mode
+
+Start the packet forwarding::
+
+  testpmd> start
+    csum packet forwarding - CRC stripping disabled - packets/burst=32
+    nb forwarding cores=1 - nb forwarding ports=10
+    RX queues=1 - RX desc=128 - RX free threshold=64
+    RX threshold registers: pthresh=8 hthresh=8 wthresh=4
+    TX queues=1 - TX desc=512 - TX free threshold=0
+    TX threshold registers: pthresh=32 hthresh=8 wthresh=8
+
+Send a packet with a good checksum::
+
+   port=0, mbuf=0x2269df8780, pkt_len=96, nb_segs=1:
+   rx: l2_len=18 ethertype=800 l3_len=20 l4_proto=17 l4_len=8 flags=PKT_RX_L4_CKSUM_GOOD PKT_RX_IP_CKSUM_GOOD  PKT_RX_OUTER_L4_CKSUM_UNKNOWN
+   tx: flags=PKT_TX_L4_NO_CKSUM PKT_TX_IPV4
+
+Send a packet with a bad checksum::
+
+   port=0, mbuf=0x2269df7e40, pkt_len=96, nb_segs=1:
+   rx: l2_len=18 ethertype=800 l3_len=20 l4_proto=17 l4_len=8 flags=PKT_RX_L4_CKSUM_BAD PKT_RX_IP_CKSUM_BAD PKT_RX_OUTER_L4_CKSUM_UNKNOWN
+   tx: flags=PKT_TX_L4_NO_CKSUM PKT_TX_IPV4
+
+Verify flags are as expected.
+
+Test Case: Hardware Checksum Check L4 TX
+===========================================
+This test involves testing many different scenarios with a L4 checksum.
+A variety of tunneling protocols, L3 protocols and L4 protocols are combined
+to test as many scenarios as possible. Currently, UDP, TCP and SCTP are used
+as L4 protocols, with IP and IPv6 being used at level 3. The tested tunneling
+protocols are VXLAN and GRE. This test is used to determine whether the
+hardware offloading of checksums works properly.
+
+Setup the ``csum`` forwarding mode::
+
+  testpmd> set fwd csum
+  Set csum packet forwarding mode
+
+Start the packet forwarding::
+
+  testpmd> start
+    csum packet forwarding - CRC stripping disabled - packets/burst=32
+    nb forwarding cores=1 - nb forwarding ports=10
+    RX queues=1 - RX desc=128 - RX free threshold=64
+    RX threshold registers: pthresh=8 hthresh=8 wthresh=4
+    TX queues=1 - TX desc=512 - TX free threshold=0
+    TX threshold registers: pthresh=32 hthresh=8 wthresh=8
+
+
+Start a packet capture on the tester in the backround::
+
+   # tcpdump -i <iface> -s 65535 -w /tmp/tester/test_hardware_checksum_check_l4_tx_capture.pcap &
+
+Send a packet with a good checksum::
+
+   port=0, mbuf=0x2269df8780, pkt_len=96, nb_segs=1:
+   rx: l2_len=18 ethertype=800 l3_len=20 l4_proto=17 l4_len=8 flags=PKT_RX_L4_CKSUM_GOOD PKT_RX_IP_CKSUM_GOOD  PKT_RX_OUTER_L4_CKSUM_UNKNOWN
+   tx: flags=PKT_TX_L4_NO_CKSUM PKT_TX_IPV4
+
+Send a packet with a bad checksum::
+
+   port=0, mbuf=0x2269df7e40, pkt_len=96, nb_segs=1:
+   rx: l2_len=18 ethertype=800 l3_len=20 l4_proto=17 l4_len=8 flags=PKT_RX_L4_CKSUM_BAD PKT_RX_IP_CKSUM_GOOD PKT_RX_OUTER_L4_CKSUM_UNKNOWN
+   tx: flags=PKT_TX_L4_NO_CKSUM PKT_TX_IPV4
+
+Inspect the pcap file from the packet capture and verify the checksums.
+
+Test Case: Hardware Checksum Check L3 TX
+===========================================
+This test involves testing L3 checksum hardware offload.
+Due to the relative dominance of IPv4 and IPv6 as L3 protocols, and IPv6's
+lack of a checksum, only IPv4's checksum is tested.
+
+Setup the ``csum`` forwarding mode::
+
+  testpmd> set fwd csum
+  Set csum packet forwarding mode
+
+Start the packet forwarding::
+
+  testpmd> start
+    csum packet forwarding - CRC stripping disabled - packets/burst=32
+    nb forwarding cores=1 - nb forwarding ports=10
+    RX queues=1 - RX desc=128 - RX free threshold=64
+    RX threshold registers: pthresh=8 hthresh=8 wthresh=4
+    TX queues=1 - TX desc=512 - TX free threshold=0
+    TX threshold registers: pthresh=32 hthresh=8 wthresh=8
+
+
+Start a packet capture on the tester in the backround::
+
+   # tcpdump -i <iface> -s 65535 -w /tmp/tester/test_hardware_checksum_check_l3_tx_capture.pcap &
+
+Send a packet with a good checksum with a 1 in it's payload::
+
+   port=0, mbuf=0x2269df8780, pkt_len=96, nb_segs=1:
+   rx: l2_len=18 ethertype=800 l3_len=20 l4_proto=17 l4_len=8 flags=PKT_RX_L4_CKSUM_GOOD PKT_RX_IP_CKSUM_GOOD  PKT_RX_OUTER_L4_CKSUM_UNKNOWN
+   tx: flags=PKT_TX_L4_NO_CKSUM PKT_TX_IPV4
+
+Send a packet with a bad checksum with a 0 in it's payload::
+
+   port=0, mbuf=0x2269df7e40, pkt_len=96, nb_segs=1:
+   rx: l2_len=18 ethertype=800 l3_len=20 l4_proto=17 l4_len=8 flags=PKT_RX_L4_CKSUM_GOOD PKT_RX_IP_CKSUM_BAD PKT_RX_OUTER_L4_CKSUM_UNKNOWN
+   tx: flags=PKT_TX_L4_NO_CKSUM PKT_TX_IPV4
+
+Inspect the pcap file from the packet capture and verify the checksums.
diff --git a/tests/TestSuite_checksum_offload.py b/tests/TestSuite_checksum_offload.py
index 9ff5a41..c4a877d 100644
--- a/tests/TestSuite_checksum_offload.py
+++ b/tests/TestSuite_checksum_offload.py
@@ -47,8 +47,18 @@ import time
 
 from rst import RstReport
 import utils
-
+from exception import VerifyFailure
+from pktgen import PacketGeneratorHelper
+from scapy.layers.inet import UDP, TCP, IP
+from scapy.layers.inet6 import IPv6
+from scapy.layers.l2 import Ether, GRE
+from scapy.layers.sctp import SCTP
+from scapy.layers.vxlan import VXLAN
+from scapy.packet import Raw
+from scapy.utils import wrpcap, rdpcap
+from test_capabilities import DRIVER_TEST_LACK_CAPA
 from test_case import TestCase
+
 from framework.pmd_output import PmdOutput
 from test_capabilities import DRIVER_TEST_LACK_CAPA
 from pktgen import PacketGeneratorHelper
@@ -57,6 +67,33 @@ import packet
 
 from settings import FOLDERS
 
+l3_proto_classes = [
+    IP,
+    IPv6
+]
+
+l4_proto_classes = [
+    UDP,
+    TCP,
+    SCTP
+]
+
+tunnelling_proto_classes = [
+    VXLAN,
+    GRE,
+]
+
+l3_protos = [
+    "IP",
+    "IPv6"
+]
+
+l4_protos = [
+    "UDP",
+    "TCP",
+    "SCTP",
+]
+
 
 class TestChecksumOffload(TestCase):
 
@@ -79,8 +116,6 @@ class TestChecksumOffload(TestCase):
             cur_path = os.path.dirname(
                 os.path.dirname(os.path.realpath(__file__)))
             self.output_path = os.sep.join([cur_path, self.logger.log_path])
-        # create an instance to set stream field setting
-        self.pktgen_helper = PacketGeneratorHelper()
 
     def set_up(self):
         """
@@ -243,7 +278,6 @@ class TestChecksumOffload(TestCase):
     def send_pkt_expect_good_bad_from_flag(self, pkt_str: str, flag: str, test_name: str, should_pass: bool = True):
         self.pmdout.get_output(timeout=5)  # Remove any old output
         self.scapy_exec(f"sendp({pkt_str}, iface=iface)")
-        time.sleep(1)
         testpmd_output: str = self.pmdout.get_output(timeout=5)
         self.verify(flag in testpmd_output,
                     f"Flag {flag[:-1]} not found for test {test_name}, please run test_rx_checksum_valid_flags.")
@@ -269,11 +303,147 @@ class TestChecksumOffload(TestCase):
 
         return None
 
-    def scapy_exec(self, cmd: str):
-        return self.tester.send_expect(cmd, ">>>")
+    def validate_checksum(self, pkt, layer) -> bool:
+        """
+        @param pkt: The packet to validate the checksum of.
+        @return: Whether the checksum was valid.
+        """
+        if pkt is None:
+            return False
+
+        csum = pkt[layer].chksum
+        del pkt[layer].chksum
+        # Converting it to raw will calculate the checksum
+        return layer(Raw(pkt[layer])).chksum == csum
 
-    def scapy_send_append(self, packet: str):
-        return self.tester.scapy_append(f'sendp({packet}, iface=iface)')
+    def scapy_exec(self, cmd: str, timeout=1) -> str:
+        return self.tester.send_expect(cmd, ">>>", timeout=timeout)
+
+    def get_packets(self, dut_mac, tester_mac):
+        eth = Ether(dst=dut_mac, src=tester_mac)
+        packets = []
+        checksum_options = ({}, {'chksum': 0xf},)
+        # Untunneled
+        for l3 in l3_proto_classes:
+            for l4 in l4_proto_classes:
+                for chksum in checksum_options:
+                    # The packet's data can be used to identify how the packet was constructed so avoid any issues with
+                    # ordering
+                    pkt = eth / l3() / l4(**chksum) / (
+                        f'UNTUNNELED,{l3.__name__},{l4.__name__},{" " if len(chksum.values()) == 0 else chksum["chksum"]}'
+                    )
+
+                    # Prevents the default behavior which adds DNS headers
+                    if l4 == UDP:
+                        pkt[UDP].dport, pkt[UDP].sport = 1001, 1001
+
+                    packets.append(pkt)
+
+        # Tunneled
+        # VXLAN
+        for l3 in l3_proto_classes:
+            for l4 in l4_proto_classes:
+                for outer_arg in checksum_options:
+                    for inner_arg in checksum_options:
+                        pkt = eth / l3() / UDP(**outer_arg) / VXLAN() / Ether() / l3() / l4(**inner_arg) / (
+                            f'VXLAN,{l3.__name__},{l4.__name__},'
+                            f'{" " if len(outer_arg.values()) == 0 else outer_arg["chksum"]},'
+                            f'{" " if len(inner_arg.values()) == 0 else inner_arg["chksum"]}'
+                        )
+                        # Prevents the default behavior which adds DNS headers
+                        if l4 == UDP:
+                            pkt[VXLAN][UDP].dport, pkt[VXLAN][UDP].sport = 1001, 1001
+
+                        packets.append(pkt)
+        # GRE
+        for l3 in l3_proto_classes:
+            for l4 in l4_proto_classes:
+                for chksum in checksum_options:
+                    pkt = eth / l3() / GRE() / l3() / l4(**chksum) / (
+                        f'GRE,{l3.__name__},{l4.__name__},{" " if len(chksum.values()) == 0 else chksum["chksum"]}'
+                    )
+
+                    # Prevents the default behavior which adds DNS headers
+                    if l4 == UDP:
+                        pkt[GRE][UDP].dport, pkt[GRE][UDP].sport = 1001, 1001
+
+                    packets.append(pkt)
+
+        return packets
+
+    def replay_pcap_file_on_tester(self, iface, packet_file_path):
+        self.tester.send_expect("scapy", ">>>")
+        self.scapy_exec(f"packets = rdpcap('{packet_file_path}')")
+        self.scapy_exec(f"sendp(packets, iface={iface})")
+        self.tester.send_expect("quit()", "# ")
+
+    def validate_packet_list_checksums(self, packets):
+        name_to_class_dict = {
+            'UDP': UDP,
+            'TCP': TCP,
+            'SCTP': SCTP,
+            'IP': IP,
+            'IPv6': IPv6,
+            'VXLAN': VXLAN,
+            'GRE': GRE,
+        }
+
+        error_messages = []
+
+        untunnelled_error_message = f"Invalid untunneled checksum state for %s/%s with a %s checksum."
+
+        vxlan_error_message = f"Invalid VXLAN tunnelled %s checksum state for %s/%s" \
+                              f" with a %s inner checksum and a %s outer checksum."
+
+        gre_error_message = f"Invalid GRE tunnelled checksum state for %s/%s with a %s checksum."
+
+        for packet in packets:
+            payload: str
+            # try:
+            payload = packet[Raw].load.decode('utf-8').split(",")
+            # # This error usually happens with tunneling protocols, and means that an additional cast is needed
+            # except UnicodeDecodeError:
+            #     for proto in tunnelling_proto_classes:
+
+            l3 = name_to_class_dict[payload[1]]
+            l4 = name_to_class_dict[payload[2]]
+            if payload[0] == "UNTUNNELED":
+                chksum_should_be_valid = payload[3] == " "
+                if self.validate_checksum(packet, l4) != chksum_should_be_valid:
+                    error_messages.append(
+                        untunnelled_error_message % (
+                            l3.__name__, l4.__name__, 'valid' if chksum_should_be_valid == '' else 'invalid'
+                        )
+                    )
+            elif payload[0] == "VXLAN":
+                outer_chksum_should_be_valid = payload[3] == " "
+                inner_chksum_should_be_valid = payload[4] == " "
+                if self.validate_checksum(packet[VXLAN], l4) != inner_chksum_should_be_valid:
+                    error_messages.append(
+                        vxlan_error_message % (
+                            "inner", l4.__name__, l3.__name__,
+                            'valid' if inner_chksum_should_be_valid == '' else 'invalid',
+                            'valid' if outer_chksum_should_be_valid == '' else 'invalid'
+                        )
+                    )
+                if self.validate_checksum(packet, l4) != outer_chksum_should_be_valid:
+                    error_messages.append(
+                        vxlan_error_message % (
+                            "outer", l3.__name__, l4.__name__,
+                            'valid' if inner_chksum_should_be_valid == '' else 'invalid',
+                            'valid' if outer_chksum_should_be_valid == '' else 'invalid'
+                        )
+                    )
+            elif payload[0] == "GRE":
+                chksum_should_be_valid = payload[3] == " "
+                if self.validate_checksum(packet, l4) != chksum_should_be_valid:
+                    error_messages.append(
+                        gre_error_message % (
+                            l3.__name__, l4.__name__, 'valid' if chksum_should_be_valid == '' else 'invalid'
+                        )
+                    )
+
+        return error_messages
 
     #
     #
@@ -448,6 +618,9 @@ class TestChecksumOffload(TestCase):
 
             # clear streams before add new streams
             self.tester.pktgen.clear_streams()
+            # create an instance to set stream field setting
+            # Moved here because it messes with the ability of the functional tests to use scapy.
+            self.pktgen_helper = PacketGeneratorHelper()
             # run packet generator
             streams = self.pktgen_helper.prepare_stream_from_tginput(tgenInput, 100,
                     None, self.tester.pktgen)
@@ -511,53 +684,93 @@ class TestChecksumOffload(TestCase):
             self.dut.send_expect("quit", "#", 10)
             self.result_table_print()
 
-    def test_hardware_checksum_check_ip(self):
+    def test_hardware_checksum_check_ip_rx(self):
         self.dut.send_expect("start", "testpmd>")
+        self.tester.send_expect("scapy", ">>>")
         self.checksum_enablehw(self.dut_ports[0])
+        self.dut.send_expect("start", "testpmd>")
 
         verification_errors: List[VerifyFailure] = []
 
-        # untunnelled
-        vf = self.try_helper_hardware_checksum_check_catch_failure('Ether(dst="%s", src="%s")/IP(%s)/UDP()/("X"*50)',
-                                                                   "PKT_RX_IP_CKSUM_")
-        if vf is not None:
-            verification_errors.append(vf)
+        iface = self.tester.get_interface(self.tester.get_local_port(self.dut_ports[0]))
+        dut_mac = self.dut.get_mac_address(self.dut_ports[0])
+        tester_mac = self.tester.get_mac(self.tester.get_local_port(self.dut_ports[0]))
 
-        # tunneled inner
-        vf = self.try_helper_hardware_checksum_check_catch_failure(
-            'Ether(dst="%s", src="%s")/IP()/UDP()/IP(%s)/("X"*50)',
-            "PKT_RX_OUTER_IP_CKSUM_")
-        if vf is not None:
-            verification_errors.append(vf)
+        self.scapy_exec(f"eth = Ether(dst='{dut_mac}', src='{tester_mac}')")
+        self.scapy_exec(f"iface = '{iface}'")
 
-        # tunneled outer
-        vf = self.try_helper_hardware_checksum_check_catch_failure(
-            'Ether(dst="%s", src="%s")/IP()/UDP()/IP(%s)/("X"*50)',
-            "PKT_RX_OUTER_IP_CKSUM_")
-        if vf is not None:
-            verification_errors.append(vf)
+        # Untunnelled
+        for l4 in l4_protos:
+            for chksum in "", "chksum=0xf":
+                vf = self.send_pkt_expect_good_bad_from_flag_catch_failure(
+                    f"eth/IP({chksum})/{l4}()/(X'*50)",
+                    "PKT_RX_IP_CKSUM_", f"{l4}",
+                    should_pass=(chksum == ""))
+                if vf is not None:
+                    verification_errors.append(vf)
 
-        self.verify(len(verification_errors) == 0, "\n".join(verification_errors))
+        for err in verification_errors:
+            self.logger.error(str(err))
+        self.verify(len(verification_errors) == 0, "See previous output")
 
+        self.tester.send_expect("quit()", "# ")
         self.dut.send_expect("stop", "testpmd>")
 
-    def test_hardware_checksum_check_l4(self):
+    def test_hardware_checksum_check_ip_tx(self):
         self.checksum_enablehw(self.dut_ports[0])
         self.dut.send_expect("start", "testpmd>")
 
         verification_errors: List[VerifyFailure] = []
 
-        l3_protos: List[str] = [
-            "IP",
-            "IPv6"
-        ]
+        iface = self.tester.get_interface(self.tester.get_local_port(self.dut_ports[0]))
+        dut_mac = self.dut.get_mac_address(self.dut_ports[0])
+        tester_mac = self.tester.get_mac(self.tester.get_local_port(self.dut_ports[0]))
+        eth = Ether(dst=dut_mac, src=tester_mac)
 
-        l4_protos: List[str] = [
-            "UDP",
-            "TCP",
-            "SCTP",
+        checksum_options = ({}, {'chksum': 0xf},)
+
+        packets = [
+            eth / IP(**chksum) / TCP() / Raw(load=str(int(len(chksum) != 1))) for chksum in checksum_options
         ]
 
+        capture_file_name = "test_hardware_checksum_check_l3_tx_capture.pcap"
+
+        packet_file_path = "/tmp/test_hardware_checksum_check_l3_tx_packets.pcap"
+        capture_file_path = "/tmp/tester/" + capture_file_name
+
+        self.tester.send_expect(f"tcpdump -i {iface} -s 65535 -w {capture_file_path} &", "# ")
+
+        wrpcap(packet_file_path, packets)
+        self.tester.session.copy_file_to(packet_file_path, packet_file_path)
+
+        self.replay_pcap_file_on_tester(iface, packet_file_path)
+
+        self.tester.session.copy_file_from(packet_file_path, "output/tmp/pcap/" + capture_file_name)
+
+        captured_packets = rdpcap("output/tmp/pcap/" + capture_file_name)
+
+        self.verify(len(packets) == len(captured_packets), "Not all packets were received")
+
+        error_messages = []
+        for pkt in captured_packets:
+            should_pass = pkt[TCP].payload.build() == b'1'
+            if not (self.validate_checksum(pkt, IP) == should_pass):
+                error_messages.append(f"A packet was marked as having a"
+                                      f"{' valid' if should_pass == '' else 'n invalid'}"
+                                      f" checksum when it should have had the opposite.")
+
+        self.dut.send_expect("stop", "testpmd>")
+        if len(error_messages) != 0:
+            for error_msg in error_messages:
+                self.logger.error(error_msg)
+            self.verify(False, "See prior output")
+
+    def test_hardware_checksum_check_l4_rx(self):
+        self.checksum_enablehw(self.dut_ports[0])
+        self.dut.send_expect("start", "testpmd>")
+
+        verification_errors: List[VerifyFailure] = []
+
         iface = self.tester.get_interface(self.tester.get_local_port(self.dut_ports[0]))
         dut_mac = self.dut.get_mac_address(self.dut_ports[0])
         tester_mac = self.tester.get_mac(self.tester.get_local_port(self.dut_ports[0]))
@@ -589,7 +802,7 @@ class TestChecksumOffload(TestCase):
                                 should_pass = outer_arg == ""
                             vf = self.send_pkt_expect_good_bad_from_flag_catch_failure(
                                 f"eth/{l3}()/{l4}({outer_arg})/VXLAN()/{l3}()/"
-                                f"{l4}({inner_arg})/('X'*50)",
+                                f"{l4}(chksum={inner_arg})/('X'*50)",
                                 flag, f"{l3}/{l4}/VXLAN/{l3}/{l4}",
                                 should_pass=should_pass)
 
@@ -602,8 +815,7 @@ class TestChecksumOffload(TestCase):
                 for inner_arg in "", "chksum=0xf":
                     should_pass: bool = inner_arg == ""
                     vf = self.send_pkt_expect_good_bad_from_flag_catch_failure(
-                        f"eth/{l3}()/GRE()/{l3}()/"
-                        f"{l4}({inner_arg})/('X'*50)",
+                        f"eth/{l3}()/GRE()/{l3}()/{l4}({inner_arg})/('X'*50)",
                         "PKT_RX_L4_CKSUM_", f"{l3}/GRE/{l3}/{l4}",
                         should_pass=should_pass)
 
@@ -615,6 +827,8 @@ class TestChecksumOffload(TestCase):
         # updated this test case can easily take advantage of the new functionality.
 
         # # GENEVE
+        # # This import is over here so that it is not forgotten when the update happens
+        # from scapy.contrib.geneve import GENEVE
         # for l3_outer in l3_protos:
         #     for l4_outer in l4_protos:
         #         for l3_inner in l3_protos:
@@ -638,7 +852,46 @@ class TestChecksumOffload(TestCase):
             self.logger.error(str(err))
         self.verify(len(verification_errors) == 0, "See previous output")
 
+        self.tester.send_expect("quit", "#")
+        self.dut.send_expect("stop", "testpmd>")
+
+    def test_hardware_checksum_check_l4_tx(self):
+        self.checksum_enablehw(self.dut_ports[0])
+        self.dut.send_expect("start", "testpmd>")
+
+        verification_errors: List[VerifyFailure] = []
+
+        iface = self.tester.get_interface(self.tester.get_local_port(self.dut_ports[0]))
+        dut_mac = self.dut.get_mac_address(self.dut_ports[0])
+        tester_mac = self.tester.get_mac(self.tester.get_local_port(self.dut_ports[0]))
+
+        packets = self.get_packets(dut_mac, tester_mac)
+
+        capture_file_name = "test_hardware_checksum_check_l4_tx_capture.pcap"
+
+        packet_file_path = "/tmp/test_hardware_checksum_check_l4_tx_packets.pcap"
+        capture_file_path = "/tmp/tester/" + capture_file_name
+
+        self.tester.send_expect(f"tcpdump -i {iface} -s 65535 -w {capture_file_path} &", "# ")
+
+        wrpcap(packet_file_path, packets)
+        self.tester.session.copy_file_to(packet_file_path, packet_file_path)
+
+        self.replay_pcap_file_on_tester(iface, packet_file_path)
+
+        self.tester.session.copy_file_from(packet_file_path, "output/tmp/pcap/" + capture_file_name)
+
+        captured_packets = rdpcap("output/tmp/pcap/" + capture_file_name)
+
+        self.verify(len(packets) == len(captured_packets), "Not all packets were received")
+
+        error_messages = self.validate_packet_list_checksums(captured_packets)
+
         self.dut.send_expect("stop", "testpmd>")
+        if len(error_messages) != 0:
+            for error_msg in error_messages:
+                self.logger.error(error_msg)
+            self.verify(False, "See prior output")
 
     def tear_down(self):
         """
-- 
2.25.1