[PATCH v1 2/2] dts: port over unified packet suite

[Dean Marx <dmarx@iol.unh.edu>]

Port over unified packet testing suite from old DTS. This suite
tests the ability of the PMD to recognize valid or invalid packet flags.

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_uni_pkt.py             | 230 +++++++++++++++++++--
 2 files changed, 219 insertions(+), 14 deletions(-)

diff --git a/dts/framework/config/conf_yaml_schema.json b/dts/framework/config/conf_yaml_schema.json
index f02a310bb5..03ee0ce68a 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",
+        "uni_pkt"
       ]
     },
     "test_target": {
diff --git a/dts/tests/TestSuite_uni_pkt.py b/dts/tests/TestSuite_uni_pkt.py
index 11a856031c..78e0800092 100644
--- a/dts/tests/TestSuite_uni_pkt.py
+++ b/dts/tests/TestSuite_uni_pkt.py
@@ -3,24 +3,36 @@
 
 """Unified packet type flag testing suite.
 
-This suite verifies the ability of the PMD to recognize what
-type of packet is being sent to the DUT and print out the type
-for each layer in verbose output.
+According to DPDK documentation, each Poll Mode Driver should reserve 32 bits
+of packet headers for unified packet type flags. These flags serve as an
+identifier for user applications, and are divided into subcategories:
+L2, L3, L4, tunnel, inner L2, inner L3, and inner L4 types.
+This suite verifies the ability of the driver to recognize these types.
 
 """
 
-from typing import List
+from scapy.packet import Packet  # type: ignore[import-untyped]
+from scapy.layers.vxlan import VXLAN  # type: ignore[import-untyped]
+from scapy.contrib.nsh import NSH  # type: ignore[import-untyped]
+from scapy.layers.inet import IP, ICMP, TCP, UDP, GRE  # type: ignore[import-untyped]
+from scapy.layers.l2 import Ether, ARP  # type: ignore[import-untyped]
+from scapy.packet import Raw
+from scapy.layers.inet6 import IPv6, IPv6ExtHdrFragment  # type: ignore[import-untyped]
+from scapy.layers.sctp import SCTP, SCTPChunkData  # type: ignore[import-untyped]
 
-from scapy.packet import Packet
-from scapy.layers.inet import IP  # type: ignore[import-untyped]
-from scapy.layers.l2 import Ether  # type: ignore[import-untyped]
-from scapy.packet import Raw  # type: ignore[import-untyped]
-
-from framework.remote_session.testpmd_shell import SimpleForwardingModes, TestPmdShell
+from framework.remote_session.testpmd_shell import (SimpleForwardingModes, TestPmdShell,
+                                                    RtePTypes, TestPmdVerbosePacket)
 from framework.test_suite import TestSuite
 
+
 class TestUniPkt(TestSuite):
-    """DPDK Unified packet test suite."""
+    """DPDK Unified packet test suite.
+
+    This testing suite uses testpmd's verbose output hardware/software
+    packet type field to verify the ability of the driver to recognize
+    unified packet types when receiving different packets.
+
+    """
 
     def set_up_suite(self) -> None:
         """Set up the test suite.
@@ -30,5 +42,197 @@ def set_up_suite(self) -> None:
         """
         self.verify(len(self._port_links) > 1, "Not enough ports")
 
-    def send_packet_and_verify_flags(self, expected_flags: List[RtePTypes], packet: Packet) -> None:
-        """"""
\ No newline at end of file
+    def send_packet_and_verify_flags(self, expected_flags: list[RtePTypes],
+                                     packet: Packet, testpmd: TestPmdShell) -> None:
+        """Sends a packet to the DUT and verifies the verbose ptype flags."""
+        testpmd.start()
+        self.send_packet_and_capture(packet=packet)
+        verbose_output = testpmd.extract_verbose_output(testpmd.stop())
+        valid = self.check_for_matching_packet(output=verbose_output, flags=expected_flags)
+        self.verify(valid, f"Packet type flag did not match the expected flag: {expected_flags}.")
+
+    def check_for_matching_packet(self, output: list[TestPmdVerbosePacket],
+                                  flags: list[RtePTypes]) -> bool:
+        """Returns :data:`True` if the packet in verbose output contains all specified flags."""
+        for packet in output:
+            if packet.dst_mac == "00:00:00:00:00:01":
+                for flag in flags:
+                    if (flag not in packet.hw_ptype and flag not in packet.sw_ptype):
+                        return False
+        return True
+
+    def setup_session(self, testpmd: TestPmdShell) -> 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)
+
+    def test_l2_packet_detect(self) -> None:
+        """Verify the correct flags are shown in verbose output when sending L2 packets."""
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id, type=0x88f7) / Raw(),
+            Ether(dst=mac_id) / ARP() / Raw()
+        ]
+        flag_list = [
+            [RtePTypes.L2_ETHER_TIMESYNC],
+            [RtePTypes.L2_ETHER_ARP]
+        ]
+        with TestPmdShell(node=self.sut_node) as testpmd:
+            self.setup_session(testpmd=testpmd)
+            for i in range(0, len(packet_list)):
+                self.send_packet_and_verify_flags(expected_flags=flag_list[i],
+                                                  packet=packet_list[i], testpmd=testpmd)
+
+    def test_l3_l4_packet_detect(self) -> None:
+        """Verify correct flags are shown in the verbose output when sending IP/L4 packets."""
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / Raw(),
+            Ether(dst=mac_id) / IP() / UDP() / Raw(),
+            Ether(dst=mac_id) / IP() / TCP() / Raw(),
+            Ether(dst=mac_id) / IP() / SCTP() / Raw(),
+            Ether(dst=mac_id) / IP() / ICMP() / Raw(),
+            Ether(dst=mac_id) / IP(frag=5) / TCP() / Raw(),
+        ]
+        flag_list = [
+            [RtePTypes.L3_IPV4, RtePTypes.L2_ETHER],
+            [RtePTypes.L4_UDP],
+            [RtePTypes.L4_TCP],
+            [RtePTypes.L4_SCTP],
+            [RtePTypes.L4_ICMP],
+            [RtePTypes.L4_FRAG, RtePTypes.L3_IPV4_EXT_UNKNOWN, RtePTypes.L2_ETHER]
+        ]
+        with TestPmdShell(node=self.sut_node) as testpmd:
+            self.setup_session(testpmd=testpmd)
+            for i in range(0, len(packet_list)):
+                self.send_packet_and_verify_flags(expected_flags=flag_list[i],
+                                                  packet=packet_list[i], testpmd=testpmd)
+
+    def test_ipv6_l4_packet_detect(self) -> None:
+        """Verify correct flags are shown in the verbose output when sending IPv6/L4 packets."""
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id) / IPv6() / Raw(),
+            Ether(dst=mac_id) / IPv6() / UDP() / Raw(),
+            Ether(dst=mac_id) / IPv6() / TCP() / Raw(),
+            Ether(dst=mac_id) / IPv6() / IPv6ExtHdrFragment() / Raw()
+        ]
+        flag_list = [
+            [RtePTypes.L2_ETHER, RtePTypes.L3_IPV6],
+            [RtePTypes.L4_UDP],
+            [RtePTypes.L4_TCP],
+            [RtePTypes.L3_IPV6_EXT_UNKNOWN]
+        ]
+        with TestPmdShell(node=self.sut_node) as testpmd:
+            self.setup_session(testpmd=testpmd)
+            for i in range(0, len(packet_list)):
+                self.send_packet_and_verify_flags(expected_flags=flag_list[i],
+                                                  packet=packet_list[i], testpmd=testpmd)
+
+    def test_l3_tunnel_packet_detect(self) -> None:
+        """Verify correct flags are shown in the verbose output when sending IPv6/L4 packets."""
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / IP(frag=5) / UDP() / Raw(),
+            Ether(dst=mac_id) / IP() / IP() / Raw(),
+            Ether(dst=mac_id) / IP() / IP() / UDP() / Raw(),
+            Ether(dst=mac_id) / IP() / IP() / TCP() / Raw(),
+            Ether(dst=mac_id) / IP() / IP() / SCTP() / Raw(),
+            Ether(dst=mac_id) / IP() / IP() / ICMP() / Raw(),
+            Ether(dst=mac_id) / IP() / IPv6() / IPv6ExtHdrFragment() / Raw()
+        ]
+        flag_list = [
+            [RtePTypes.TUNNEL_IP, RtePTypes.L3_IPV4_EXT_UNKNOWN, RtePTypes.INNER_L4_FRAG],
+            [RtePTypes.TUNNEL_IP, RtePTypes.INNER_L4_NONFRAG],
+            [RtePTypes.TUNNEL_IP, RtePTypes.INNER_L4_UDP],
+            [RtePTypes.TUNNEL_IP, RtePTypes.INNER_L4_TCP],
+            [RtePTypes.TUNNEL_IP, RtePTypes.INNER_L4_SCTP],
+            [RtePTypes.TUNNEL_IP, RtePTypes.INNER_L4_ICMP],
+            [RtePTypes.TUNNEL_IP, RtePTypes.INNER_L3_IPV6_EXT_UNKNOWN, RtePTypes.INNER_L4_FRAG]
+        ]
+        with TestPmdShell(node=self.sut_node) as testpmd:
+            self.setup_session(testpmd=testpmd)
+            for i in range(0, len(packet_list)):
+                self.send_packet_and_verify_flags(expected_flags=flag_list[i],
+                                                  packet=packet_list[i], testpmd=testpmd)
+
+    def test_gre_tunnel_packet_detect(self) -> None:
+        """Verify the correct flags are shown in the verbose output when sending GRE packets."""
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / GRE() / IP(frag=5) / Raw(),
+            Ether(dst=mac_id) / IP() / GRE() / IP() / Raw(),
+            Ether(dst=mac_id) / IP() / GRE() / IP() / UDP() / Raw(),
+            Ether(dst=mac_id) / IP() / GRE() / IP() / TCP() / Raw(),
+            Ether(dst=mac_id) / IP() / GRE() / IP() / SCTP() / Raw(),
+            Ether(dst=mac_id) / IP() / GRE() / IP() / ICMP() / Raw()
+        ]
+        flag_list = [
+            [RtePTypes.TUNNEL_GRENAT, RtePTypes.INNER_L4_FRAG, RtePTypes.INNER_L3_IPV4_EXT_UNKNOWN],
+            [RtePTypes.TUNNEL_GRENAT, RtePTypes.INNER_L4_NONFRAG],
+            [RtePTypes.TUNNEL_GRENAT, RtePTypes.INNER_L4_UDP],
+            [RtePTypes.TUNNEL_GRENAT, RtePTypes.INNER_L4_TCP],
+            [RtePTypes.TUNNEL_GRENAT, RtePTypes.INNER_L4_SCTP],
+            [RtePTypes.TUNNEL_GRENAT, RtePTypes.INNER_L4_ICMP]
+        ]
+        with TestPmdShell(node=self.sut_node) as testpmd:
+            self.setup_session(testpmd=testpmd)
+            for i in range(0, len(packet_list)):
+                self.send_packet_and_verify_flags(expected_flags=flag_list[i],
+                                                  packet=packet_list[i], testpmd=testpmd)
+
+    def test_vxlan_tunnel_packet_detect(self) -> None:
+        """Verify the correct flags are shown in the verbose output when sending VXLAN packets."""
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id) / IP() / UDP() / VXLAN() / Ether() / IP(frag=5) / Raw(),
+            Ether(dst=mac_id) / IP() / UDP() / VXLAN() / Ether() / IP() / Raw(),
+            Ether(dst=mac_id) / IP() / UDP() / VXLAN() / Ether() / IP() / UDP() / Raw(),
+            Ether(dst=mac_id) / IP() / UDP() / VXLAN() / Ether() / IP() / TCP() / Raw(),
+            Ether(dst=mac_id) / IP() / UDP() / VXLAN() / Ether() / IP() / SCTP() / Raw(),
+            Ether(dst=mac_id) / IP() / UDP() / VXLAN() / Ether() / IP() / ICMP() / Raw(),
+            (Ether(dst=mac_id) / IP() / UDP() / VXLAN() / Ether()
+             / IPv6() / IPv6ExtHdrFragment() / Raw())
+        ]
+        flag_list = [
+            [RtePTypes.TUNNEL_GRENAT, RtePTypes.INNER_L4_FRAG, RtePTypes.INNER_L3_IPV4_EXT_UNKNOWN],
+            [RtePTypes.TUNNEL_GRENAT, RtePTypes.INNER_L4_NONFRAG],
+            [RtePTypes.TUNNEL_GRENAT, RtePTypes.INNER_L4_UDP],
+            [RtePTypes.TUNNEL_GRENAT, RtePTypes.INNER_L4_TCP],
+            [RtePTypes.TUNNEL_GRENAT, RtePTypes.INNER_L4_SCTP],
+            [RtePTypes.TUNNEL_GRENAT, RtePTypes.INNER_L4_ICMP],
+            [RtePTypes.TUNNEL_GRENAT, RtePTypes.INNER_L3_IPV6_EXT_UNKNOWN, RtePTypes.INNER_L4_FRAG]
+        ]
+        with TestPmdShell(node=self.sut_node) as testpmd:
+            self.setup_session(testpmd=testpmd)
+            testpmd.rx_vxlan(vxlan_id=4789, port_id=0, add=True)
+            for i in range(0, len(packet_list)):
+                self.send_packet_and_verify_flags(expected_flags=flag_list[i],
+                                                  packet=packet_list[i], testpmd=testpmd)
+
+    def test_nsh_packet_detect(self) -> None:
+        """Verify the correct flags are shown in the verbose output when sending NSH packets."""
+        mac_id = "00:00:00:00:00:01"
+        packet_list = [
+            Ether(dst=mac_id, type=0x894f) / NSH() / IP(),
+            Ether(dst=mac_id, type=0x894f) / NSH() / IP() / ICMP(),
+            Ether(dst=mac_id, type=0x894f) / NSH() / IP(frag=1, flags="MF"),
+            Ether(dst=mac_id, type=0x894f) / NSH() / IP() / TCP(),
+            Ether(dst=mac_id, type=0x894f) / NSH() / IP() / UDP(),
+            Ether(dst=mac_id, type=0x894f) / NSH() / IP() / SCTP(tag=1) / SCTPChunkData(data='x'),
+            Ether(dst=mac_id, type=0x894f) / NSH() / IPv6()
+        ]
+        flag_list = [
+            [RtePTypes.L2_ETHER_NSH, RtePTypes.L3_IPV4_EXT_UNKNOWN, RtePTypes.L4_NONFRAG],
+            [RtePTypes.L2_ETHER_NSH, RtePTypes.L3_IPV4_EXT_UNKNOWN, RtePTypes.L4_ICMP],
+            [RtePTypes.L2_ETHER_NSH, RtePTypes.L3_IPV4_EXT_UNKNOWN, RtePTypes.L4_FRAG],
+            [RtePTypes.L2_ETHER_NSH, RtePTypes.L3_IPV4_EXT_UNKNOWN, RtePTypes.L4_TCP],
+            [RtePTypes.L2_ETHER_NSH, RtePTypes.L3_IPV4_EXT_UNKNOWN, RtePTypes.L4_UDP],
+            [RtePTypes.L2_ETHER_NSH, RtePTypes.L3_IPV4_EXT_UNKNOWN, RtePTypes.L4_SCTP],
+            [RtePTypes.L2_ETHER_NSH, RtePTypes.L3_IPV6_EXT_UNKNOWN, RtePTypes.L4_NONFRAG],
+        ]
+        with TestPmdShell(node=self.sut_node) as testpmd:
+            self.setup_session(testpmd=testpmd)
+            for i in range(0, len(packet_list)):
+                self.send_packet_and_verify_flags(expected_flags=flag_list[i],
+                                                  packet=packet_list[i], testpmd=testpmd)
-- 
2.44.0