[dts] [PATCH V1 4/4] tests: add new suite enable_package_download_in_ice_driver

[Wenjie Li <wenjiex.a.li@intel.com>]

add new suite enable_package_download_in_ice_driver.

Signed-off-by: Wenjie Li <wenjiex.a.li@intel.com>
---
 ...e_enable_package_download_in_ice_driver.py | 255 ++++++++++++++++++
 1 file changed, 255 insertions(+)
 create mode 100644 tests/TestSuite_enable_package_download_in_ice_driver.py

diff --git a/tests/TestSuite_enable_package_download_in_ice_driver.py b/tests/TestSuite_enable_package_download_in_ice_driver.py
new file mode 100644
index 0000000..d6d6634
--- /dev/null
+++ b/tests/TestSuite_enable_package_download_in_ice_driver.py
@@ -0,0 +1,255 @@
+# BSD LICENSE
+#
+# Copyright(c) 2019 Intel Corporation. All rights reserved.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions
+# are met:
+#
+#   * Redistributions of source code must retain the above copyright
+#     notice, this list of conditions and the following disclaimer.
+#   * Redistributions in binary form must reproduce the above copyright
+#     notice, this list of conditions and the following disclaimer in
+#     the documentation and/or other materials provided with the
+#     distribution.
+#   * Neither the name of Intel Corporation nor the names of its
+#     contributors may be used to endorse or promote products derived
+#     from this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+import time
+import re
+from test_case import TestCase
+from pmd_output import PmdOutput
+
+class TestEnable_Package_Download_In_Ice_Driver(TestCase):
+
+    def set_up_all(self):
+        self.verify(self.nic in ["columbiaville_100g", "columbiaville_25g"], "NIC Unsupported: " + str(self.nic))
+        self.dut_ports = self.dut.get_ports(self.nic)
+        self.verify(len(self.dut_ports) >= 2, "Insufficient ports")
+        self.PF_QUEUE = 16
+
+        localPort0 = self.tester.get_local_port(self.dut_ports[0])
+        localPort1 = self.tester.get_local_port(self.dut_ports[1])
+        self.tester_p0 = self.tester.get_interface(localPort0)
+        self.tester_p1 = self.tester.get_interface(localPort1)
+        self.tester.send_expect("ifconfig %s -promisc" % self.tester_p0, "#")
+        self.tester.send_expect("ifconfig %s -promisc" % self.tester_p1, "#")
+
+        self.dut_p0_mac = self.dut.get_mac_address(self.dut_ports[0])
+        self.tester_p0_mac = self.tester.get_mac(localPort0)
+        self.dut_testpmd = PmdOutput(self.dut)
+
+    def set_up(self):
+        pass
+
+    def copy_ice_pkg_to_lib(self, flag):
+        """
+        if flag == true: Copy dep/ice.pkg to DUT /lib/firmware/intel/ice/ddp/ice.pkg
+        else: touch a wrong /lib/firmware/intel/ice/ddp/ice.pkg
+        """
+        file = 'ice.pkg'
+        src_file = r'./dep/%s' % file
+        dst = '/lib/firmware/intel/ice/ddp'
+        self.dut.send_expect("rm -rf %s/%s" % (dst, file), "#")
+        if flag == "true":
+            self.dut.session.copy_file_to(src_file, dst)
+        else:
+            self.dut.send_expect("touch %s/%s" % (dst, file), "#")
+
+    def start_testpmd(self, flag="true"):
+        out = self.dut_testpmd.start_testpmd("all", "--nb-cores=8 --rxq=%s --txq=%s --port-topology=chained" % (self.PF_QUEUE, self.PF_QUEUE))
+        if flag != "true":
+            error_messages = ["ice_load_pkg(): failed to allocate buf of size 0 for package", \
+                    "ice_load_pkg(): failed to allocate buf of size 0 for package", \
+                    "ice_dev_init(): Failed to load the DDP package,Entering Safe Mode", \
+                    "ice_init_rss(): RSS is not supported in safe mode"]
+            for error_message in error_messages:
+                self.verify(error_message in out, "There should be error messages in out: %s" % out)
+        self.dut_testpmd.execute_cmd('set promisc all off')
+        self.dut_testpmd.execute_cmd('set verbose 1')
+
+    def tcpdump_start_sniffing(self, ifaces=[]):
+        """
+        Starts tcpdump in the background to sniff the tester interface where
+        the packets are transmitted to and from the self.dut.
+        All the captured packets are going to be stored in a file for a
+        post-analysis.
+        """
+
+        for iface in ifaces:
+            command = ('tcpdump -w tcpdump_{0}.pcap -i {0} 2>tcpdump_{0}.out &').format(iface)
+            del_cmd = ('rm -f tcpdump_{0}.pcap').format(iface)
+            self.tester.send_expect(del_cmd, '#')
+            self.tester.send_expect(command, '#')
+
+    def tcpdump_stop_sniff(self):
+        """
+        Stops the tcpdump process running in the background.
+        """
+        self.tester.send_expect('killall tcpdump', '#')
+        time.sleep(1)
+        self.tester.send_expect('echo "Cleaning buffer"', '#')
+        time.sleep(1)
+
+    def tcpdump_command(self, command):
+        """
+        Sends a tcpdump related command and returns an integer from the output
+        """
+        result = self.tester.send_expect(command, '#')
+        print result
+        return int(result.strip())
+
+    def number_of_packets(self, iface):
+        """
+        By reading the file generated by tcpdump it counts how many packets were
+        forwarded by the sample app and received in the self.tester. The sample app
+        will add a known MAC address for the test to look for.
+        """
+        command = ('tcpdump -A -nn -e -v -r tcpdump_{iface}.pcap 2>/dev/null | ' +
+                   'grep -c "IPv4"')
+        return self.tcpdump_command(command.format(**locals()))
+
+    def tcpdump_scanner(self, scanner):
+        """
+        Execute scanner to return results
+        """
+        scanner_result = self.tester.send_expect(scanner, '#')
+        fially_result = re.findall(r'length( \d+)', scanner_result)
+        return list(fially_result)
+
+    def send_packet(self, tran_type, flag):
+        """
+        Sends packets.
+        """
+        self.loading_size = 30
+        self.tester.scapy_foreground()
+        self.tester.scapy_append('sys.path.append("./")')
+        self.tester.scapy_append('from sctp import *')
+        if tran_type == "ipv4-other":
+            for i in range(1):
+                packet = r'sendp([Ether(dst="%s", src="%s")/IP(src="192.168.0.%d", dst="192.168.0.%d")/("X"*%s)], iface="%s")' % (
+                    self.dut_p0_mac, self.tester_p0_mac, i + 1, i + 2, self.loading_size, self.tester_p0)
+                self.tester.scapy_append(packet)
+            self.tester.scapy_execute()
+            time.sleep(.5)
+        elif tran_type == "ipv4-tcp":
+            for i in range(16):
+                packet = r'sendp([Ether(dst="%s", src="%s")/IP(src="192.168.0.%d", dst="192.168.0.%d")/TCP(sport=1024,dport=1024)], iface="%s")' % (
+                    self.dut_p0_mac, self.tester_p0_mac, i + 1, i + 2, self.tester_p0)
+                self.tester.scapy_append(packet)
+            self.tester.scapy_execute()
+            time.sleep(.5)
+        elif tran_type == "ipv4-udp":
+            for i in range(16):
+                packet = r'sendp([Ether(dst="%s", src="%s")/IP(src="192.168.0.%d", dst="192.168.0.%d")/UDP(sport=1024,dport=1024)], iface="%s")' % (
+                    self.dut_p0_mac, self.tester_p0_mac, i + 1, i + 2, self.tester_p0)
+                self.tester.scapy_append(packet)
+            self.tester.scapy_execute()
+            time.sleep(.5)
+        elif tran_type == "ipv4-sctp":
+            for i in range(16):
+                packet = r'sendp([Ether(dst="%s", src="%s")/IP(src="192.168.0.%d", dst="192.168.0.%d")/SCTP(sport=1024,dport=1024)], iface="%s")' % (
+                    self.dut_p0_mac, self.tester_p0_mac, i + 1, i + 2, self.tester_p0)
+                self.tester.scapy_append(packet)
+            self.tester.scapy_execute()
+            time.sleep(.5)
+        elif tran_type == "ipv6-tcp":
+            for i in range(16):
+                packet = r'sendp([Ether(dst="%s", src="%s")/IPv6(src="::%d", dst="::%d")/TCP(sport=1024,dport=1024)], iface="%s")' % (
+                    self.dut_p0_mac, self.tester_p0_mac, i + 1, i + 2, self.tester_p0)
+                self.tester.scapy_append(packet)
+            self.tester.scapy_execute()
+            time.sleep(.5)
+        elif tran_type == "ipv6-udp":
+            for i in range(16):
+                packet = r'sendp([Ether(dst="%s", src="%s")/IPv6(src="::%d", dst="::%d")/UDP(sport=1024,dport=1024)], iface="%s")' % (
+                    self.dut_p0_mac, self.tester_p0_mac, i + 1, i + 2, self.tester_p0)
+                self.tester.scapy_append(packet)
+            self.tester.scapy_execute()
+            time.sleep(.5)
+        elif tran_type == "ipv6-sctp":
+            for i in range(16):
+                packet = r'sendp([Ether(dst="%s", src="%s")/IPv6(src="::%d", dst="::%d",nh=132)/SCTP(sport=1024,dport=1024)], iface="%s")' % (
+                    self.dut_p0_mac, self.tester_p0_mac, i + 1, i + 2, self.tester_p0)
+                self.tester.scapy_append(packet)
+            self.tester.scapy_execute()
+            time.sleep(.5)
+        else:
+            print "\ntran_type error!\n"
+
+        self.verifyResult(tran_type=tran_type, flag=flag)
+
+    def verifyResult(self, tran_type, flag):
+        """
+        Verify whether or not the result passes.
+        """
+        if tran_type == "ipv4-other":
+            self.tcpdump_stop_sniff()
+            p0_stats = self.number_of_packets(self.tester_p0)
+            p1_stats = self.number_of_packets(self.tester_p1)
+            self.verify(p0_stats == p1_stats, "tester p0 and p1: packet number match")
+        else:
+            out = self.dut.get_session_output()
+            queue_list = []
+            lines = out.split("\r\n")
+            for line in lines:
+                line = line.strip()
+                if len(line) != 0 and line.startswith(("port 0/queue ",)):
+                    for item in line.split(":"):
+                        item = item.strip()
+                        if(item.startswith("port 0/queue ")):
+                            queue_id = item.split(" ", 2)[-1]
+                            queue_list.append(queue_id)
+            print list(set(queue_list))
+            if flag == "true":
+                self.verify(len(list(set(queue_list))) > 1, "All packets enter the same queue: %s" % queue_list)
+            else:
+                self.verify(len(list(set(queue_list))) == 1, "All packets should enter queue 0, but entered %s" % queue_list)
+
+    def download_the_package(self, flag):
+        """
+        if flag == true: use the correct ice.pkg file; in rxonly mode, all packets should enter different queues of port 0
+        else: use wrong ice.pkg, all packets enter queue 0 of port 0
+        """
+        self.copy_ice_pkg_to_lib(flag=flag)
+        self.start_testpmd(flag=flag)
+        self.dut_testpmd.execute_cmd('set fwd mac')
+        self.dut_testpmd.execute_cmd('start')
+        self.tcpdump_start_sniffing([self.tester_p0, self.tester_p1])
+        self.send_packet(tran_type="ipv4-other", flag=flag)
+
+        self.dut_testpmd.execute_cmd('stop')
+        self.dut_testpmd.execute_cmd('set fwd rxonly')
+        self.dut_testpmd.execute_cmd('start')
+        for tran_types in ["ipv4-tcp", "ipv4-udp", "ipv4-sctp", "ipv6-tcp", "ipv6-udp", "ipv6-sctp"]:
+            print tran_types
+            self.send_packet(tran_type=tran_types, flag=flag)
+
+    def test_download_the_package_successfully(self):
+        self.download_the_package(flag="true")
+
+    def test_download_the_package_failed(self):
+        self.download_the_package(flag="false")
+
+    def tear_down(self):
+        self.dut_testpmd.quit()
+
+    def tear_down_all(self):
+        """
+        After test, apply the correct ice.pkg
+        """
+        self.copy_ice_pkg_to_lib(flag='true')
-- 
2.17.1