[dts] [PATCH] checksum offload: add test cases for checking offload

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

add test case for ip checksums
add test case for l4 checksums with and without tunneling
add test plans for both test cases

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

diff --git a/test_plans/checksum_offload_test_plan.rst b/test_plans/checksum_offload_test_plan.rst
index 82fd12f..4fd8c5a 100644
--- a/test_plans/checksum_offload_test_plan.rst
+++ b/test_plans/checksum_offload_test_plan.rst
@@ -1,4 +1,5 @@
 .. Copyright (c) <2010-2017>, Intel Corporation
+   Copyright © 2018[, 2019] The University of New Hampshire. All rights reserved.
    All rights reserved.
 
    Redistribution and use in source and binary forms, with or without
@@ -219,3 +220,40 @@ Configure the traffic generator to send the multiple packets with the following
 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
+===========================================
+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.
+
+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.
\ No newline at end of file
diff --git a/tests/TestSuite_checksum_offload.py b/tests/TestSuite_checksum_offload.py
index 9bedd1e..ce5a864 100644
--- a/tests/TestSuite_checksum_offload.py
+++ b/tests/TestSuite_checksum_offload.py
@@ -1,6 +1,7 @@
 # BSD LICENSE
 #
 # Copyright(c) 2010-2014 Intel Corporation. All rights reserved.
+# Copyright © 2018[, 2019] The University of New Hampshire. All rights reserved.
 # All rights reserved.
 #
 # Redistribution and use in source and binary forms, with or without
@@ -38,15 +39,24 @@ Test support of RX/TX Checksum Offload Features by Poll Mode Drivers.
 
 import os
 import re
+import subprocess
+import time
+from typing import List, Union, Tuple, Pattern
+
+import time
+
 from rst import RstReport
 import utils
 
 from test_case import TestCase
-from pmd_output import PmdOutput
+from framework.pmd_output import PmdOutput
 from test_capabilities import DRIVER_TEST_LACK_CAPA
 from pktgen import PacketGeneratorHelper
+from exception import VerifyFailure
 import packet
 
+from settings import FOLDERS
+
 
 class TestChecksumOffload(TestCase):
 
@@ -59,7 +69,7 @@ class TestChecksumOffload(TestCase):
         self.dut_ports = self.dut.get_ports(self.nic)
         # Verify that enough ports are available
         self.verify(len(self.dut_ports) >= 1, "Insufficient ports for testing")
-        self.pmdout = PmdOutput(self.dut)
+        self.pmdout: PmdOutput = PmdOutput(self.dut)
         self.portMask = utils.create_mask([self.dut_ports[0]])
         self.ports_socket = self.dut.get_numa_id(self.dut_ports[0])
         # get dts output path
@@ -67,7 +77,7 @@ class TestChecksumOffload(TestCase):
             self.output_path = self.logger.log_path
         else:
             cur_path = os.path.dirname(
-                                os.path.dirname(os.path.realpath(__file__)))
+                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()
@@ -83,20 +93,23 @@ class TestChecksumOffload(TestCase):
         self.dut.send_expect("set fwd csum", "testpmd>")
 
     def checksum_enablehw(self, port):
-            self.dut.send_expect("port stop all", "testpmd>")
-            self.dut.send_expect("csum set ip hw %d" % port, "testpmd>")
-            self.dut.send_expect("csum set udp hw %d" % port, "testpmd>")
-            self.dut.send_expect("csum set tcp hw %d" % port, "testpmd>")
-            self.dut.send_expect("csum set sctp hw %d" % port, "testpmd>")
-            self.dut.send_expect("port start all", "testpmd>")
+        self.dut.send_expect("port stop all", "testpmd>")
+        self.dut.send_expect("csum set ip hw %d" % port, "testpmd>")
+        self.dut.send_expect("csum set udp hw %d" % port, "testpmd>")
+        self.dut.send_expect("csum set tcp hw %d" % port, "testpmd>")
+        self.dut.send_expect("csum set sctp hw %d" % port, "testpmd>")
+        self.dut.send_expect("csum set outer-ip hw %d" % port, "testpmd>")
+        self.dut.send_expect("csum set outer-udp hw %d" % port, "testpmd>")
+        self.dut.send_expect("csum parse-tunnel on %d" % port, "testpmd>")
+        self.dut.send_expect("port start all", "testpmd>")
 
     def checksum_enablesw(self, port):
