test/test_l2xc.py

   1 #!/usr/bin/env python
   2
   3 import unittest
   4 import random
   5
   6 from scapy.packet import Raw
   7 from scapy.layers.l2 import Ether
   8 from scapy.layers.inet import IP, UDP
   9
  10 from framework import VppTestCase, VppTestRunner
  11 from util import Host
  12
  13
  14 class TestL2xc(VppTestCase):
  15     """ L2XC Test Case """
  16
  17     @classmethod
  18     def setUpClass(cls):
  19         """
  20         Perform standard class setup (defined by class method setUpClass in
  21         class VppTestCase) before running the test case, set test case related
  22         variables and configure VPP.
  23
  24         :var int hosts_nr: Number of hosts to be created.
  25         :var int dl_pkts_per_burst: Number of packets in burst for dual-loop
  26             test.
  27         :var int sl_pkts_per_burst: Number of packets in burst for single-loop
  28             test.
  29         """
  30         super(TestL2xc, cls).setUpClass()
  31
  32         # Test variables
  33         cls.hosts_nr = 10
  34         cls.dl_pkts_per_burst = 257
  35         cls.sl_pkts_per_burst = 2
  36
  37         try:
  38             # create 4 pg interfaces
  39             cls.create_pg_interfaces(range(4))
  40
  41             # packet flows mapping pg0 -> pg1, pg2 -> pg3, etc.
  42             cls.flows = dict()
  43             cls.flows[cls.pg0] = [cls.pg1]
  44             cls.flows[cls.pg1] = [cls.pg0]
  45             cls.flows[cls.pg2] = [cls.pg3]
  46             cls.flows[cls.pg3] = [cls.pg2]
  47
  48             # packet sizes
  49             cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
  50
  51             cls.interfaces = list(cls.pg_interfaces)
  52
  53             # Create bi-directional cross-connects between pg0 and pg1
  54             cls.vapi.sw_interface_set_l2_xconnect(
  55                 cls.pg0.sw_if_index, cls.pg1.sw_if_index, enable=1)
  56             cls.vapi.sw_interface_set_l2_xconnect(
  57                 cls.pg1.sw_if_index, cls.pg0.sw_if_index, enable=1)
  58
  59             # Create bi-directional cross-connects between pg2 and pg3
  60             cls.vapi.sw_interface_set_l2_xconnect(
  61                 cls.pg2.sw_if_index, cls.pg3.sw_if_index, enable=1)
  62             cls.vapi.sw_interface_set_l2_xconnect(
  63                 cls.pg3.sw_if_index, cls.pg2.sw_if_index, enable=1)
  64
  65             # mapping between packet-generator index and lists of test hosts
  66             cls.hosts_by_pg_idx = dict()
  67
  68             # Create host MAC and IPv4 lists
  69             cls.create_host_lists(cls.hosts_nr)
  70
  71             # setup all interfaces
  72             for i in cls.interfaces:
  73                 i.admin_up()
  74
  75         except Exception:
  76             super(TestL2xc, cls).tearDownClass()
  77             raise
  78
  79     def setUp(self):
  80         """
  81         Clear trace and packet infos before running each test.
  82         """
  83         super(TestL2xc, self).setUp()
  84         self.packet_infos = {}
  85
  86     def tearDown(self):
  87         """
  88         Show various debug prints after each test.
  89         """
  90         super(TestL2xc, self).tearDown()
  91         if not self.vpp_dead:
  92             self.logger.info(self.vapi.ppcli("show l2patch"))
  93
  94     @classmethod
  95     def create_host_lists(cls, count):
  96         """
  97         Method to create required number of MAC and IPv4 addresses.
  98         Create required number of host MAC addresses and distribute them among
  99         interfaces. Create host IPv4 address for every host MAC address too.
 100
 101         :param count: Number of hosts to create MAC and IPv4 addresses for.
 102         """
 103         for pg_if in cls.pg_interfaces:
 104             cls.hosts_by_pg_idx[pg_if.sw_if_index] = []
 105             hosts = cls.hosts_by_pg_idx[pg_if.sw_if_index]
 106             for j in range(0, count):
 107                 host = Host(
 108                     "00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
 109                     "172.17.1%02x.%u" % (pg_if.sw_if_index, j))
 110                 hosts.append(host)
 111
 112     def create_stream(self, src_if, packet_sizes, packets_per_burst):
 113         """
 114         Create input packet stream for defined interface.
 115
 116         :param object src_if: Interface to create packet stream for.
 117         :param list packet_sizes: List of required packet sizes.
 118         :param int packets_per_burst: Number of packets in burst.
 119         :return: Stream of packets.
 120         """
 121         pkts = []
 122         for i in range(0, packets_per_burst):
 123             dst_if = self.flows[src_if][0]
 124             dst_host = random.choice(self.hosts_by_pg_idx[dst_if.sw_if_index])
 125             src_host = random.choice(self.hosts_by_pg_idx[src_if.sw_if_index])
 126             pkt_info = self.create_packet_info(
 127                 src_if.sw_if_index, dst_if.sw_if_index)
 128             payload = self.info_to_payload(pkt_info)
 129             p = (Ether(dst=dst_host.mac, src=src_host.mac) /
 130                  IP(src=src_host.ip4, dst=dst_host.ip4) /
 131                  UDP(sport=1234, dport=1234) /
 132                  Raw(payload))
 133             pkt_info.data = p.copy()
 134             size = random.choice(packet_sizes)
 135             self.extend_packet(p, size)
 136             pkts.append(p)
 137         return pkts
 138
 139     def verify_capture(self, pg_if, capture):
 140         """
 141         Verify captured input packet stream for defined interface.
 142
 143         :param object pg_if: Interface to verify captured packet stream for.
 144         :param list capture: Captured packet stream.
 145         """
 146         last_info = dict()
 147         for i in self.interfaces:
 148             last_info[i.sw_if_index] = None
 149         dst_sw_if_index = pg_if.sw_if_index
 150         for packet in capture:
 151             try:
 152                 ip = packet[IP]
 153                 udp = packet[UDP]
 154                 payload_info = self.payload_to_info(str(packet[Raw]))
 155                 packet_index = payload_info.index
 156                 self.assertEqual(payload_info.dst, dst_sw_if_index)
 157                 self.logger.debug("Got packet on port %s: src=%u (id=%u)" %
 158                                   (pg_if.name, payload_info.src, packet_index))
 159                 next_info = self.get_next_packet_info_for_interface2(
 160                     payload_info.src, dst_sw_if_index,
 161                     last_info[payload_info.src])
 162                 last_info[payload_info.src] = next_info
 163                 self.assertTrue(next_info is not None)
 164                 self.assertEqual(packet_index, next_info.index)
 165                 saved_packet = next_info.data
 166                 # Check standard fields
 167                 self.assertEqual(ip.src, saved_packet[IP].src)
 168                 self.assertEqual(ip.dst, saved_packet[IP].dst)
 169                 self.assertEqual(udp.sport, saved_packet[UDP].sport)
 170                 self.assertEqual(udp.dport, saved_packet[UDP].dport)
 171             except:
 172                 self.logger.error("Unexpected or invalid packet:")
 173                 self.logger.error(packet.show())
 174                 raise
 175         for i in self.interfaces:
 176             remaining_packet = self.get_next_packet_info_for_interface2(
 177                 i, dst_sw_if_index, last_info[i.sw_if_index])
 178             self.assertTrue(remaining_packet is None,
 179                             "Port %u: Packet expected from source %u didn't"
 180                             " arrive" % (dst_sw_if_index, i.sw_if_index))
 181
 182     def run_l2xc_test(self, pkts_per_burst):
 183         """ L2XC test """
 184
 185         # Create incoming packet streams for packet-generator interfaces
 186         for i in self.interfaces:
 187             pkts = self.create_stream(i, self.pg_if_packet_sizes,
 188                                       pkts_per_burst)
 189             i.add_stream(pkts)
 190
 191         # Enable packet capturing and start packet sending
 192         self.pg_enable_capture(self.pg_interfaces)
 193         self.pg_start()
 194
 195         # Verify outgoing packet streams per packet-generator interface
 196         for i in self.pg_interfaces:
 197             capture = i.get_capture()
 198             self.logger.info("Verifying capture on interface %s" % i.name)
 199             self.verify_capture(i, capture)
 200
 201     def test_l2xc_sl(self):
 202         """ L2XC single-loop test
 203
 204         Test scenario:
 205             1. config
 206                 2 pairs of 2 interfaces, l2xconnected
 207
 208             2. sending l2 eth packets between 4 interfaces
 209                 64B, 512B, 1518B, 9018B (ether_size)
 210                 burst of 2 packets per interface
 211         """
 212
 213         self.run_l2xc_test(self.sl_pkts_per_burst)
 214
 215     def test_l2xc_dl(self):
 216         """ L2XC dual-loop test
 217
 218         Test scenario:
 219             1. config
 220                 2 pairs of 2 interfaces, l2xconnected
 221
 222             2. sending l2 eth packets between 4 interfaces
 223                 64B, 512B, 1518B, 9018B (ether_size)
 224                 burst of 257 packets per interface
 225         """
 226
 227         self.run_l2xc_test(self.dl_pkts_per_burst)
 228
 229
 230 if __name__ == '__main__':
 231     unittest.main(testRunner=VppTestRunner)