f62ae1288a
Change-Id: I31da3b1857b6399f9899276a2d99cdd19436296c Signed-off-by: Klement Sekera <ksekera@cisco.com> Signed-off-by: Matej Klotton <mklotton@cisco.com> Signed-off-by: Jan Gelety <jgelety@cisco.com> Signed-off-by: Juraj Sloboda <jsloboda@cisco.com>
100 lines
2.9 KiB
Python
100 lines
2.9 KiB
Python
import os
|
|
from logging import error
|
|
from scapy.utils import wrpcap, rdpcap
|
|
from vpp_interface import VppInterface
|
|
|
|
|
|
class VppPGInterface(VppInterface):
|
|
"""
|
|
VPP packet-generator interface
|
|
"""
|
|
|
|
@property
|
|
def pg_index(self):
|
|
"""packet-generator interface index assigned by VPP"""
|
|
return self._pg_index
|
|
|
|
@property
|
|
def out_path(self):
|
|
"""pcap file path - captured packets"""
|
|
return self._out_path
|
|
|
|
@property
|
|
def in_path(self):
|
|
""" pcap file path - injected packets"""
|
|
return self._in_path
|
|
|
|
@property
|
|
def capture_cli(self):
|
|
"""CLI string to start capture on this interface"""
|
|
return self._capture_cli
|
|
|
|
@property
|
|
def cap_name(self):
|
|
"""capture name for this interface"""
|
|
return self._cap_name
|
|
|
|
@property
|
|
def input_cli(self):
|
|
"""CLI string to load the injected packets"""
|
|
return self._input_cli
|
|
|
|
def post_init_setup(self):
|
|
""" Perform post-init setup for super class and add our own setup """
|
|
super(VppPGInterface, self).post_init_setup()
|
|
self._out_path = self.test.tempdir + "/pg%u_out.pcap" % self.sw_if_index
|
|
self._in_path = self.test.tempdir + "/pg%u_in.pcap" % self.sw_if_index
|
|
self._capture_cli = "packet-generator capture pg%u pcap %s" % (
|
|
self.pg_index, self.out_path)
|
|
self._cap_name = "pcap%u" % self.sw_if_index
|
|
self._input_cli = "packet-generator new pcap %s source pg%u name %s" % (
|
|
self.in_path, self.pg_index, self.cap_name)
|
|
|
|
def __init__(self, test, pg_index):
|
|
""" Create VPP packet-generator interface """
|
|
self._pg_index = pg_index
|
|
self._test = test
|
|
r = self.test.vapi.pg_create_interface(self.pg_index)
|
|
self._sw_if_index = r.sw_if_index
|
|
self.post_init_setup()
|
|
|
|
def enable_capture(self):
|
|
""" Enable capture on this packet-generator interface"""
|
|
try:
|
|
os.unlink(self.out_path)
|
|
except:
|
|
pass
|
|
# FIXME this should be an API, but no such exists atm
|
|
self.test.vapi.cli(self.capture_cli)
|
|
|
|
def add_stream(self, pkts):
|
|
"""
|
|
Add a stream of packets to this packet-generator
|
|
|
|
:param pkts: iterable packets
|
|
|
|
"""
|
|
try:
|
|
os.remove(self.in_path)
|
|
except:
|
|
pass
|
|
wrpcap(self.in_path, pkts)
|
|
# FIXME this should be an API, but no such exists atm
|
|
self.test.vapi.cli(self.input_cli)
|
|
self.test.pg_streams.append(self.cap_name)
|
|
self.test.vapi.cli("trace add pg-input %d" % len(pkts))
|
|
|
|
def get_capture(self):
|
|
"""
|
|
Get captured packets
|
|
|
|
:returns: iterable packets
|
|
"""
|
|
try:
|
|
output = rdpcap(self.out_path)
|
|
except IOError: # TODO
|
|
error("File %s does not exist, probably because no"
|
|
" packets arrived" % self.out_path)
|
|
return []
|
|
return output
|