* Re: [dts] [PATCH V4]tests/packet_capture: upload automation script
2019-07-30 5:53 ` yufengmx
@ 2019-07-30 12:18 ` Wang, Yinan
2019-08-08 5:48 ` Tu, Lijuan
1 sibling, 0 replies; 5+ messages in thread
From: Wang, Yinan @ 2019-07-30 12:18 UTC (permalink / raw)
To: Mo, YufengX, dts; +Cc: Mo, YufengX
Acked-by: Wang, Yinan <yinan.wang@intel.com>
> -----Original Message-----
> From: dts [mailto:dts-bounces@dpdk.org] On Behalf Of yufengmx
> Sent: 2019年7月30日 13:54
> To: dts@dpdk.org
> Cc: Mo, YufengX <yufengx.mo@intel.com>
> Subject: [dts] [PATCH V4]tests/packet_capture: upload automation script
>
>
> Packet capture framework feature support packet capturing on dpdk Ethernet
> devices. DPDK provides dpdk-pdump tool under app/pdump directory for packet
> capturing on dpdk.
>
> Signed-off-by: yufengmx <yufengx.mo@intel.com>
> ---
> tests/TestSuite_packet_capture.py | 869
> ++++++++++++++++++++++++++++++++++++++
> 1 file changed, 869 insertions(+)
> create mode 100644 tests/TestSuite_packet_capture.py
>
> diff --git a/tests/TestSuite_packet_capture.py
> b/tests/TestSuite_packet_capture.py
> new file mode 100644
> index 0000000..4bab902
> --- /dev/null
> +++ b/tests/TestSuite_packet_capture.py
> @@ -0,0 +1,869 @@
> +# BSD LICENSE
> +#
> +# Copyright(c) 2010-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.
> +'''
> +DPDK Test suite.
> +Test support dpdk pdump tool features
> +'''
> +import random
> +import os
> +import traceback
> +import time
> +import re
> +import signal
> +import subprocess
> +from pprint import pformat
> +
> +from scapy.utils import rdpcap
> +from scapy.packet import NoPayload
> +from scapy.packet import Packet as scapyPacket from scapy.fields import
> +ConditionalField
> +
> +from test_case import TestCase
> +from exception import VerifyFailure
> +from pmd_output import PmdOutput
> +from packet import Packet
> +
> +
> +# These source code copy from packet.py module before
> +sniff_packets/load_sniff_packets # refactor. New refactor methods have
> +much more longer time consumption than # old methods.
> +
> +# Saved back groud sniff process id
> +SNIFF_PIDS = {}
> +
> +
> +def sniff_packets(intf, count=0, timeout=5, pcap=None):
> + """
> + sniff all packets for certain port in certain seconds.
> + """
> + param = ""
> + direct_param = r"(\s+)\[ -(\w) in\|out\|inout \]"
> + tcpdump_help = subprocess.check_output(
> + "tcpdump -h; echo 0", stderr=subprocess.STDOUT, shell=True)
> + for line in tcpdump_help.split('\n'):
> + m = re.match(direct_param, line)
> + if m:
> + param = "-" + m.group(2) + " in"
> +
> + if len(param) == 0:
> + print "tcpdump not support direction choice!!!"
> +
> + sniff_cmd = 'tcpdump -i %(INTF)s %(IN_PARAM)s -w %(FILE)s'
> + options = {'INTF': intf, 'COUNT': count, 'IN_PARAM': param,
> + 'FILE': '/tmp/sniff_%s.pcap' % intf if not pcap else pcap}
> + if count:
> + sniff_cmd += ' -c %(COUNT)d'
> + cmd = sniff_cmd % options
> + else:
> + cmd = sniff_cmd % options
> +
> + args = cmd.split()
> + pipe = subprocess.Popen(args)
> + index = str(time.time())
> + SNIFF_PIDS[index] = (pipe, intf, timeout)
> + time.sleep(0.5)
> + return index
> +
> +
> +def load_sniff_packets(index=''):
> + pkts = []
> + child_exit = False
> + if index in SNIFF_PIDS.keys():
> + pipe, intf, timeout = SNIFF_PIDS[index]
> + time_elapse = int(time.time() - float(index))
> + while time_elapse < timeout:
> + if pipe.poll() is not None:
> + child_exit = True
> + break
> +
> + time.sleep(1)
> + time_elapse += 1
> +
> + if not child_exit:
> + pipe.send_signal(signal.SIGINT)
> + pipe.wait()
> +
> + # wait pcap file ready
> + time.sleep(1)
> + try:
> + cap_pkts = rdpcap("/tmp/sniff_%s.pcap" % intf)
> + for pkt in cap_pkts:
> + # packet gen should be scapy
> + packet = Packet(tx_port=intf)
> + packet.pktgen.assign_pkt(pkt)
> + pkts.append(packet)
> + except Exception as e:
> + pass
> +
> + return pkts
> +
> +
> +class parsePacket(object):
> +
> + def __init__(self, filename, skip_flag=False):
> + self.pcapFile = filename
> + self.packetLayers = dict()
> + self.skip = skip_flag
> +
> + def parse_packet_layer(self, pkt_object):
> + if pkt_object is None:
> + return
> + self.packetLayers[pkt_object.name] = dict()
> + for curfield in pkt_object.fields_desc:
> + if isinstance(curfield, ConditionalField) and \
> + not curfield._evalcond(pkt_object):
> + continue
> + field_value = pkt_object.getfieldval(curfield.name)
> + if isinstance(field_value, scapyPacket) or \
> + (curfield.islist and
> + curfield.holds_packets and
> + type(field_value) is list):
> + continue
> + repr_value = curfield.i2repr(pkt_object, field_value)
> + if isinstance(repr_value, str):
> + repr_value = repr_value.replace(
> + os.linesep, os.linesep + " " * (len(curfield.name) + 4))
> + self.packetLayers[pkt_object.name][curfield.name] = repr_value
> + if not isinstance(pkt_object.payload, NoPayload):
> + self.parse_packet_layer(pkt_object.payload)
> +
> + def get_valid_packet(self, pcap_pkts_origin, number):
> + cnt = 0
> + for packet in pcap_pkts_origin:
> + self.parse_packet_layer(packet)
> + flag = 'LLDP' not in self.packetLayers.keys() \
> + if self.skip else \
> + True
> + src_ether = self.packetLayers["Ethernet"]['src']
> + if src_ether == '00:00:00:00:00:00' and flag:
> + if number == cnt:
> + break
> + cnt += 1
> + self.packetLayers.clear()
> +
> + def parse_pcap(self, number=0):
> + pcap_pkts = []
> + try:
> + if not os.path.exists(self.pcapFile):
> + warning = "{0} is not exist !".format(self.pcapFile)
> + return warning
> + pcap_pkts = rdpcap(self.pcapFile)
> + if len(pcap_pkts) == 0:
> + warning = "{0} is empty".format(self.pcapFile)
> + return warning
> + elif number >= len(pcap_pkts):
> + warning = "{0} is missing No.{1} packet".format(
> + self.pcapFile, number)
> + return warning
> + self.get_valid_packet(pcap_pkts, number)
> + except Exception, e:
> + print (e)
> +
> + return None
> +
> +
> +class TestPacketCapture(TestCase):
> +
> + def is_existed_on_dut(self, check_path):
> + self.dut.alt_session.send_expect("ls %s" % check_path, "# ")
> + cmd = "echo $?"
> + output = self.dut.alt_session.send_expect(cmd, "# ")
> + ret = True if output and output.strip() == "0" else False
> + return ret
> +
> + def get_dut_iface_with_kernel_driver(self):
> + # only physical nic support PROMISC
> + cmd = "ip link show | grep PROMISC | awk {'print $2'}"
> + out = self.dut.alt_session.send_expect(cmd, "# ")
> + pat = "(.*):"
> + ifaces = [intf for intf in re.findall(pat, out, re.M) if intf]
> + # get dut/tester ip to check if they are in a platform
> + tester_ip = self.tester.get_ip_address()
> + dut_ip = self.dut.get_ip_address()
> + for link_port in range(len(self.dut_ports)):
> + # if they are in a platform, ignore interface used by tester
> + if tester_ip == dut_ip:
> + tester_port = self.tester.get_local_port(link_port)
> + intf = self.tester.get_interface(tester_port)
> + if intf in ifaces:
> + ifaces.remove(intf)
> + # ignore interface used by dut
> + intf = self.dut.ports_info[link_port]['intf']
> + if intf in ifaces:
> + ifaces.remove(intf)
> +
> + # set ports up
> + tmp_ifaces = ifaces[:]
> + for iface in tmp_ifaces:
> + # ignore current interface used by system
> + cmd = "ifconfig %s | grep 'inet ' " % iface
> + if self.dut.alt_session.send_expect(cmd, "# ") != "":
> + ifaces.remove(iface)
> + self.dut.alt_session.send_expect(
> + "ifconfig {0} up".format(iface), "# ")
> + time.sleep(10)
> + # get ports on link status
> + tmp_ifaces = ifaces[:]
> + for iface in tmp_ifaces:
> + cmd = "ip link show {0} | grep LOWER_UP".format(iface)
> + self.dut.alt_session.send_expect(cmd, "# ")
> + output = self.dut.alt_session.send_expect(
> + "echo $?".format(iface), "# ").strip()
> + if output != '0':
> + ifaces.remove(iface)
> +
> + self.verify(len(ifaces) >= 2, "Insufficient ports for iface testing")
> + # set iface port for pdump tool output dump packets
> + self.rx_iface = ifaces[0]
> + self.tx_iface = ifaces[1]
> + self.rxtx_iface = ifaces[0]
> +
> + def verify_supported_nic(self):
> + supported_drivers = ['i40e', 'ixgbe']
> + result = all([self.dut.ports_info[index]['port'].default_driver in
> + supported_drivers
> + for index in self.dut_ports])
> + msg = "current nic is not supported"
> + self.verify(result, msg)
> +
> + def get_tcpdump_options(self):
> + param = ""
> + direct_param = r"(\s+)\[ -(\w) in\|out\|inout \]"
> + tcpdump_help = self.dut.alt_session.send_expect("tcpdump -h", "# ")
> + for line in tcpdump_help.split('\n'):
> + m = re.match(direct_param, line)
> + if m:
> + param = "-" + m.group(2) + " out"
> + self.tcpdump = "tcpdump -i {0} " + param + " -w {1} >/dev/null 2>&1 &"
> +
> + def check_pcap_lib(self):
> + pcap_lib_dir = os.sep.join([self.dut.base_dir,
> + self.target,
> + "lib/librte_pmd_pcap.a"])
> + return self.is_existed_on_dut(pcap_lib_dir)
> +
> + def get_packet_types(self):
> + packet_types = ["TCP",
> + "UDP",
> + "SCTP",
> + "TIMESYNC",
> + "ARP",
> + "LLDP",
> + "IP_RAW",
> + "IPv6_TCP",
> + "IPv6_UDP",
> + "IPv6_SCTP",
> + "VLAN_UDP"]
> + if self.pkt_index is None: # only initialize once
> + self.pkt_index = random.randint(0, len(packet_types) - 1)
> + test_packet_types = packet_types if self.full_test else \
> + [packet_types[self.pkt_index]]
> + return test_packet_types
> +
> + def generate_options(self, port_id, pci, intf, types):
> + port_types = ["port=%d," % port_id,
> + "device_id=%s," % pci]
> + dump_pcap_types = [
> + ["tx-dev={0},rx-dev={1},".format(self.tx_pcap, self.rx_pcap),
> + {"rx": [self.rx_pcap, 1],
> + "tx": [self.tx_pcap, 1]}],
> + ["rx-dev={0},".format(self.rx_pcap),
> + {"rx": [self.rx_pcap, 1],
> + "tx": [None, 0]}],
> + ["tx-dev={0},".format(self.tx_pcap),
> + {"rx": [None, 0],
> + "tx": [self.tx_pcap, 1]}], ]
> + dump_iface_types = [
> + ["tx-dev={0},rx-dev={1},".format(self.tx_iface, self.rx_iface),
> + {"rx": [self.rx_pcap, 1],
> + "tx": [self.tx_pcap, 1]}],
> + ["rx-dev={0},".format(self.rx_iface),
> + {"rx": [self.rx_pcap, 1],
> + "tx": [None, 0]}],
> + ["tx-dev={0},".format(self.tx_iface),
> + {"rx": [None, 0],
> + "tx": [self.tx_pcap, 1]}]]
> +
> + queue_types = ["queue=*,", "queue=0,"]
> + # ring size
> + maxPower = 27
> + minPower = 1
> + ring_size_types = [
> + "ring-size=%d," % (2 ** minPower),
> + "ring-size=%d," % (2 ** random.randint(minPower + 1,
> maxPower)),
> + "ring-size=%d," % (2 ** maxPower)]
> + # mbuf size
> + max_mbuf_size = 50000
> + min_mbuf_size = 252
> + mbuf_size_types = [
> + "mbuf-size=%d," % min_mbuf_size,
> + "mbuf-size=%d," % random.randint(min_mbuf_size + 1,
> max_mbuf_size),
> + "mbuf-size=%d," % max_mbuf_size]
> + # total number mbuf
> + max_total_num_mbufs = 65535
> + min_total_num_mbufs = 1025
> + total_num_mbufs_types = [
> + "total-num-mbufs=%d," % min_total_num_mbufs,
> + "total-num-mbufs=%d," %
> random.randint(min_total_num_mbufs + 1,
> +
> max_total_num_mbufs),
> + "total-num-mbufs=%d," % max_total_num_mbufs]
> +
> + port_num = len(port_types) if "port" in types else 1
> +
> + if "dev-pcap" in types:
> + dump_types = dump_pcap_types
> + elif "dev-iface" in types:
> + dump_types = dump_iface_types
> + else:
> + dump_types = dump_pcap_types[:1]
> +
> + queue_num = len(queue_types) if "queue" in types else 1
> +
> + option_exds = [""]
> + if "ring_size" in types:
> + option_exds = ring_size_types[:]
> +
> + if "mbuf_size" in types:
> + option_exds = mbuf_size_types[:]
> +
> + if "total_num_mbufs" in types:
> + option_exds = total_num_mbufs_types[:]
> +
> + options = list()
> + for port in port_types[:port_num]:
> + for queue in queue_types[:queue_num]:
> + for dump in dump_types:
> + for option_exd in option_exds:
> + opt = ((port + queue + dump[0]) + option_exd)[:-1]
> + msg = 'command line option string should be <= 256'
> + self.verify(len(opt) <= 256, msg)
> + options.append([opt, dump[1]])
> +
> + return options
> +
> + def compare_pkts(self, refPkt=None, targetPkt=None, pkt_type=None,
> + refPacketNo=0, targetPacketNo=0):
> + ''' compare two pcap files packet content '''
> + warning = None
> + refObj = parsePacket(refPkt)
> + warning = refObj.parse_pcap(number=refPacketNo)
> + if warning:
> + return warning
> + targetObj = parsePacket(targetPkt)
> + warning = targetObj.parse_pcap(number=targetPacketNo)
> + if warning:
> + return warning
> + # remove some fields, which are filled by dpdk automatically
> + # if packet is filled with `Padding`, remove this
> + if "Padding" in targetObj.packetLayers.keys():
> + targetObj.packetLayers.pop("Padding")
> + if len(refObj.packetLayers) != len(targetObj.packetLayers):
> + refObj_layer = pformat(refObj.packetLayers)
> + targetObj_layer = pformat(targetObj.packetLayers)
> + self.logger.info(os.linesep + "refObj: %s" % refObj_layer)
> + self.logger.info(os.linesep + "targetObj: %s" % targetObj_layer)
> + warning = "packet {0} layers are not as
> expected".format(targetPkt)
> + return warning
> +
> + for layer in refObj.packetLayers.keys():
> + if layer not in targetObj.packetLayers.keys():
> + warning = "{0} has no [{1}] layer".format(targetPkt, layer)
> + return warning
> +
> + if layer == 'Raw':
> + continue
> +
> + refLayerFields = refObj.packetLayers[layer]
> + targetLayerFields = targetObj.packetLayers[layer]
> + if len(refLayerFields) != len(targetLayerFields):
> + warning = "{0} [{1}] layer has no expected fields".format(
> + targetPkt, layer)
> + return warning
> +
> + for field in refLayerFields.keys():
> + if field == 'src' or field == 'dst':
> + continue
> + if field not in targetLayerFields.keys():
> + warning = ("{0} layer [{1}] "
> + "has no [{2}] field").format(
> + targetPkt, layer, field)
> + return warning
> + if refLayerFields[field] != targetLayerFields[field]:
> + warning = ("{0} [{1}] layer [{2}] "
> + "field has no expected value").format(
> + targetPkt, layer, field)
> + return warning
> +
> + return warning
> +
> + def clear_ASLR(self):
> + cmd = "echo 0 > /proc/sys/kernel/randomize_va_space"
> + self.dut.alt_session.send_expect(cmd, "# ", timeout=10)
> + time.sleep(2)
> +
> + def reset_ASLR(self):
> + cmd = "echo 2 > /proc/sys/kernel/randomize_va_space"
> + self.dut.alt_session.send_expect(cmd, "# ", timeout=10)
> + time.sleep(4)
> +
> + def start_testpmd(self):
> + self.dut.alt_session.send_expect(
> + "rm -fr {0}/*".format(self.pdump_log), "# ", 10)
> + param_opt = "--port-topology=chained"
> + eal_param = '--file-prefix=test'
> + self.testpmd.start_testpmd("Default", param=param_opt,
> + eal_param=eal_param)
> + self.testpmd.execute_cmd("set fwd io")
> + self.testpmd.execute_cmd("start")
> + time.sleep(2)
> +
> + def stop_testpmd(self):
> + self.testpmd.execute_cmd("stop")
> + self.testpmd.quit()
> + time.sleep(2)
> +
> + def start_tcpdump_iface(self, option):
> + if option["rx"][0] is not None and option["tx"][0] is not None and \
> + option["rx"][0] == option["tx"][0]:
> + if self.is_existed_on_dut(self.rxtx_pcap):
> + self.dut.alt_session.send_expect(
> + "rm -f %s" % self.rxtx_pcap, "# ")
> + cmd = self.tcpdump.format(self.rxtx_iface, self.rxtx_pcap)
> + self.session_ex.send_expect(cmd, "# ")
> + else:
> + if option["rx"][0] is not None:
> + if self.is_existed_on_dut(self.rx_pcap):
> + self.dut.alt_session.send_expect(
> + "rm -f %s" % self.rx_pcap, "# ")
> + cmd = self.tcpdump.format(self.rx_iface, self.rx_pcap)
> + self.session_ex.send_expect(cmd, "# ")
> +
> + if option["tx"][0] is not None:
> + if self.is_existed_on_dut(self.tx_pcap):
> + self.dut.alt_session.send_expect(
> + "rm -f %s" % self.tx_pcap, "# ")
> + cmd = self.tcpdump.format(self.tx_iface, self.tx_pcap)
> + self.session_ex.send_expect(cmd, "# ")
> + time.sleep(4)
> +
> + def stop_tcpdump_iface(self):
> + self.dut.alt_session.send_expect("killall tcpdump", "# ", 5)
> + time.sleep(2)
> +
> + def start_dpdk_pdump(self, option):
> + length_limit = 256
> + msg = ('pdump option string length should be less than {}'
> + ).format(length_limit)
> + self.verify(len(option) < length_limit, msg)
> + self.dut.alt_session.send_expect(
> + "rm -fr {0}/*".format(self.pdump_log), "# ", 20)
> + cmd = ';'.join([
> + "cd %s" % self.dut.base_dir,
> + self.dpdk_pdump + " '%s' >/dev/null 2>&1 &" % (option[0])])
> + self.session_ex.send_expect(cmd, "# ", 15)
> + time.sleep(6)
> +
> + def check_pdump_ready(self, option):
> + rx_dump_pcap = option["rx"][0]
> + if rx_dump_pcap:
> + self.verify(self.is_existed_on_dut(rx_dump_pcap),
> + "{1} {0} is not ready".format(rx_dump_pcap,
> + self.tool_name))
> + tx_dump_pcap = option["tx"][0]
> + if tx_dump_pcap:
> + self.verify(self.is_existed_on_dut(tx_dump_pcap),
> + "{1} {0} is not ready".format(tx_dump_pcap,
> + self.tool_name))
> +
> + def stop_dpdk_pdump(self):
> + self.dut.alt_session.send_expect(
> + "killall %s" % self.tool_name, "# ", 5)
> + time.sleep(2)
> +
> + def packet_capture_dump_packets(self, pkt_type, number, **kwargs):
> + self.logger.info('send <{}> packet'.format(pkt_type))
> + if pkt_type == 'VLAN_UDP':
> + self.testpmd.execute_cmd("vlan set filter off 0")
> + self.testpmd.execute_cmd("vlan set filter off 1")
> + self.testpmd.execute_cmd("start")
> + # tester's port 0 and port 1 work under chained mode.
> + port_0 = self.dut_ports[self.test_port0_id]
> + port_1 = self.dut_ports[self.test_port1_id]
> + # check send tx packet by port 0
> + # send packet to dut and compare dpdk-pdump dump pcap with
> + # scapy pcap file
> + intf = self.tester.get_interface(self.tester.get_local_port(port_1))
> + # prepare to catch replay packet in out port
> + recPkt = os.path.join('/tmp', "sniff_%s.pcap" % intf)
> + if os.path.exists(recPkt):
> + os.remove(recPkt)
> + index = sniff_packets(intf, count=1, timeout=20, pcap=recPkt)
> + pkt = Packet(pkt_type=pkt_type)
> + if pkt_type == 'VLAN_UDP':
> + pkt.config_layer('dot1q', {'vlan': 20})
> + src_mac = self.tester.get_mac(self.tester.get_local_port(port_0))
> + pkt.config_layer('ether', {'src': src_mac})
> + # save send packet in a pcap file
> + refPkt = self.send_pcap % (pkt_type, "rx", number)
> + if os.path.exists(refPkt):
> + os.remove(refPkt)
> + pkt.pktgen.write_pcap(refPkt)
> + # send out test packet
> + tester_port = self.tester.get_local_port(port_0)
> + intf = self.tester.get_interface(tester_port)
> + pkt.send_pkt(tx_port=intf)
> + # load pcap file caught by out port
> + time.sleep(1)
> + load_sniff_packets(index)
> + # compare pcap file received by out port with scapy reference
> + # packet pcap file
> + warning = self.compare_pkts(refPkt, recPkt, pkt_type)
> + msg = "tcpdump rx Receive Packet error: {0}".format(warning)
> + self.verify(not warning, msg)
> + # check send tx packet by port 1
> + # send packet to dut and compare dpdk-pdump dump pcap
> + # with scapy pcap file
> + intf = self.tester.get_interface(self.tester.get_local_port(port_0))
> + # prepare to catch replay packet in out port
> + recPkt = os.path.join('/tmp', "sniff_%s.pcap" % intf)
> + if os.path.exists(recPkt):
> + os.remove(recPkt)
> + index = sniff_packets(intf, count=1, timeout=20, pcap=recPkt)
> + pkt = Packet(pkt_type=pkt_type)
> + if pkt_type == 'VLAN_UDP':
> + pkt.config_layer('dot1q', {'vlan': 20})
> + src_mac = self.tester.get_mac(self.tester.get_local_port(port_1))
> + pkt.config_layer('ether', {'src': src_mac})
> + # save send packet in a pcap file
> + refPkt = self.send_pcap % (pkt_type, "tx", number)
> + if os.path.exists(refPkt):
> + os.remove(refPkt)
> + pkt.pktgen.write_pcap(refPkt)
> + # send out test packet
> + tester_port = self.tester.get_local_port(port_1)
> + intf = self.tester.get_interface(tester_port)
> + pkt.send_pkt(tx_port=intf)
> + # load pcap file caught by out port
> + time.sleep(1)
> + load_sniff_packets(index)
> + # compare pcap file received by out port
> + # with scapy reference packet pcap file
> + warning = self.compare_pkts(refPkt, recPkt, pkt_type)
> + msg = "tcpdump tx Receive Packet error: {0}".format(warning)
> + self.verify(not warning, msg)
> +
> + def check_pdump_pcaps(self, pkt_type, number, **kwargs):
> + rx_dump_pcap = kwargs["rx"][0]
> + if rx_dump_pcap:
> + self.verify(os.path.exists(rx_dump_pcap),
> + "{0} doesn't exist".format(rx_dump_pcap))
> + # save send packet in a pcap file
> + refPkt = self.send_pcap % (pkt_type, "rx", number)
> + self.verify(os.path.exists(refPkt),
> + "{0} doesn't exist".format(refPkt))
> + # compare pcap file dumped by dpdk-pdump
> + # with scapy reference packet pcap file
> + warning = self.compare_pkts(refPkt, rx_dump_pcap, pkt_type,
> +
> targetPacketNo=self.rx_packet_pos)
> + if kwargs["rx"][1] == 2:
> + self.rx_packet_pos += kwargs["rx"][1]
> + else:
> + self.rx_packet_pos += 1
> + self.verify(warning is None,
> + "dpdk-pdump rx dump packet error:
> {0}".format(warning))
> + msg = "pdump rx {0} packet content is correct".format(pkt_type)
> + self.logger.info(msg)
> + tx_dump_pcap = kwargs["tx"][0]
> + if tx_dump_pcap:
> + self.verify(os.path.exists(tx_dump_pcap),
> + "{0} doesn't exist".format(tx_dump_pcap))
> + # set send packet
> + refPkt = self.send_pcap % (pkt_type, "tx", number)
> + self.verify(os.path.exists(refPkt),
> + "{0} doesn't exist".format(refPkt))
> + # compare pcap file dumped by dpdk-pdump
> + # with scapy reference packet pcap file
> + if kwargs["tx"][1] == 2 and self.tx_packet_pos == 0:
> + self.tx_packet_pos = 1
> + warning = self.compare_pkts(refPkt, tx_dump_pcap, pkt_type,
> +
> targetPacketNo=self.tx_packet_pos)
> + if kwargs["tx"][1] == 2:
> + self.tx_packet_pos += kwargs["tx"][1]
> + else:
> + self.tx_packet_pos += 1
> + msg = "dpdk-pdump tx dump packet error: {0}".format(warning)
> + self.verify(warning is None, msg)
> + msg = "pdump tx {0} packet content is correct".format(pkt_type)
> + self.logger.info(msg)
> +
> + def packet_capture_test_packets(self, option):
> + self.clear_ASLR()
> + self.start_testpmd()
> + self.start_dpdk_pdump(option)
> + if self.dev_iface_flag:
> + self.start_tcpdump_iface(option[1])
> + self.check_pdump_ready(option[1])
> + for number, packet_type in enumerate(self.get_packet_types()):
> + self.packet_capture_dump_packets(packet_type, number,
> **option[1])
> + time.sleep(2)
> + if self.dev_iface_flag:
> + self.stop_tcpdump_iface()
> + if self.dut.get_ip_address() != self.tester.get_ip_address():
> + # copy rx pdump data from dut
> + if self.is_existed_on_dut(self.rx_pcap):
> + if os.path.exists(self.rx_pcap):
> + os.remove(self.rx_pcap)
> + self.dut.session.copy_file_from(self.rx_pcap, self.rx_pcap)
> + # copy tx pdump data from dut
> + if self.is_existed_on_dut(self.tx_pcap):
> + if os.path.exists(self.tx_pcap):
> + os.remove(self.tx_pcap)
> + self.dut.session.copy_file_from(self.tx_pcap, self.tx_pcap)
> + self.stop_dpdk_pdump()
> + self.stop_testpmd()
> + self.reset_ASLR()
> +
> + self.rx_packet_pos = 0
> + self.tx_packet_pos = 0
> + for number, packet_type in enumerate(self.get_packet_types()):
> + self.check_pdump_pcaps(packet_type, number, **option[1])
> + time.sleep(2)
> +
> + def packet_capture_test_options(self, pdump_options):
> + try:
> + for option in pdump_options:
> + self.packet_capture_test_packets(option)
> + self.exit_flag = True
> + except Exception as e:
> + self.logger.error(traceback.format_exc())
> + raise VerifyFailure(e)
> +
> + def test_pdump_port(self):
> + '''
> + test different port options::
> + *. port=<dut port id>
> + *. device_id=<dut pci address>
> + '''
> + port_id = self.test_port0_id
> + port_name = self.dut.ports_info[port_id]['intf']
> + port_pci = self.dut.ports_info[port_id]['pci']
> + test_types = ["port"]
> + options = self.generate_options(
> + port_id, port_pci, port_name, test_types)
> + self.packet_capture_test_options(options)
> +
> + def test_pdump_dev_pcap(self):
> + '''
> + test different dump options with pcap files as output::
> + *. tx-dev=/xxx/pdump-tx.pcap,rx-dev=/xxx/pdump-rx.pcap
> + *. rx-dev=/xxx/pdump-rx.pcap
> + *. tx-dev=/xxx/pdump-tx.pcap
> + '''
> + port_id = self.test_port0_id
> + port_name = self.dut.ports_info[port_id]['intf']
> + port_pci = self.dut.ports_info[port_id]['pci']
> + test_types = ["dev-pcap"]
> + options = self.generate_options(
> + port_id, port_pci, port_name, test_types)
> + self.packet_capture_test_options(options)
> +
> + def test_pdump_dev_iface(self):
> + '''
> + test different dump options with interfaces as output::
> + *. tx-dev=<dut tx port name>,rx-dev=<dut rx port name>
> + *. rx-dev=<dut rx port name>
> + *. tx-dev=<dut tx port name>
> + '''
> + self.get_dut_iface_with_kernel_driver()
> + port_id = self.test_port0_id
> + port_name = self.dut.ports_info[port_id]['intf']
> + port_pci = self.dut.ports_info[port_id]['pci']
> + test_types = ["dev-iface"]
> + self.dev_iface_flag = True
> + options = self.generate_options(
> + port_id, port_pci, port_name, test_types)
> + self.packet_capture_test_options(options)
> +
> + def test_pdump_queue(self):
> + '''
> + test different queue options::
> + *. first queue:
> + queue=0
> + *. all:
> + queue=*
> + '''
> + port_id = self.test_port0_id
> + port_name = self.dut.ports_info[port_id]['intf']
> + port_pci = self.dut.ports_info[port_id]['pci']
> + test_types = ["queue"]
> + options = self.generate_options(
> + port_id, port_pci, port_name, test_types)
> + self.packet_capture_test_options(options)
> +
> + def test_pdump_ring_size(self):
> + '''
> + test ring size option, set value within 2^[1~27]
> + '''
> + port_id = self.test_port0_id
> + port_name = self.dut.ports_info[port_id]['intf']
> + port_pci = self.dut.ports_info[port_id]['pci']
> + test_types = ["ring_size"]
> + options = self.generate_options(
> + port_id, port_pci, port_name, test_types)
> + self.packet_capture_test_options(options)
> +
> + def test_pdump_mbuf_size(self):
> + '''
> + test mbuf size option, set value within [252~50000].
> + min value is decided by single packet size,
> + max value is decided by test platform memory size.
> + '''
> + port_id = self.test_port0_id
> + port_name = self.dut.ports_info[port_id]['intf']
> + port_pci = self.dut.ports_info[port_id]['pci']
> + test_types = ["mbuf_size"]
> +
> + options = self.generate_options(
> + port_id, port_pci, port_name, test_types)
> + self.packet_capture_test_options(options)
> +
> + def test_pdump_total_num_mbufs(self):
> + '''
> + test total-num-mbufs option, set value within [1025~65535]
> + '''
> + port_id = self.test_port0_id
> + port_name = self.dut.ports_info[port_id]['intf']
> + port_pci = self.dut.ports_info[port_id]['pci']
> + test_types = ["total_num_mbufs"]
> + options = self.generate_options(
> + port_id, port_pci, port_name, test_types)
> + self.packet_capture_test_options(options)
> +
> + def set_up_all(self):
> + '''
> + Run at the start of each test suite.
> + '''
> + self.verify(self.target == "x86_64-native-linuxapp-gcc",
> + "only support x86_64-native-linuxapp-gcc")
> + self.dut_ports = self.dut.get_ports()
> + self.verify(len(self.dut_ports) == 2, "Insufficient ports for testing")
> + self.verify_supported_nic()
> + self.test_port0_id = 0
> + self.test_port1_id = 1
> + # compile dpdk app with SW CONFIG_RTE_LIBRTE_PMD_PCAP open
> + self.dut_skip_compile = self.dut.skip_setup
> + # used for save log
> + self.pdump_log = os.sep.join(["/root", 'pdumpLog'])
> + if not self.is_existed_on_dut(self.pdump_log):
> + cmd = "mkdir -p {0}".format(self.pdump_log)
> + self.dut.alt_session.send_expect(cmd, "# ")
> + if not (self.dut_skip_compile and self.check_pcap_lib()):
> + self.pcap_SW = "CONFIG_RTE_LIBRTE_PMD_PCAP"
> + cmd = "sed -i -e 's/{0}=n$/{0}=y/' config/common_base".format(
> + self.pcap_SW)
> + self.dut.alt_session.send_expect(cmd, "# ", 30)
> + self.dut.skip_setup = False
> + self.dut.build_install_dpdk(self.target)
> + self.session_ex = self.dut.new_session()
> + # secondary process (dpdk-pdump)
> + self.pdump_dir = self.dut.base_dir + os.sep + \
> + "%s/build/app/pdump/" % self.target
> + cmd = "ls {0} -F | grep '*'".format(self.pdump_dir)
> + exe_files = self.dut.alt_session.send_expect(cmd, "# ").splitlines()
> + if len(exe_files) == 1:
> + self.tool_name = exe_files[0][:-1]
> + else:
> + self.verify(False, "tool name exception !")
> + self.dut_dpdk_pdump_dir = self.dut.base_dir + os.sep + \
> + "%s/app/%s" % (self.target, self.tool_name)
> + self.dpdk_pdump = self.dut_dpdk_pdump_dir + \
> + " -v --file-prefix=test -- --pdump "
> + self.send_pcap = os.sep.join([self.pdump_log,
> "scapy_%s_%s_%d.pcap"])
> + self.rx_pcap = os.sep.join([self.pdump_log, "pdump-rx.pcap"])
> + self.tx_pcap = os.sep.join([self.pdump_log, "pdump-tx.pcap"])
> + self.rxtx_pcap = os.sep.join([self.pdump_log, "pdump-rxtx.pcap"])
> + self.rx_iface = None
> + self.tx_iface = None
> + self.rxtx_iface = None
> + self.rx_packet_pos = 0
> + self.tx_packet_pos = 0
> + self.dev_iface_flag = False
> + # primary process
> + self.testpmd = PmdOutput(self.dut)
> + # False: reduce test items for regression testing,
> + # shut off base test environment checking
> + # True: make a full range testing
> + self.full_test = True
> + # used on daily testing to avoid long time running
> + self.pkt_index = None
> + # get tcpdump options
> + self.get_tcpdump_options()
> +
> + def set_up(self):
> + '''
> + Run before each test case.
> + '''
> + self.exit_flag = False
> +
> + def tear_down(self):
> + '''
> + Run after each test case.
> + '''
> + if not self.exit_flag:
> + self.stop_dpdk_pdump()
> + self.dut.alt_session.send_expect("killall testpmd", "# ")
> + self.dut.alt_session.send_expect("killall tcpdump", "# ")
> + self.reset_ASLR()
> + if self.dev_iface_flag:
> + self.stop_tcpdump_iface()
> + self.dev_iface_flag = False
> +
> + def tear_down_all(self):
> + '''
> + Run after each test suite.
> + '''
> + if self.session_ex:
> + self.reset_ASLR()
> + self.session_ex.close()
> + self.session_ex = None
> + # Restore the config file and recompile the package.
> + if not (self.dut_skip_compile and self.check_pcap_lib()):
> +
> #############################################################
> ######
> + self.dut.alt_session.send_expect(
> + ("sed -i -e 's/{0}=y$/{0}=n/' "
> + "config/common_base").format(self.pcap_SW), "# ", 120)
> + # temporary disable skip_setup
> + skip_setup = self.dut.skip_setup
> + self.dut.skip_setup = True
> + self.dut.build_install_dpdk(self.target)
> + self.dut.skip_setup = skip_setup
> +
> #############################################################
> ######
> + self.dut.kill_all()
> --
> 1.9.3
^ permalink raw reply [flat|nested] 5+ messages in thread* Re: [dts] [PATCH V4]tests/packet_capture: upload automation script
2019-07-30 5:53 ` yufengmx
2019-07-30 12:18 ` Wang, Yinan
@ 2019-08-08 5:48 ` Tu, Lijuan
1 sibling, 0 replies; 5+ messages in thread
From: Tu, Lijuan @ 2019-08-08 5:48 UTC (permalink / raw)
To: Mo, YufengX, dts; +Cc: Mo, YufengX
Applied, thanks
> -----Original Message-----
> From: dts [mailto:dts-bounces@dpdk.org] On Behalf Of yufengmx
> Sent: Tuesday, July 30, 2019 1:54 PM
> To: dts@dpdk.org
> Cc: Mo, YufengX <yufengx.mo@intel.com>
> Subject: [dts] [PATCH V4]tests/packet_capture: upload automation script
>
>
> Packet capture framework feature support packet capturing on dpdk
> Ethernet devices. DPDK provides dpdk-pdump tool under app/pdump
> directory for packet capturing on dpdk.
>
> Signed-off-by: yufengmx <yufengx.mo@intel.com>
> ---
> tests/TestSuite_packet_capture.py | 869
> ++++++++++++++++++++++++++++++++++++++
> 1 file changed, 869 insertions(+)
> create mode 100644 tests/TestSuite_packet_capture.py
>
> diff --git a/tests/TestSuite_packet_capture.py
> b/tests/TestSuite_packet_capture.py
> new file mode 100644
> index 0000000..4bab902
> --- /dev/null
> +++ b/tests/TestSuite_packet_capture.py
> @@ -0,0 +1,869 @@
> +# BSD LICENSE
> +#
> +# Copyright(c) 2010-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.
> +'''
> +DPDK Test suite.
> +Test support dpdk pdump tool features
> +'''
> +import random
> +import os
> +import traceback
> +import time
> +import re
> +import signal
> +import subprocess
> +from pprint import pformat
> +
> +from scapy.utils import rdpcap
> +from scapy.packet import NoPayload
> +from scapy.packet import Packet as scapyPacket from scapy.fields import
> +ConditionalField
> +
> +from test_case import TestCase
> +from exception import VerifyFailure
> +from pmd_output import PmdOutput
> +from packet import Packet
> +
> +
> +# These source code copy from packet.py module before
> +sniff_packets/load_sniff_packets # refactor. New refactor methods have
> +much more longer time consumption than # old methods.
> +
> +# Saved back groud sniff process id
> +SNIFF_PIDS = {}
> +
> +
> +def sniff_packets(intf, count=0, timeout=5, pcap=None):
> + """
> + sniff all packets for certain port in certain seconds.
> + """
> + param = ""
> + direct_param = r"(\s+)\[ -(\w) in\|out\|inout \]"
> + tcpdump_help = subprocess.check_output(
> + "tcpdump -h; echo 0", stderr=subprocess.STDOUT, shell=True)
> + for line in tcpdump_help.split('\n'):
> + m = re.match(direct_param, line)
> + if m:
> + param = "-" + m.group(2) + " in"
> +
> + if len(param) == 0:
> + print "tcpdump not support direction choice!!!"
> +
> + sniff_cmd = 'tcpdump -i %(INTF)s %(IN_PARAM)s -w %(FILE)s'
> + options = {'INTF': intf, 'COUNT': count, 'IN_PARAM': param,
> + 'FILE': '/tmp/sniff_%s.pcap' % intf if not pcap else pcap}
> + if count:
> + sniff_cmd += ' -c %(COUNT)d'
> + cmd = sniff_cmd % options
> + else:
> + cmd = sniff_cmd % options
> +
> + args = cmd.split()
> + pipe = subprocess.Popen(args)
> + index = str(time.time())
> + SNIFF_PIDS[index] = (pipe, intf, timeout)
> + time.sleep(0.5)
> + return index
> +
> +
> +def load_sniff_packets(index=''):
> + pkts = []
> + child_exit = False
> + if index in SNIFF_PIDS.keys():
> + pipe, intf, timeout = SNIFF_PIDS[index]
> + time_elapse = int(time.time() - float(index))
> + while time_elapse < timeout:
> + if pipe.poll() is not None:
> + child_exit = True
> + break
> +
> + time.sleep(1)
> + time_elapse += 1
> +
> + if not child_exit:
> + pipe.send_signal(signal.SIGINT)
> + pipe.wait()
> +
> + # wait pcap file ready
> + time.sleep(1)
> + try:
> + cap_pkts = rdpcap("/tmp/sniff_%s.pcap" % intf)
> + for pkt in cap_pkts:
> + # packet gen should be scapy
> + packet = Packet(tx_port=intf)
> + packet.pktgen.assign_pkt(pkt)
> + pkts.append(packet)
> + except Exception as e:
> + pass
> +
> + return pkts
> +
> +
> +class parsePacket(object):
> +
> + def __init__(self, filename, skip_flag=False):
> + self.pcapFile = filename
> + self.packetLayers = dict()
> + self.skip = skip_flag
> +
> + def parse_packet_layer(self, pkt_object):
> + if pkt_object is None:
> + return
> + self.packetLayers[pkt_object.name] = dict()
> + for curfield in pkt_object.fields_desc:
> + if isinstance(curfield, ConditionalField) and \
> + not curfield._evalcond(pkt_object):
> + continue
> + field_value = pkt_object.getfieldval(curfield.name)
> + if isinstance(field_value, scapyPacket) or \
> + (curfield.islist and
> + curfield.holds_packets and
> + type(field_value) is list):
> + continue
> + repr_value = curfield.i2repr(pkt_object, field_value)
> + if isinstance(repr_value, str):
> + repr_value = repr_value.replace(
> + os.linesep, os.linesep + " " * (len(curfield.name) + 4))
> + self.packetLayers[pkt_object.name][curfield.name] = repr_value
> + if not isinstance(pkt_object.payload, NoPayload):
> + self.parse_packet_layer(pkt_object.payload)
> +
> + def get_valid_packet(self, pcap_pkts_origin, number):
> + cnt = 0
> + for packet in pcap_pkts_origin:
> + self.parse_packet_layer(packet)
> + flag = 'LLDP' not in self.packetLayers.keys() \
> + if self.skip else \
> + True
> + src_ether = self.packetLayers["Ethernet"]['src']
> + if src_ether == '00:00:00:00:00:00' and flag:
> + if number == cnt:
> + break
> + cnt += 1
> + self.packetLayers.clear()
> +
> + def parse_pcap(self, number=0):
> + pcap_pkts = []
> + try:
> + if not os.path.exists(self.pcapFile):
> + warning = "{0} is not exist !".format(self.pcapFile)
> + return warning
> + pcap_pkts = rdpcap(self.pcapFile)
> + if len(pcap_pkts) == 0:
> + warning = "{0} is empty".format(self.pcapFile)
> + return warning
> + elif number >= len(pcap_pkts):
> + warning = "{0} is missing No.{1} packet".format(
> + self.pcapFile, number)
> + return warning
> + self.get_valid_packet(pcap_pkts, number)
> + except Exception, e:
> + print (e)
> +
> + return None
> +
> +
> +class TestPacketCapture(TestCase):
> +
> + def is_existed_on_dut(self, check_path):
> + self.dut.alt_session.send_expect("ls %s" % check_path, "# ")
> + cmd = "echo $?"
> + output = self.dut.alt_session.send_expect(cmd, "# ")
> + ret = True if output and output.strip() == "0" else False
> + return ret
> +
> + def get_dut_iface_with_kernel_driver(self):
> + # only physical nic support PROMISC
> + cmd = "ip link show | grep PROMISC | awk {'print $2'}"
> + out = self.dut.alt_session.send_expect(cmd, "# ")
> + pat = "(.*):"
> + ifaces = [intf for intf in re.findall(pat, out, re.M) if intf]
> + # get dut/tester ip to check if they are in a platform
> + tester_ip = self.tester.get_ip_address()
> + dut_ip = self.dut.get_ip_address()
> + for link_port in range(len(self.dut_ports)):
> + # if they are in a platform, ignore interface used by tester
> + if tester_ip == dut_ip:
> + tester_port = self.tester.get_local_port(link_port)
> + intf = self.tester.get_interface(tester_port)
> + if intf in ifaces:
> + ifaces.remove(intf)
> + # ignore interface used by dut
> + intf = self.dut.ports_info[link_port]['intf']
> + if intf in ifaces:
> + ifaces.remove(intf)
> +
> + # set ports up
> + tmp_ifaces = ifaces[:]
> + for iface in tmp_ifaces:
> + # ignore current interface used by system
> + cmd = "ifconfig %s | grep 'inet ' " % iface
> + if self.dut.alt_session.send_expect(cmd, "# ") != "":
> + ifaces.remove(iface)
> + self.dut.alt_session.send_expect(
> + "ifconfig {0} up".format(iface), "# ")
> + time.sleep(10)
> + # get ports on link status
> + tmp_ifaces = ifaces[:]
> + for iface in tmp_ifaces:
> + cmd = "ip link show {0} | grep LOWER_UP".format(iface)
> + self.dut.alt_session.send_expect(cmd, "# ")
> + output = self.dut.alt_session.send_expect(
> + "echo $?".format(iface), "# ").strip()
> + if output != '0':
> + ifaces.remove(iface)
> +
> + self.verify(len(ifaces) >= 2, "Insufficient ports for iface testing")
> + # set iface port for pdump tool output dump packets
> + self.rx_iface = ifaces[0]
> + self.tx_iface = ifaces[1]
> + self.rxtx_iface = ifaces[0]
> +
> + def verify_supported_nic(self):
> + supported_drivers = ['i40e', 'ixgbe']
> + result = all([self.dut.ports_info[index]['port'].default_driver in
> + supported_drivers
> + for index in self.dut_ports])
> + msg = "current nic is not supported"
> + self.verify(result, msg)
> +
> + def get_tcpdump_options(self):
> + param = ""
> + direct_param = r"(\s+)\[ -(\w) in\|out\|inout \]"
> + tcpdump_help = self.dut.alt_session.send_expect("tcpdump -h", "# ")
> + for line in tcpdump_help.split('\n'):
> + m = re.match(direct_param, line)
> + if m:
> + param = "-" + m.group(2) + " out"
> + self.tcpdump = "tcpdump -i {0} " + param + " -w {1} >/dev/null 2>&1 &"
> +
> + def check_pcap_lib(self):
> + pcap_lib_dir = os.sep.join([self.dut.base_dir,
> + self.target,
> + "lib/librte_pmd_pcap.a"])
> + return self.is_existed_on_dut(pcap_lib_dir)
> +
> + def get_packet_types(self):
> + packet_types = ["TCP",
> + "UDP",
> + "SCTP",
> + "TIMESYNC",
> + "ARP",
> + "LLDP",
> + "IP_RAW",
> + "IPv6_TCP",
> + "IPv6_UDP",
> + "IPv6_SCTP",
> + "VLAN_UDP"]
> + if self.pkt_index is None: # only initialize once
> + self.pkt_index = random.randint(0, len(packet_types) - 1)
> + test_packet_types = packet_types if self.full_test else \
> + [packet_types[self.pkt_index]]
> + return test_packet_types
> +
> + def generate_options(self, port_id, pci, intf, types):
> + port_types = ["port=%d," % port_id,
> + "device_id=%s," % pci]
> + dump_pcap_types = [
> + ["tx-dev={0},rx-dev={1},".format(self.tx_pcap, self.rx_pcap),
> + {"rx": [self.rx_pcap, 1],
> + "tx": [self.tx_pcap, 1]}],
> + ["rx-dev={0},".format(self.rx_pcap),
> + {"rx": [self.rx_pcap, 1],
> + "tx": [None, 0]}],
> + ["tx-dev={0},".format(self.tx_pcap),
> + {"rx": [None, 0],
> + "tx": [self.tx_pcap, 1]}], ]
> + dump_iface_types = [
> + ["tx-dev={0},rx-dev={1},".format(self.tx_iface, self.rx_iface),
> + {"rx": [self.rx_pcap, 1],
> + "tx": [self.tx_pcap, 1]}],
> + ["rx-dev={0},".format(self.rx_iface),
> + {"rx": [self.rx_pcap, 1],
> + "tx": [None, 0]}],
> + ["tx-dev={0},".format(self.tx_iface),
> + {"rx": [None, 0],
> + "tx": [self.tx_pcap, 1]}]]
> +
> + queue_types = ["queue=*,", "queue=0,"]
> + # ring size
> + maxPower = 27
> + minPower = 1
> + ring_size_types = [
> + "ring-size=%d," % (2 ** minPower),
> + "ring-size=%d," % (2 ** random.randint(minPower + 1, maxPower)),
> + "ring-size=%d," % (2 ** maxPower)]
> + # mbuf size
> + max_mbuf_size = 50000
> + min_mbuf_size = 252
> + mbuf_size_types = [
> + "mbuf-size=%d," % min_mbuf_size,
> + "mbuf-size=%d," % random.randint(min_mbuf_size + 1,
> max_mbuf_size),
> + "mbuf-size=%d," % max_mbuf_size]
> + # total number mbuf
> + max_total_num_mbufs = 65535
> + min_total_num_mbufs = 1025
> + total_num_mbufs_types = [
> + "total-num-mbufs=%d," % min_total_num_mbufs,
> + "total-num-mbufs=%d," % random.randint(min_total_num_mbufs +
> 1,
> + max_total_num_mbufs),
> + "total-num-mbufs=%d," % max_total_num_mbufs]
> +
> + port_num = len(port_types) if "port" in types else 1
> +
> + if "dev-pcap" in types:
> + dump_types = dump_pcap_types
> + elif "dev-iface" in types:
> + dump_types = dump_iface_types
> + else:
> + dump_types = dump_pcap_types[:1]
> +
> + queue_num = len(queue_types) if "queue" in types else 1
> +
> + option_exds = [""]
> + if "ring_size" in types:
> + option_exds = ring_size_types[:]
> +
> + if "mbuf_size" in types:
> + option_exds = mbuf_size_types[:]
> +
> + if "total_num_mbufs" in types:
> + option_exds = total_num_mbufs_types[:]
> +
> + options = list()
> + for port in port_types[:port_num]:
> + for queue in queue_types[:queue_num]:
> + for dump in dump_types:
> + for option_exd in option_exds:
> + opt = ((port + queue + dump[0]) + option_exd)[:-1]
> + msg = 'command line option string should be <= 256'
> + self.verify(len(opt) <= 256, msg)
> + options.append([opt, dump[1]])
> +
> + return options
> +
> + def compare_pkts(self, refPkt=None, targetPkt=None, pkt_type=None,
> + refPacketNo=0, targetPacketNo=0):
> + ''' compare two pcap files packet content '''
> + warning = None
> + refObj = parsePacket(refPkt)
> + warning = refObj.parse_pcap(number=refPacketNo)
> + if warning:
> + return warning
> + targetObj = parsePacket(targetPkt)
> + warning = targetObj.parse_pcap(number=targetPacketNo)
> + if warning:
> + return warning
> + # remove some fields, which are filled by dpdk automatically
> + # if packet is filled with `Padding`, remove this
> + if "Padding" in targetObj.packetLayers.keys():
> + targetObj.packetLayers.pop("Padding")
> + if len(refObj.packetLayers) != len(targetObj.packetLayers):
> + refObj_layer = pformat(refObj.packetLayers)
> + targetObj_layer = pformat(targetObj.packetLayers)
> + self.logger.info(os.linesep + "refObj: %s" % refObj_layer)
> + self.logger.info(os.linesep + "targetObj: %s" % targetObj_layer)
> + warning = "packet {0} layers are not as expected".format(targetPkt)
> + return warning
> +
> + for layer in refObj.packetLayers.keys():
> + if layer not in targetObj.packetLayers.keys():
> + warning = "{0} has no [{1}] layer".format(targetPkt, layer)
> + return warning
> +
> + if layer == 'Raw':
> + continue
> +
> + refLayerFields = refObj.packetLayers[layer]
> + targetLayerFields = targetObj.packetLayers[layer]
> + if len(refLayerFields) != len(targetLayerFields):
> + warning = "{0} [{1}] layer has no expected fields".format(
> + targetPkt, layer)
> + return warning
> +
> + for field in refLayerFields.keys():
> + if field == 'src' or field == 'dst':
> + continue
> + if field not in targetLayerFields.keys():
> + warning = ("{0} layer [{1}] "
> + "has no [{2}] field").format(
> + targetPkt, layer, field)
> + return warning
> + if refLayerFields[field] != targetLayerFields[field]:
> + warning = ("{0} [{1}] layer [{2}] "
> + "field has no expected value").format(
> + targetPkt, layer, field)
> + return warning
> +
> + return warning
> +
> + def clear_ASLR(self):
> + cmd = "echo 0 > /proc/sys/kernel/randomize_va_space"
> + self.dut.alt_session.send_expect(cmd, "# ", timeout=10)
> + time.sleep(2)
> +
> + def reset_ASLR(self):
> + cmd = "echo 2 > /proc/sys/kernel/randomize_va_space"
> + self.dut.alt_session.send_expect(cmd, "# ", timeout=10)
> + time.sleep(4)
> +
> + def start_testpmd(self):
> + self.dut.alt_session.send_expect(
> + "rm -fr {0}/*".format(self.pdump_log), "# ", 10)
> + param_opt = "--port-topology=chained"
> + eal_param = '--file-prefix=test'
> + self.testpmd.start_testpmd("Default", param=param_opt,
> + eal_param=eal_param)
> + self.testpmd.execute_cmd("set fwd io")
> + self.testpmd.execute_cmd("start")
> + time.sleep(2)
> +
> + def stop_testpmd(self):
> + self.testpmd.execute_cmd("stop")
> + self.testpmd.quit()
> + time.sleep(2)
> +
> + def start_tcpdump_iface(self, option):
> + if option["rx"][0] is not None and option["tx"][0] is not None and \
> + option["rx"][0] == option["tx"][0]:
> + if self.is_existed_on_dut(self.rxtx_pcap):
> + self.dut.alt_session.send_expect(
> + "rm -f %s" % self.rxtx_pcap, "# ")
> + cmd = self.tcpdump.format(self.rxtx_iface, self.rxtx_pcap)
> + self.session_ex.send_expect(cmd, "# ")
> + else:
> + if option["rx"][0] is not None:
> + if self.is_existed_on_dut(self.rx_pcap):
> + self.dut.alt_session.send_expect(
> + "rm -f %s" % self.rx_pcap, "# ")
> + cmd = self.tcpdump.format(self.rx_iface, self.rx_pcap)
> + self.session_ex.send_expect(cmd, "# ")
> +
> + if option["tx"][0] is not None:
> + if self.is_existed_on_dut(self.tx_pcap):
> + self.dut.alt_session.send_expect(
> + "rm -f %s" % self.tx_pcap, "# ")
> + cmd = self.tcpdump.format(self.tx_iface, self.tx_pcap)
> + self.session_ex.send_expect(cmd, "# ")
> + time.sleep(4)
> +
> + def stop_tcpdump_iface(self):
> + self.dut.alt_session.send_expect("killall tcpdump", "# ", 5)
> + time.sleep(2)
> +
> + def start_dpdk_pdump(self, option):
> + length_limit = 256
> + msg = ('pdump option string length should be less than {}'
> + ).format(length_limit)
> + self.verify(len(option) < length_limit, msg)
> + self.dut.alt_session.send_expect(
> + "rm -fr {0}/*".format(self.pdump_log), "# ", 20)
> + cmd = ';'.join([
> + "cd %s" % self.dut.base_dir,
> + self.dpdk_pdump + " '%s' >/dev/null 2>&1 &" % (option[0])])
> + self.session_ex.send_expect(cmd, "# ", 15)
> + time.sleep(6)
> +
> + def check_pdump_ready(self, option):
> + rx_dump_pcap = option["rx"][0]
> + if rx_dump_pcap:
> + self.verify(self.is_existed_on_dut(rx_dump_pcap),
> + "{1} {0} is not ready".format(rx_dump_pcap,
> + self.tool_name))
> + tx_dump_pcap = option["tx"][0]
> + if tx_dump_pcap:
> + self.verify(self.is_existed_on_dut(tx_dump_pcap),
> + "{1} {0} is not ready".format(tx_dump_pcap,
> + self.tool_name))
> +
> + def stop_dpdk_pdump(self):
> + self.dut.alt_session.send_expect(
> + "killall %s" % self.tool_name, "# ", 5)
> + time.sleep(2)
> +
> + def packet_capture_dump_packets(self, pkt_type, number, **kwargs):
> + self.logger.info('send <{}> packet'.format(pkt_type))
> + if pkt_type == 'VLAN_UDP':
> + self.testpmd.execute_cmd("vlan set filter off 0")
> + self.testpmd.execute_cmd("vlan set filter off 1")
> + self.testpmd.execute_cmd("start")
> + # tester's port 0 and port 1 work under chained mode.
> + port_0 = self.dut_ports[self.test_port0_id]
> + port_1 = self.dut_ports[self.test_port1_id]
> + # check send tx packet by port 0
> + # send packet to dut and compare dpdk-pdump dump pcap with
> + # scapy pcap file
> + intf = self.tester.get_interface(self.tester.get_local_port(port_1))
> + # prepare to catch replay packet in out port
> + recPkt = os.path.join('/tmp', "sniff_%s.pcap" % intf)
> + if os.path.exists(recPkt):
> + os.remove(recPkt)
> + index = sniff_packets(intf, count=1, timeout=20, pcap=recPkt)
> + pkt = Packet(pkt_type=pkt_type)
> + if pkt_type == 'VLAN_UDP':
> + pkt.config_layer('dot1q', {'vlan': 20})
> + src_mac = self.tester.get_mac(self.tester.get_local_port(port_0))
> + pkt.config_layer('ether', {'src': src_mac})
> + # save send packet in a pcap file
> + refPkt = self.send_pcap % (pkt_type, "rx", number)
> + if os.path.exists(refPkt):
> + os.remove(refPkt)
> + pkt.pktgen.write_pcap(refPkt)
> + # send out test packet
> + tester_port = self.tester.get_local_port(port_0)
> + intf = self.tester.get_interface(tester_port)
> + pkt.send_pkt(tx_port=intf)
> + # load pcap file caught by out port
> + time.sleep(1)
> + load_sniff_packets(index)
> + # compare pcap file received by out port with scapy reference
> + # packet pcap file
> + warning = self.compare_pkts(refPkt, recPkt, pkt_type)
> + msg = "tcpdump rx Receive Packet error: {0}".format(warning)
> + self.verify(not warning, msg)
> + # check send tx packet by port 1
> + # send packet to dut and compare dpdk-pdump dump pcap
> + # with scapy pcap file
> + intf = self.tester.get_interface(self.tester.get_local_port(port_0))
> + # prepare to catch replay packet in out port
> + recPkt = os.path.join('/tmp', "sniff_%s.pcap" % intf)
> + if os.path.exists(recPkt):
> + os.remove(recPkt)
> + index = sniff_packets(intf, count=1, timeout=20, pcap=recPkt)
> + pkt = Packet(pkt_type=pkt_type)
> + if pkt_type == 'VLAN_UDP':
> + pkt.config_layer('dot1q', {'vlan': 20})
> + src_mac = self.tester.get_mac(self.tester.get_local_port(port_1))
> + pkt.config_layer('ether', {'src': src_mac})
> + # save send packet in a pcap file
> + refPkt = self.send_pcap % (pkt_type, "tx", number)
> + if os.path.exists(refPkt):
> + os.remove(refPkt)
> + pkt.pktgen.write_pcap(refPkt)
> + # send out test packet
> + tester_port = self.tester.get_local_port(port_1)
> + intf = self.tester.get_interface(tester_port)
> + pkt.send_pkt(tx_port=intf)
> + # load pcap file caught by out port
> + time.sleep(1)
> + load_sniff_packets(index)
> + # compare pcap file received by out port
> + # with scapy reference packet pcap file
> + warning = self.compare_pkts(refPkt, recPkt, pkt_type)
> + msg = "tcpdump tx Receive Packet error: {0}".format(warning)
> + self.verify(not warning, msg)
> +
> + def check_pdump_pcaps(self, pkt_type, number, **kwargs):
> + rx_dump_pcap = kwargs["rx"][0]
> + if rx_dump_pcap:
> + self.verify(os.path.exists(rx_dump_pcap),
> + "{0} doesn't exist".format(rx_dump_pcap))
> + # save send packet in a pcap file
> + refPkt = self.send_pcap % (pkt_type, "rx", number)
> + self.verify(os.path.exists(refPkt),
> + "{0} doesn't exist".format(refPkt))
> + # compare pcap file dumped by dpdk-pdump
> + # with scapy reference packet pcap file
> + warning = self.compare_pkts(refPkt, rx_dump_pcap, pkt_type,
> + targetPacketNo=self.rx_packet_pos)
> + if kwargs["rx"][1] == 2:
> + self.rx_packet_pos += kwargs["rx"][1]
> + else:
> + self.rx_packet_pos += 1
> + self.verify(warning is None,
> + "dpdk-pdump rx dump packet error: {0}".format(warning))
> + msg = "pdump rx {0} packet content is correct".format(pkt_type)
> + self.logger.info(msg)
> + tx_dump_pcap = kwargs["tx"][0]
> + if tx_dump_pcap:
> + self.verify(os.path.exists(tx_dump_pcap),
> + "{0} doesn't exist".format(tx_dump_pcap))
> + # set send packet
> + refPkt = self.send_pcap % (pkt_type, "tx", number)
> + self.verify(os.path.exists(refPkt),
> + "{0} doesn't exist".format(refPkt))
> + # compare pcap file dumped by dpdk-pdump
> + # with scapy reference packet pcap file
> + if kwargs["tx"][1] == 2 and self.tx_packet_pos == 0:
> + self.tx_packet_pos = 1
> + warning = self.compare_pkts(refPkt, tx_dump_pcap, pkt_type,
> + targetPacketNo=self.tx_packet_pos)
> + if kwargs["tx"][1] == 2:
> + self.tx_packet_pos += kwargs["tx"][1]
> + else:
> + self.tx_packet_pos += 1
> + msg = "dpdk-pdump tx dump packet error: {0}".format(warning)
> + self.verify(warning is None, msg)
> + msg = "pdump tx {0} packet content is correct".format(pkt_type)
> + self.logger.info(msg)
> +
> + def packet_capture_test_packets(self, option):
> + self.clear_ASLR()
> + self.start_testpmd()
> + self.start_dpdk_pdump(option)
> + if self.dev_iface_flag:
> + self.start_tcpdump_iface(option[1])
> + self.check_pdump_ready(option[1])
> + for number, packet_type in enumerate(self.get_packet_types()):
> + self.packet_capture_dump_packets(packet_type, number,
> **option[1])
> + time.sleep(2)
> + if self.dev_iface_flag:
> + self.stop_tcpdump_iface()
> + if self.dut.get_ip_address() != self.tester.get_ip_address():
> + # copy rx pdump data from dut
> + if self.is_existed_on_dut(self.rx_pcap):
> + if os.path.exists(self.rx_pcap):
> + os.remove(self.rx_pcap)
> + self.dut.session.copy_file_from(self.rx_pcap, self.rx_pcap)
> + # copy tx pdump data from dut
> + if self.is_existed_on_dut(self.tx_pcap):
> + if os.path.exists(self.tx_pcap):
> + os.remove(self.tx_pcap)
> + self.dut.session.copy_file_from(self.tx_pcap, self.tx_pcap)
> + self.stop_dpdk_pdump()
> + self.stop_testpmd()
> + self.reset_ASLR()
> +
> + self.rx_packet_pos = 0
> + self.tx_packet_pos = 0
> + for number, packet_type in enumerate(self.get_packet_types()):
> + self.check_pdump_pcaps(packet_type, number, **option[1])
> + time.sleep(2)
> +
> + def packet_capture_test_options(self, pdump_options):
> + try:
> + for option in pdump_options:
> + self.packet_capture_test_packets(option)
> + self.exit_flag = True
> + except Exception as e:
> + self.logger.error(traceback.format_exc())
> + raise VerifyFailure(e)
> +
> + def test_pdump_port(self):
> + '''
> + test different port options::
> + *. port=<dut port id>
> + *. device_id=<dut pci address>
> + '''
> + port_id = self.test_port0_id
> + port_name = self.dut.ports_info[port_id]['intf']
> + port_pci = self.dut.ports_info[port_id]['pci']
> + test_types = ["port"]
> + options = self.generate_options(
> + port_id, port_pci, port_name, test_types)
> + self.packet_capture_test_options(options)
> +
> + def test_pdump_dev_pcap(self):
> + '''
> + test different dump options with pcap files as output::
> + *. tx-dev=/xxx/pdump-tx.pcap,rx-dev=/xxx/pdump-rx.pcap
> + *. rx-dev=/xxx/pdump-rx.pcap
> + *. tx-dev=/xxx/pdump-tx.pcap
> + '''
> + port_id = self.test_port0_id
> + port_name = self.dut.ports_info[port_id]['intf']
> + port_pci = self.dut.ports_info[port_id]['pci']
> + test_types = ["dev-pcap"]
> + options = self.generate_options(
> + port_id, port_pci, port_name, test_types)
> + self.packet_capture_test_options(options)
> +
> + def test_pdump_dev_iface(self):
> + '''
> + test different dump options with interfaces as output::
> + *. tx-dev=<dut tx port name>,rx-dev=<dut rx port name>
> + *. rx-dev=<dut rx port name>
> + *. tx-dev=<dut tx port name>
> + '''
> + self.get_dut_iface_with_kernel_driver()
> + port_id = self.test_port0_id
> + port_name = self.dut.ports_info[port_id]['intf']
> + port_pci = self.dut.ports_info[port_id]['pci']
> + test_types = ["dev-iface"]
> + self.dev_iface_flag = True
> + options = self.generate_options(
> + port_id, port_pci, port_name, test_types)
> + self.packet_capture_test_options(options)
> +
> + def test_pdump_queue(self):
> + '''
> + test different queue options::
> + *. first queue:
> + queue=0
> + *. all:
> + queue=*
> + '''
> + port_id = self.test_port0_id
> + port_name = self.dut.ports_info[port_id]['intf']
> + port_pci = self.dut.ports_info[port_id]['pci']
> + test_types = ["queue"]
> + options = self.generate_options(
> + port_id, port_pci, port_name, test_types)
> + self.packet_capture_test_options(options)
> +
> + def test_pdump_ring_size(self):
> + '''
> + test ring size option, set value within 2^[1~27]
> + '''
> + port_id = self.test_port0_id
> + port_name = self.dut.ports_info[port_id]['intf']
> + port_pci = self.dut.ports_info[port_id]['pci']
> + test_types = ["ring_size"]
> + options = self.generate_options(
> + port_id, port_pci, port_name, test_types)
> + self.packet_capture_test_options(options)
> +
> + def test_pdump_mbuf_size(self):
> + '''
> + test mbuf size option, set value within [252~50000].
> + min value is decided by single packet size,
> + max value is decided by test platform memory size.
> + '''
> + port_id = self.test_port0_id
> + port_name = self.dut.ports_info[port_id]['intf']
> + port_pci = self.dut.ports_info[port_id]['pci']
> + test_types = ["mbuf_size"]
> +
> + options = self.generate_options(
> + port_id, port_pci, port_name, test_types)
> + self.packet_capture_test_options(options)
> +
> + def test_pdump_total_num_mbufs(self):
> + '''
> + test total-num-mbufs option, set value within [1025~65535]
> + '''
> + port_id = self.test_port0_id
> + port_name = self.dut.ports_info[port_id]['intf']
> + port_pci = self.dut.ports_info[port_id]['pci']
> + test_types = ["total_num_mbufs"]
> + options = self.generate_options(
> + port_id, port_pci, port_name, test_types)
> + self.packet_capture_test_options(options)
> +
> + def set_up_all(self):
> + '''
> + Run at the start of each test suite.
> + '''
> + self.verify(self.target == "x86_64-native-linuxapp-gcc",
> + "only support x86_64-native-linuxapp-gcc")
> + self.dut_ports = self.dut.get_ports()
> + self.verify(len(self.dut_ports) == 2, "Insufficient ports for testing")
> + self.verify_supported_nic()
> + self.test_port0_id = 0
> + self.test_port1_id = 1
> + # compile dpdk app with SW CONFIG_RTE_LIBRTE_PMD_PCAP open
> + self.dut_skip_compile = self.dut.skip_setup
> + # used for save log
> + self.pdump_log = os.sep.join(["/root", 'pdumpLog'])
> + if not self.is_existed_on_dut(self.pdump_log):
> + cmd = "mkdir -p {0}".format(self.pdump_log)
> + self.dut.alt_session.send_expect(cmd, "# ")
> + if not (self.dut_skip_compile and self.check_pcap_lib()):
> + self.pcap_SW = "CONFIG_RTE_LIBRTE_PMD_PCAP"
> + cmd = "sed -i -e 's/{0}=n$/{0}=y/' config/common_base".format(
> + self.pcap_SW)
> + self.dut.alt_session.send_expect(cmd, "# ", 30)
> + self.dut.skip_setup = False
> + self.dut.build_install_dpdk(self.target)
> + self.session_ex = self.dut.new_session()
> + # secondary process (dpdk-pdump)
> + self.pdump_dir = self.dut.base_dir + os.sep + \
> + "%s/build/app/pdump/" % self.target
> + cmd = "ls {0} -F | grep '*'".format(self.pdump_dir)
> + exe_files = self.dut.alt_session.send_expect(cmd, "# ").splitlines()
> + if len(exe_files) == 1:
> + self.tool_name = exe_files[0][:-1]
> + else:
> + self.verify(False, "tool name exception !")
> + self.dut_dpdk_pdump_dir = self.dut.base_dir + os.sep + \
> + "%s/app/%s" % (self.target, self.tool_name)
> + self.dpdk_pdump = self.dut_dpdk_pdump_dir + \
> + " -v --file-prefix=test -- --pdump "
> + self.send_pcap = os.sep.join([self.pdump_log,
> "scapy_%s_%s_%d.pcap"])
> + self.rx_pcap = os.sep.join([self.pdump_log, "pdump-rx.pcap"])
> + self.tx_pcap = os.sep.join([self.pdump_log, "pdump-tx.pcap"])
> + self.rxtx_pcap = os.sep.join([self.pdump_log, "pdump-rxtx.pcap"])
> + self.rx_iface = None
> + self.tx_iface = None
> + self.rxtx_iface = None
> + self.rx_packet_pos = 0
> + self.tx_packet_pos = 0
> + self.dev_iface_flag = False
> + # primary process
> + self.testpmd = PmdOutput(self.dut)
> + # False: reduce test items for regression testing,
> + # shut off base test environment checking
> + # True: make a full range testing
> + self.full_test = True
> + # used on daily testing to avoid long time running
> + self.pkt_index = None
> + # get tcpdump options
> + self.get_tcpdump_options()
> +
> + def set_up(self):
> + '''
> + Run before each test case.
> + '''
> + self.exit_flag = False
> +
> + def tear_down(self):
> + '''
> + Run after each test case.
> + '''
> + if not self.exit_flag:
> + self.stop_dpdk_pdump()
> + self.dut.alt_session.send_expect("killall testpmd", "# ")
> + self.dut.alt_session.send_expect("killall tcpdump", "# ")
> + self.reset_ASLR()
> + if self.dev_iface_flag:
> + self.stop_tcpdump_iface()
> + self.dev_iface_flag = False
> +
> + def tear_down_all(self):
> + '''
> + Run after each test suite.
> + '''
> + if self.session_ex:
> + self.reset_ASLR()
> + self.session_ex.close()
> + self.session_ex = None
> + # Restore the config file and recompile the package.
> + if not (self.dut_skip_compile and self.check_pcap_lib()):
> +
> ################################################################
> ###
> + self.dut.alt_session.send_expect(
> + ("sed -i -e 's/{0}=y$/{0}=n/' "
> + "config/common_base").format(self.pcap_SW), "# ", 120)
> + # temporary disable skip_setup
> + skip_setup = self.dut.skip_setup
> + self.dut.skip_setup = True
> + self.dut.build_install_dpdk(self.target)
> + self.dut.skip_setup = skip_setup
> +
> ################################################################
> ###
> + self.dut.kill_all()
> --
> 1.9.3
^ permalink raw reply [flat|nested] 5+ messages in thread