Re: [dts] [PATCH V1] tests: add new script enable_package_download_in_ice_driver

["Peng, Yuan" <yuan.peng@intel.com>]

Acked by Peng, Yuan <yuan.peng@intel.com>

-----Original Message-----
From: Li, WenjieX A 
Sent: Thursday, June 27, 2019 4:29 PM
To: dts@dpdk.org
Cc: Peng, Yuan <yuan.peng@intel.com>; Li, WenjieX A <wenjiex.a.li@intel.com>
Subject: RE: [dts][PATCH V1] tests: add new script enable_package_download_in_ice_driver

Tested-by: Li, WenjieX A <wenjiex.a.li@intel.com>

> -----Original Message-----
> From: Li, WenjieX A
> Sent: Friday, June 28, 2019 12:39 AM
> To: dts@dpdk.org
> Cc: Peng, Yuan <yuan.peng@intel.com>; Li, WenjieX A 
> <wenjiex.a.li@intel.com>
> Subject: [dts][PATCH V1] tests: add new script 
> enable_package_download_in_ice_driver
> 
> add new script
> 
> Signed-off-by: Wenjie Li <wenjiex.a.li@intel.com>
> ---
>  ...e_enable_package_download_in_ice_driver.py | 270 
> ++++++++++++++++++
>  1 file changed, 270 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..b5e3e2f
> --- /dev/null
> +++ b/tests/TestSuite_enable_package_download_in_ice_driver.py
> @@ -0,0 +1,270 @@
> +# 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
> +import os
> +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)
> +
> +        self.pkg_file = '/lib/firmware/intel/ice/ddp/ice.pkg'
> +        out = self.dut.send_expect("ls %s" % self.pkg_file, "#")
> +        self.verify("No such file or directory" not in out, "Cannot 
> + find %s, please
> check you system/driver." % self.pkg_file)
> +        self.backup_recover_ice_pkg("backup")
> +
> +    def set_up(self):
> +        pass
> +
> +    def backup_recover_ice_pkg(self, flag="backup"):
> +        """
> +        if backup == true: backup /lib/firmware/intel/ice/ddp/ice.pkg 
> + to
> ~/ice.pkg_backup
> +        else: recover ~/ice.pkg_backup to /lib/firmware/intel/ice/ddp/ice.pkg
> +        """
> +        backup_file = '/opt/ice.pkg_backup'
> +        if flag == "backup":
> +            self.dut.send_expect("\cp %s %s" % (self.pkg_file, backup_file), "#")
> +        else:
> +            self.dut.send_expect("\cp %s %s" % (backup_file, 
> + self.pkg_file), "#")
> +
> +    def use_correct_ice_pkg(self, flag="true"):
> +        """
> +        if flag == true: use /lib/firmware/intel/ice/ddp/ice.pkg
> +        else: touch a wrong /lib/firmware/intel/ice/ddp/ice.pkg
> +        """
> +        if flag == "true":
> +            self.backup_recover_ice_pkg("recover")
> +        else:
> +            self.dut.send_expect("rm -rf %s" % self.pkg_file, "#")
> +            self.dut.send_expect("touch %s" % self.pkg_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 and 
> + int(list(set(queue_list))[0])
> == 0, \
> +                            "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.use_correct_ice_pkg(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, recover the default ice.pkg
> +        """
> +        self.backup_recover_ice_pkg("recover")
> --
> 2.17.1