test/vpp_pg_interface.py

   1 import os
   2 from logging import error
   3 from scapy.utils import wrpcap, rdpcap
   4 from vpp_interface import VppInterface
   5
   6
   7 class VppPGInterface(VppInterface):
   8     """
   9     VPP packet-generator interface
  10     """
  11
  12     @property
  13     def pg_index(self):
  14         """packet-generator interface index assigned by VPP"""
  15         return self._pg_index
  16
  17     @property
  18     def out_path(self):
  19         """pcap file path - captured packets"""
  20         return self._out_path
  21
  22     @property
  23     def in_path(self):
  24         """ pcap file path - injected packets"""
  25         return self._in_path
  26
  27     @property
  28     def capture_cli(self):
  29         """CLI string to start capture on this interface"""
  30         return self._capture_cli
  31
  32     @property
  33     def cap_name(self):
  34         """capture name for this interface"""
  35         return self._cap_name
  36
  37     @property
  38     def input_cli(self):
  39         """CLI string to load the injected packets"""
  40         return self._input_cli
  41
  42     def post_init_setup(self):
  43         """ Perform post-init setup for super class and add our own setup """
  44         super(VppPGInterface, self).post_init_setup()
  45         self._out_path = self.test.tempdir + "/pg%u_out.pcap" % self.sw_if_index
  46         self._in_path = self.test.tempdir + "/pg%u_in.pcap" % self.sw_if_index
  47         self._capture_cli = "packet-generator capture pg%u pcap %s" % (
  48             self.pg_index, self.out_path)
  49         self._cap_name = "pcap%u" % self.sw_if_index
  50         self._input_cli = "packet-generator new pcap %s source pg%u name %s" % (
  51             self.in_path, self.pg_index, self.cap_name)
  52
  53     def __init__(self, test, pg_index):
  54         """ Create VPP packet-generator interface """
  55         self._pg_index = pg_index
  56         self._test = test
  57         r = self.test.vapi.pg_create_interface(self.pg_index)
  58         self._sw_if_index = r.sw_if_index
  59         self.post_init_setup()
  60
  61     def enable_capture(self):
  62         """ Enable capture on this packet-generator interface"""
  63         try:
  64             os.unlink(self.out_path)
  65         except:
  66             pass
  67         # FIXME this should be an API, but no such exists atm
  68         self.test.vapi.cli(self.capture_cli)
  69
  70     def add_stream(self, pkts):
  71         """
  72         Add a stream of packets to this packet-generator
  73
  74         :param pkts: iterable packets
  75
  76         """
  77         try:
  78             os.remove(self.in_path)
  79         except:
  80             pass
  81         wrpcap(self.in_path, pkts)
  82         # FIXME this should be an API, but no such exists atm
  83         self.test.vapi.cli(self.input_cli)
  84         self.test.pg_streams.append(self.cap_name)
  85         self.test.vapi.cli("trace add pg-input %d" % len(pkts))
  86
  87     def get_capture(self):
  88         """
  89         Get captured packets
  90
  91         :returns: iterable packets
  92         """
  93         try:
  94             output = rdpcap(self.out_path)
  95         except IOError:  # TODO
  96             error("File %s does not exist, probably because no"
  97                   " packets arrived" % self.out_path)
  98             return []
  99         return output