Re: [dts] [PATCH V1] tests/port_control: automation port_contol

["Tu, Lijuan" <lijuan.tu@intel.com>]

applied

> -----Original Message-----
> From: dts [mailto:dts-bounces@dpdk.org] On Behalf Of Zeng Xiaoxiao
> Sent: Tuesday, December 31, 2019 4:59 PM
> To: dts@dpdk.org
> Cc: Zeng, XiaoxiaoX <xiaoxiaox.zeng@intel.com>
> Subject: [dts] [PATCH V1] tests/port_control: automation port_contol
> 
> *.according to port_control_test_plan.rst,automation suite port_control
> 
> Signed-off-by: Zeng Xiaoxiao <xiaoxiaox.zeng@intel.com>
> ---
>  tests/TestSuite_port_control.py | 260
> ++++++++++++++++++++++++++++++++++++++++
>  1 file changed, 260 insertions(+)
>  create mode 100644 tests/TestSuite_port_control.py
> 
> diff --git a/tests/TestSuite_port_control.py b/tests/TestSuite_port_control.py
> new file mode 100644 index 0000000..a458add
> --- /dev/null
> +++ b/tests/TestSuite_port_control.py
> @@ -0,0 +1,260 @@
> +# 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 os
> +import utils
> +import time
> +import re
> +import packet
> +from test_case import TestCase
> +from pmd_output import PmdOutput
> +from virt_common import VM
> +
> +
> +class TestPortControl(TestCase):
> +
> +    def set_up_all(self):
> +        """
> +        Run before each test suite
> +        """
> +        # initialize ports topology
> +        self.vm0 = None
> +        self.env_done = False
> +        self.port_id_0 = 0
> +        self.pkt_count = 1000
> +        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.pf_mac = self.dut.get_mac_address(self.dut_ports[0])
> +        self.vf_mac = "00:01:23:45:67:89"
> +        self.txitf =
> self.tester.get_interface(self.tester.get_local_port(self.dut_ports[0]))
> +        self.host_testpmd = PmdOutput(self.dut)
> +        self.vf_assign_method = 'vfio-pci'
> +        self.dut.send_expect('modprobe vfio-pci', '#')
> +        self.socket = self.dut.get_numa_id(self.dut_ports[0])
> +        port = self.dut.ports_info[0]['port']
> +        self.pf_default_driver = port.get_nic_driver()
> +
> +    def set_up(self):
> +        """
> +        Run before each test case.
> +        """
> +        pass
> +
> +    def setup_vm_env(self, driver='default'):
> +        """
> +        Create testing environment with 1VF generated from 1PF
> +        """
> +        if self.env_done:
> +            return
> +
> +        # bind to default driver
> +        self.bind_nic_driver(self.dut_ports[:1], driver="")
> +        self.used_dut_port = self.dut_ports[0]
> +        self.host_intf = self.dut.ports_info[self.used_dut_port]['intf']
> +        self.dut.generate_sriov_vfs_by_port(self.used_dut_port, 1,
> driver=driver)
> +        self.sriov_vfs_port = self.dut.ports_info[self.used_dut_port]['vfs_port']
> +        self.dut.send_expect("ip link set %s vf 0 mac %s" % (self.host_intf,
> self.vf_mac), "# ")
> +        try:
> +            for port in self.sriov_vfs_port:
> +                port.bind_driver(self.vf_assign_method)
> +            time.sleep(1)
> +            vf_popt = {'opt_host': self.sriov_vfs_port[0].pci}
> +
> +            # set up VM ENV
> +            self.vm = VM(self.dut, 'vm0', 'port_control')
> +            self.vm.set_vm_device(driver=self.vf_assign_method, **vf_popt)
> +            self.vm_dut = self.vm.start()
> +            if self.vm_dut is None:
> +                raise Exception("Set up VM ENV failed!")
> +
> +            self.vm_testpmd = PmdOutput(self.vm_dut)
> +
> +        except Exception as e:
> +            self.destroy_vm_env()
> +            raise Exception(e)
> +
> +        self.env_done = True
> +
> +    def destroy_vm_env(self):
> +        if getattr(self, 'vm', None):
> +            if getattr(self, 'vm_dut', None):
> +                self.vm_dut.kill_all()
> +            self.vm_testpmd = None
> +            self.vm_dut_ports = None
> +            # destroy vm0
> +            self.vm.stop()
> +            self.dut.virt_exit()
> +            time.sleep(3)
> +            self.vm = None
> +
> +        if getattr(self, 'used_dut_port', None) != None:
> +            self.dut.destroy_sriov_vfs_by_port(self.used_dut_port)
> +            self.used_dut_port = None
> +        self.bind_nic_driver(self.dut_ports[:1],
> + driver=self.pf_default_driver)
> +
> +        self.env_done = False
> +
> +    def bind_nic_driver(self, ports, driver=""):
> +        # modprobe vfio driver
> +        if driver == "vfio-pci":
> +            for port in ports:
> +                netdev = self.dut.ports_info[port]['port']
> +                driver = netdev.get_nic_driver()
> +                if driver != 'vfio-pci':
> +                    netdev.bind_driver(driver=self.d)
> +
> +        elif driver == "igb_uio":
> +            # igb_uio should insmod as default, no need to check
> +            for port in ports:
> +                netdev = self.dut.ports_info[port]['port']
> +                driver = netdev.get_nic_driver()
> +                if driver != 'igb_uio':
> +                    netdev.bind_driver(driver='igb_uio')
> +        else:
> +            for port in ports:
> +                netdev = self.dut.ports_info[port]['port']
> +                driver_now = netdev.get_nic_driver()
> +                if driver is None:
> +                    driver = netdev.default_driver
> +                if driver != driver_now:
> +                    netdev.bind_driver(driver=driver)
> +
> +    def start_testpmd(self, terminal):
> +        terminal.start_testpmd(ports=[0], socket=self.socket)
> +        res = terminal.wait_link_status_up('all', timeout=5)
> +        self.verify(res is True, 'there have port link is down')
> +        terminal.execute_cmd('set fwd mac')
> +        terminal.execute_cmd('set promisc all off')
> +
> +    def start_pmd_port(self, terminal):
> +        terminal.execute_cmd("port start all")
> +        terminal.execute_cmd("start")
> +        terminal.wait_link_status_up('all', timeout=5)
> +        ret = terminal.get_port_link_status(self.port_id_0)
> +        self.verify(ret == "up", "port not up!")
> +
> +    def stop_pmd_port(self, terminal):
> +        terminal.execute_cmd("stop")
> +        terminal.execute_cmd("port stop all")
> +        ret = terminal.get_port_link_status(self.port_id_0)
> +        self.verify(ret == "down", "port not down!")
> +
> +    def reset_pmd_port(self, terminal):
> +        terminal.execute_cmd("port reset all")
> +        ret = terminal.get_port_link_status(self.port_id_0)
> +        self.verify(ret == "down", "port reset fail!")
> +
> +    def close_pmd_port(self, terminal):
> +        terminal.execute_cmd("port close all")
> +        ret = terminal.execute_cmd("show port info all")
> +        ret = ret.split('\r')
> +        self.verify(ret[1] == '', "close all port fail!")
> +
> +    def calculate_stats(self, start_stats, end_stats):
> +        ret_stats = {}
> +        ret_stats['RX-packets'] = int(end_stats['RX-packets']) -
> int(start_stats['RX-packets'])
> +        ret_stats['TX-packets'] = int(end_stats['TX-packets']) -
> int(start_stats['TX-packets'])
> +        return ret_stats
> +
> +    def send_and_verify_packets(self, terminal):
> +        """
> +        Send packets according to parameters.
> +        """
> +        if terminal is self.host_testpmd:
> +            self.dts_mac = self.pf_mac
> +        else:
> +            self.dts_mac = self.vf_mac
> +
> +        self.pkt = packet.Packet('Ether(dst="%s")/IP()/Raw("x"*40)' %
> + self.dts_mac)
> +
> +        pf_start_stats = terminal.get_pmd_stats(self.port_id_0)
> +        self.pkt.send_pkt(crb=self.tester, tx_port=self.txitf,
> count=self.pkt_count, timeout=30)
> +        pf_end_stats = terminal.get_pmd_stats(self.port_id_0)
> +        pf_ret_stats = self.calculate_stats(pf_start_stats,
> + pf_end_stats)
> +
> +        self.verify(pf_ret_stats['RX-packets'] == self.pkt_count and
> pf_ret_stats['TX-packets'] == self.pkt_count,
> +                    "Packets receive and forward fail!")
> +
> +    def test_pf_start_stop_reset_close(self):
> +        self.start_testpmd(self.host_testpmd)
> +        # start port
> +        self.start_pmd_port(self.host_testpmd)
> +        self.send_and_verify_packets(self.host_testpmd)
> +        # stop port and start port
> +        self.stop_pmd_port(self.host_testpmd)
> +        self.start_pmd_port(self.host_testpmd)
> +        self.send_and_verify_packets(self.host_testpmd)
> +        # reset port
> +        self.stop_pmd_port(self.host_testpmd)
> +        self.reset_pmd_port(self.host_testpmd)
> +        self.start_pmd_port(self.host_testpmd)
> +        self.send_and_verify_packets(self.host_testpmd)
> +        # close all port
> +        self.stop_pmd_port(self.host_testpmd)
> +        self.close_pmd_port(self.host_testpmd)
> +
> +    def test_e1000_start_stop_reset_close(self):
> +        self.setup_vm_env()
> +        self.start_testpmd(self.vm_testpmd)
> +        # start port
> +        self.start_pmd_port(self.vm_testpmd)
> +        self.send_and_verify_packets(self.vm_testpmd)
> +        # stop port and start port
> +        self.stop_pmd_port(self.vm_testpmd)
> +        self.start_pmd_port(self.vm_testpmd)
> +        self.send_and_verify_packets(self.vm_testpmd)
> +        # reset port
> +        self.stop_pmd_port(self.vm_testpmd)
> +        self.reset_pmd_port(self.vm_testpmd)
> +        self.start_pmd_port(self.vm_testpmd)
> +        self.send_and_verify_packets(self.vm_testpmd)
> +        # close all port
> +        self.stop_pmd_port(self.vm_testpmd)
> +        self.close_pmd_port(self.vm_testpmd)
> +
> +    def tear_down(self):
> +        """
> +        Run after each test case.
> +        """
> +        if self.env_done:
> +            self.vm_testpmd.quit()
> +            self.destroy_vm_env()
> +        else:
> +            self.host_testpmd.quit()
> +
> +    def tear_down_all(self):
> +        """
> +        Run after each test suite.
> +        """
> +        if self.env_done:
> +            self.destroy_vm_env()
> +        self.dut.kill_all()
> --
> 1.8.3.1