test/test_vtr.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, Dot1Q
   8 from scapy.layers.inet import IP, UDP
   9
  10 from util import Host
  11 from framework import VppTestCase, VppTestRunner
  12 from vpp_sub_interface import L2_VTR_OP, VppDot1QSubint, VppDot1ADSubint
  13 from collections import namedtuple
  14
  15 Tag = namedtuple('Tag', ['dot1', 'vlan'])
  16 DOT1AD = 0x88A8
  17 DOT1Q = 0x8100
  18
  19
  20 class TestVtr(VppTestCase):
  21     """ VTR Test Case """
  22
  23     @classmethod
  24     def setUpClass(cls):
  25         super(TestVtr, cls).setUpClass()
  26
  27         # Test variables
  28         cls.bd_id = 1
  29         cls.mac_entries_count = 5
  30         cls.Atag = 100
  31         cls.Btag = 200
  32         cls.dot1ad_sub_id = 20
  33
  34         try:
  35             ifs = range(3)
  36             cls.create_pg_interfaces(ifs)
  37
  38             cls.sub_interfaces = [
  39                 VppDot1ADSubint(cls, cls.pg1, cls.dot1ad_sub_id,
  40                                 cls.Btag, cls.Atag),
  41                 VppDot1QSubint(cls, cls.pg2, cls.Btag)]
  42
  43             interfaces = list(cls.pg_interfaces)
  44             interfaces.extend(cls.sub_interfaces)
  45
  46             # Create BD with MAC learning enabled and put interfaces and
  47             #  sub-interfaces to this BD
  48             for pg_if in cls.pg_interfaces:
  49                 sw_if_index = pg_if.sub_if.sw_if_index \
  50                     if hasattr(pg_if, 'sub_if') else pg_if.sw_if_index
  51                 cls.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=sw_if_index,
  52                                                     bd_id=cls.bd_id)
  53
  54             # setup all interfaces
  55             for i in interfaces:
  56                 i.admin_up()
  57
  58             # mapping between packet-generator index and lists of test hosts
  59             cls.hosts_by_pg_idx = dict()
  60
  61             # create test host entries and inject packets to learn MAC entries
  62             # in the bridge-domain
  63             cls.create_hosts_and_learn(cls.mac_entries_count)
  64             cls.logger.info(cls.vapi.ppcli("show l2fib"))
  65
  66         except Exception:
  67             super(TestVtr, cls).tearDownClass()
  68             raise
  69
  70     @classmethod
  71     def tearDownClass(cls):
  72         super(TestVtr, cls).tearDownClass()
  73
  74     def setUp(self):
  75         """
  76         Clear trace and packet infos before running each test.
  77         """
  78         super(TestVtr, self).setUp()
  79         self.reset_packet_infos()
  80
  81     def tearDown(self):
  82         """
  83         Show various debug prints after each test.
  84         """
  85         super(TestVtr, self).tearDown()
  86         if not self.vpp_dead:
  87             self.logger.info(self.vapi.ppcli("show l2fib verbose"))
  88             self.logger.info(self.vapi.ppcli("show bridge-domain %s detail" %
  89                                              self.bd_id))
  90
  91     @classmethod
  92     def create_hosts_and_learn(cls, count):
  93         for pg_if in cls.pg_interfaces:
  94             cls.hosts_by_pg_idx[pg_if.sw_if_index] = []
  95             hosts = cls.hosts_by_pg_idx[pg_if.sw_if_index]
  96             packets = []
  97             for j in range(1, count + 1):
  98                 host = Host(
  99                     "00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
 100                     "172.17.1%02x.%u" % (pg_if.sw_if_index, j))
 101                 packet = (Ether(dst="ff:ff:ff:ff:ff:ff", src=host.mac))
 102                 hosts.append(host)
 103                 if hasattr(pg_if, 'sub_if'):
 104                     packet = pg_if.sub_if.add_dot1_layer(packet)
 105                 packets.append(packet)
 106             pg_if.add_stream(packets)
 107         cls.logger.info("Sending broadcast eth frames for MAC learning")
 108         cls.pg_enable_capture(cls.pg_interfaces)
 109         cls.pg_start()
 110
 111     def create_packet(self, src_if, dst_if, do_dot1=True):
 112         packet_sizes = [64, 512, 1518, 9018]
 113         dst_host = random.choice(self.hosts_by_pg_idx[dst_if.sw_if_index])
 114         src_host = random.choice(self.hosts_by_pg_idx[src_if.sw_if_index])
 115         pkt_info = self.create_packet_info(src_if, dst_if)
 116         payload = self.info_to_payload(pkt_info)
 117         p = (Ether(dst=dst_host.mac, src=src_host.mac) /
 118              IP(src=src_host.ip4, dst=dst_host.ip4) /
 119              UDP(sport=1234, dport=1234) /
 120              Raw(payload))
 121         pkt_info.data = p.copy()
 122         if do_dot1 and hasattr(src_if, 'sub_if'):
 123             p = src_if.sub_if.add_dot1_layer(p)
 124         size = random.choice(packet_sizes)
 125         self.extend_packet(p, size)
 126         return p
 127
 128     def _add_tag(self, packet, vlan, tag_type):
 129         payload = packet.payload
 130         inner_type = packet.type
 131         packet.remove_payload()
 132         packet.add_payload(Dot1Q(vlan=vlan) / payload)
 133         packet.payload.type = inner_type
 134         packet.payload.vlan = vlan
 135         packet.type = tag_type
 136         return packet
 137
 138     def _remove_tag(self, packet, vlan=None, tag_type=None):
 139         if tag_type:
 140             self.assertEqual(packet.type, tag_type)
 141
 142         payload = packet.payload
 143         if vlan:
 144             self.assertEqual(payload.vlan, vlan)
 145         inner_type = payload.type
 146         payload = payload.payload
 147         packet.remove_payload()
 148         packet.add_payload(payload)
 149         packet.type = inner_type
 150
 151     def add_tags(self, packet, tags):
 152         for t in reversed(tags):
 153             self._add_tag(packet, t.vlan, t.dot1)
 154
 155     def remove_tags(self, packet, tags):
 156         for t in tags:
 157             self._remove_tag(packet, t.vlan, t.dot1)
 158
 159     def vtr_test(self, swif, tags):
 160         p = self.create_packet(swif, self.pg0)
 161         swif.add_stream(p)
 162         self.pg_enable_capture(self.pg_interfaces)
 163         self.pg_start()
 164         rx = self.pg0.get_capture(1)
 165
 166         if tags:
 167             self.remove_tags(rx[0], tags)
 168         self.assertTrue(Dot1Q not in rx[0])
 169
 170         if not tags:
 171             return
 172
 173         i = VppDot1QSubint(self, self.pg0, tags[0].vlan)
 174         self.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=i.sw_if_index,
 175                                              bd_id=self.bd_id, enable=1)
 176         i.admin_up()
 177
 178         p = self.create_packet(self.pg0, swif, do_dot1=False)
 179         self.add_tags(p, tags)
 180         self.pg0.add_stream(p)
 181         self.pg_enable_capture(self.pg_interfaces)
 182         self.pg_start()
 183         rx = swif.get_capture(1)
 184         swif.sub_if.remove_dot1_layer(rx[0])
 185         self.assertTrue(Dot1Q not in rx[0])
 186
 187         self.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=i.sw_if_index,
 188                                              bd_id=self.bd_id, enable=0)
 189         i.remove_vpp_config()
 190
 191     def test_1ad_vtr_pop_1(self):
 192         """ 1AD VTR pop 1 test
 193         """
 194         self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_POP_1)
 195         self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=100)])
 196
 197     def test_1ad_vtr_pop_2(self):
 198         """ 1AD VTR pop 2 test
 199         """
 200         self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_POP_2)
 201         self.vtr_test(self.pg1, [])
 202
 203     def test_1ad_vtr_push_1ad(self):
 204         """ 1AD VTR push 1 1AD test
 205         """
 206         self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_1, tag=300)
 207         self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=300),
 208                                  Tag(dot1=DOT1AD, vlan=200),
 209                                  Tag(dot1=DOT1Q, vlan=100)])
 210
 211     def test_1ad_vtr_push_2ad(self):
 212         """ 1AD VTR push 2 1AD test
 213         """
 214         self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_2, outer=400, inner=300)
 215         self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=400),
 216                                  Tag(dot1=DOT1Q, vlan=300),
 217                                  Tag(dot1=DOT1AD, vlan=200),
 218                                  Tag(dot1=DOT1Q, vlan=100)])
 219
 220     def test_1ad_vtr_push_1q(self):
 221         """ 1AD VTR push 1 1Q test
 222         """
 223         self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_1, tag=300, push1q=1)
 224         self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=300),
 225                                  Tag(dot1=DOT1AD, vlan=200),
 226                                  Tag(dot1=DOT1Q, vlan=100)])
 227
 228     def test_1ad_vtr_push_2q(self):
 229         """ 1AD VTR push 2 1Q test
 230         """
 231         self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_2,
 232                                 outer=400, inner=300, push1q=1)
 233         self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=400),
 234                                  Tag(dot1=DOT1Q, vlan=300),
 235                                  Tag(dot1=DOT1AD, vlan=200),
 236                                  Tag(dot1=DOT1Q, vlan=100)])
 237
 238     def test_1ad_vtr_translate_1_1ad(self):
 239         """ 1AD VTR translate 1 -> 1 1AD test
 240         """
 241         self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_1_1, tag=300)
 242         self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=300),
 243                                  Tag(dot1=DOT1Q, vlan=100)])
 244
 245     def test_1ad_vtr_translate_1_2ad(self):
 246         """ 1AD VTR translate 1 -> 2 1AD test
 247         """
 248         self.pg1.sub_if.set_vtr(
 249             L2_VTR_OP.L2_TRANSLATE_1_2, inner=300, outer=400)
 250         self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=400),
 251                                  Tag(dot1=DOT1Q, vlan=300),
 252                                  Tag(dot1=DOT1Q, vlan=100)])
 253
 254     def test_1ad_vtr_translate_2_1ad(self):
 255         """ 1AD VTR translate 2 -> 1 1AD test
 256         """
 257         self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_2_1, tag=300)
 258         self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=300)])
 259
 260     def test_1ad_vtr_translate_2_2ad(self):
 261         """ 1AD VTR translate 2 -> 2 1AD test
 262         """
 263         self.pg1.sub_if.set_vtr(
 264             L2_VTR_OP.L2_TRANSLATE_2_2, inner=300, outer=400)
 265         self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=400),
 266                                  Tag(dot1=DOT1Q, vlan=300)])
 267
 268     def test_1ad_vtr_translate_1_1q(self):
 269         """ 1AD VTR translate 1 -> 1 1Q test
 270         """
 271         self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_1_1, tag=300, push1q=1)
 272         self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=300),
 273                                  Tag(dot1=DOT1Q, vlan=100)])
 274
 275     def test_1ad_vtr_translate_1_2q(self):
 276         """ 1AD VTR translate 1 -> 2 1Q test
 277         """
 278         self.pg1.sub_if.set_vtr(
 279             L2_VTR_OP.L2_TRANSLATE_1_2, inner=300, outer=400, push1q=1)
 280         self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=400),
 281                                  Tag(dot1=DOT1Q, vlan=300),
 282                                  Tag(dot1=DOT1Q, vlan=100)])
 283
 284     def test_1ad_vtr_translate_2_1q(self):
 285         """ 1AD VTR translate 2 -> 1 1Q test
 286         """
 287         self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_2_1, tag=300, push1q=1)
 288         self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=300)])
 289
 290     def test_1ad_vtr_translate_2_2q(self):
 291         """ 1AD VTR translate 2 -> 2 1Q test
 292         """
 293         self.pg1.sub_if.set_vtr(
 294             L2_VTR_OP.L2_TRANSLATE_2_2, inner=300, outer=400, push1q=1)
 295         self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=400),
 296                                  Tag(dot1=DOT1Q, vlan=300)])
 297
 298     def test_1q_vtr_pop_1(self):
 299         """ 1Q VTR pop 1 test
 300         """
 301         self.pg2.sub_if.set_vtr(L2_VTR_OP.L2_POP_1)
 302         self.vtr_test(self.pg2, [])
 303
 304     def test_1q_vtr_push_1(self):
 305         """ 1Q VTR push 1 test
 306         """
 307         self.pg2.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_1, tag=300)
 308         self.vtr_test(self.pg2, [Tag(dot1=DOT1AD, vlan=300),
 309                                  Tag(dot1=DOT1Q, vlan=200)])
 310
 311     def test_1q_vtr_push_2(self):
 312         """ 1Q VTR push 2 test
 313         """
 314         self.pg2.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_2, outer=400, inner=300)
 315         self.vtr_test(self.pg2, [Tag(dot1=DOT1AD, vlan=400),
 316                                  Tag(dot1=DOT1Q, vlan=300),
 317                                  Tag(dot1=DOT1Q, vlan=200)])
 318
 319     def test_1q_vtr_translate_1_1(self):
 320         """ 1Q VTR translate 1 -> 1 test
 321         """
 322         self.pg2.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_1_1, tag=300)
 323         self.vtr_test(self.pg2, [Tag(dot1=DOT1AD, vlan=300)])
 324
 325     def test_1q_vtr_translate_1_2(self):
 326         """ 1Q VTR translate 1 -> 2 test
 327         """
 328         self.pg2.sub_if.set_vtr(
 329             L2_VTR_OP.L2_TRANSLATE_1_2, inner=300, outer=400)
 330         self.vtr_test(self.pg2, [Tag(dot1=DOT1AD, vlan=400),
 331                                  Tag(dot1=DOT1Q, vlan=300)])
 332
 333
 334 if __name__ == '__main__':
 335     unittest.main(testRunner=VppTestRunner)