-            self.dut.send_expect("port stop all", "testpmd>")
-            self.dut.send_expect("csum set ip sw %d" % port, "testpmd>")
-            self.dut.send_expect("csum set udp sw %d" % port, "testpmd>")
-            self.dut.send_expect("csum set tcp sw %d" % port, "testpmd>")
-            self.dut.send_expect("csum set sctp sw %d" % port, "testpmd>")
-            self.dut.send_expect("port start all", "testpmd>")
+        self.dut.send_expect("port stop all", "testpmd>")
+        self.dut.send_expect("csum set ip sw %d" % port, "testpmd>")
+        self.dut.send_expect("csum set udp sw %d" % port, "testpmd>")
+        self.dut.send_expect("csum set tcp sw %d" % port, "testpmd>")
+        self.dut.send_expect("csum set sctp sw %d" % port, "testpmd>")
+        self.dut.send_expect("port start all", "testpmd>")
 
     def get_chksum_values(self, packets_expected):
         """
@@ -214,6 +227,59 @@ class TestChecksumOffload(TestCase):
 
         return result
 
+    def send_scapy_packet(self, packet: str):
+        itf = self.tester.get_interface(self.tester.get_local_port(self.dut_ports[0]))
+
+        self.tester.scapy_foreground()
+        self.tester.scapy_append(f'sendp({packet}, iface="{itf}")')
+        return self.tester.scapy_execute()
+
+    def get_pkt_rx_l4_cksum(self, testpmd_output: str) -> bool:
+        return self.checksum_flags_are_good("PKT_RX_L4_CKSUM_", testpmd_output)
+
+    def get_pkt_rx_ip_cksum(self, testpmd_output: str) -> bool:
+        return self.checksum_flags_are_good("PKT_RX_IP_CKSUM_", testpmd_output)
+
+    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.")
+        self.verify((flag + "UNKNOWN") not in testpmd_output,
+                    f"Flag {flag[:-1]} was found to be unknown for test {test_name}, indicating a possible lack of support")
+        if should_pass:
+            if flag + "GOOD" in testpmd_output:
+                return
+            else:  # flag + "BAD" in testpmd_output
+                self.verify(False, f"{flag}BAD was found in output, expecting {flag}GOOD.")
+        else:
+            if flag + "BAD" in testpmd_output:
+                return
+            else:  # flag + "GOOD" in testpmd_output
+                self.verify(False, f"{flag}GOOD was found in output, expecting {flag}BAD.")
+
+    def send_pkt_expect_good_bad_from_flag_catch_failure(self, pkt_str: str, flag: str, test_name: str,
+                                                         should_pass: bool = True) -> Union[VerifyFailure, None]:
+        try:
+            self.send_pkt_expect_good_bad_from_flag(pkt_str, flag, test_name, should_pass=should_pass)
+        except VerifyFailure as vf:
+            return vf
+
+        return None
+
+    def scapy_exec(self, cmd: str):
+        return self.tester.send_expect(cmd, ">>>")
+
+    def scapy_send_append(self, packet: str):
+        return self.tester.scapy_append(f'sendp({packet}, iface=iface)')
+
+    #
+    #
+    #
+    # Test Cases
+    #
     def test_checksum_offload_with_vlan(self):
         """
         Do not insert IPv4/IPv6 UDP/TCP checksum on the transmit packet.
@@ -445,6 +511,131 @@ class TestChecksumOffload(TestCase):
             self.dut.send_expect("quit", "#", 10)
             self.result_table_print()
 
