test/vpp_pg_interface.py

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