+    def test_hardware_checksum_check_ip(self):
+        self.dut.send_expect("start", "testpmd>")
+        self.checksum_enablehw(self.dut_ports[0])
+
+        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)
+
+        # 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)
+
+        # 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)
+
+        self.verify(len(verification_errors) == 0, "\n".join(verification_errors))
+
+        self.dut.send_expect("stop", "testpmd>")
+
+    def test_hardware_checksum_check_l4(self):
+        self.checksum_enablehw(self.dut_ports[0])
+        self.dut.send_expect("start", "testpmd>")
+
+        verification_errors: List[VerifyFailure] = []
+
+        l3_protos: List[str] = [
+            "IP",
+            "IPv6"
+        ]
+
+        l4_protos: List[str] = [
+            "UDP",
+            "TCP",
+            "SCTP",
+        ]
+
+        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]))
+
+        self.tester.send_expect("scapy", ">>> ")
+        self.scapy_exec(f"eth = Ether(dst='{dut_mac}', src='{tester_mac}')")
+        self.scapy_exec(f"iface = '{iface}'")
+        # Untunneled
+        for l3 in l3_protos:
+            for l4 in l4_protos:
+                for chksum in "", "chksum=0xf":
+                    vf = self.send_pkt_expect_good_bad_from_flag_catch_failure(
+                        f"eth/{l3}()/{l4}({chksum})/('X'*50)",
+                        "PKT_RX_L4_CKSUM_", f"{l3}/{l4}",
+                        should_pass=(chksum == ""))
+                    if vf is not None:
+                        verification_errors.append(vf)
+
+        # Tunneled
+        # VXLAN
+        for l3 in l3_protos:
+            for l4 in l4_protos:
+                for outer_arg in "", "chksum=0xf":
+                    for inner_arg in "", "chksum=0xf":
+                        for flag in "PKT_RX_L4_CKSUM_", "PKT_RX_OUTER_L4_CKSUM_":
+                            if flag == "PKT_RX_L4_CKSUM_":
+                                should_pass = inner_arg == ""
+                            else:  # flag == PKT_RX_OUTER_L4_CKSUM_
+                                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)",
+                                flag, f"{l3}/{l4}/VXLAN/{l3}/{l4}",
+                                should_pass=should_pass)
+
+                            if vf is not None:
+                                verification_errors.append(vf)
+
+        # GRE
+        for l3 in l3_protos:
+            for l4 in l4_protos:
+                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)",
+                        "PKT_RX_L4_CKSUM_", f"{l3}/GRE/{l3}/{l4}",
+                        should_pass=should_pass)
+
+                    if vf is not None:
+                        verification_errors.append(vf)
+
+        # # GENEVE
+        # for l3_outer in l3_protos:
+        #     for l4_outer in l4_protos:
+        #         for l3_inner in l3_protos:
+        #             for l4 in l4_protos:
+        #                 for outer_arg in "", "chksum=0xf":
+        #                     for inner_arg in "", "chksum=0xf":
+        #                         for flag in "PKT_RX_L4_CKSUM_", "PKT_RX_OUTER_L4_CKSUM_":
+        #                             should_pass: bool = inner_arg == "" if flag == "PKT_RX_L4_CKSUM_" else outer_arg == ""
+        #                             vf = self.send_pkt_expect_good_bad_from_flag_catch_failure(
+        #                                 f"eth/{l3_outer}()/{l4_outer}({outer_arg})/GENEVE()/eth/{l3_inner}()/"
+        #                                 f"{l4}({inner_arg})/('X'*50)",
+        #                                 flag, f"{l3_outer}/{l4_outer}/VXLAN/{l3_inner}/{l4}",
+        #                                 should_pass=should_pass)
+        #
+        #                             if vf is not None:
+        #                                 verification_errors.append(vf)
+
+        # tunneled inner
+
+        for err in verification_errors:
+            self.logger.error(str(err))
+        self.verify(len(verification_errors) == 0, "See previous output")
+
+        self.dut.send_expect("stop", "testpmd>")
+
     def tear_down(self):
         """
         Run after each test case.
-- 
2.25.1