test/test_vtr.py

   1 #!/usr/bin/env python3
   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
  87     def show_commands_at_teardown(self):
  88         self.logger.info(self.vapi.ppcli("show l2fib verbose"))
  89         self.logger.info(self.vapi.ppcli("show bridge-domain %s detail" %
  90                                          self.bd_id))
  91
  92     @classmethod
  93     def create_hosts_and_learn(cls, count):
  94         for pg_if in cls.pg_interfaces:
  95             cls.hosts_by_pg_idx[pg_if.sw_if_index] = []
  96             hosts = cls.hosts_by_pg_idx[pg_if.sw_if_index]
  97             packets = []
  98             for j in range(1, count + 1):
  99                 host = Host(
 100                     "00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
 101                     "172.17.1%02x.%u" % (pg_if.sw_if_index, j))
 102                 packet = (Ether(dst="ff:ff:ff:ff:ff:ff", src=host.mac))
 103                 hosts.append(host)
 104                 if hasattr(pg_if, 'sub_if'):
 105                     packet = pg_if.sub_if.add_dot1_layer(packet)
 106                 packets.append(packet)
 107             pg_if.add_stream(packets)
 108         cls.logger.info("Sending broadcast eth frames for MAC learning")
 109         cls.pg_enable_capture(cls.pg_interfaces)
 110         cls.pg_start()
 111
 112     def create_packet(self, src_if, dst_if, do_dot1=True):
 113         packet_sizes = [64, 512, 1518, 9018]
 114         dst_host = random.choice(self.hosts_by_pg_idx[dst_if.sw_if_index])
 115         src_host = random.choice(self.hosts_by_pg_idx[src_if.sw_if_index])
 116         pkt_info = self.create_packet_info(src_if, dst_if)
 117         payload = self.info_to_payload(pkt_info)
 118         p = (Ether(dst=dst_host.mac, src=src_host.mac) /
 119              IP(src=src_host.ip4, dst=dst_host.ip4) /
 120              UDP(sport=1234, dport=1234) /
 121              Raw(payload))
 122         pkt_info.data = p.copy()
 123         if do_dot1 and hasattr(src_if, 'sub_if'):
 124             p = src_if.sub_if.add_dot1_layer(p)
 125         size = random.choice(packet_sizes)
 126         self.extend_packet(p, size)
 127         return p
 128
 129     def _add_tag(self, packet, vlan, tag_type):
 130         payload = packet.payload
 131         inner_type = packet.type
 132         packet.remove_payload()
 133         packet.add_payload(Dot1Q(vlan=vlan) / payload)
 134         packet.payload.type = inner_type
 135         packet.payload.vlan = vlan
 136         packet.type = tag_type
 137         return packet
 138
 139     def _remove_tag(self, packet, vlan=None, tag_type=None):
 140         if tag_type:
 141             self.assertEqual(packet.type, tag_type)
 142
 143         payload = packet.payload
 144         if vlan:
 145             self.assertEqual(payload.vlan, vlan)
 146         inner_type = payload.type
 147         payload = payload.payload
 148         packet.remove_payload()
 149         packet.add_payload(payload)
 150         packet.type = inner_type
 151
 152     def add_tags(self, packet, tags):
 153         for t in reversed(tags):
 154             self._add_tag(packet, t.vlan, t.dot1)
 155
 156     def remove_tags(self, packet, tags):
 157         for t in tags:
 158             self._remove_tag(packet, t.vlan, t.dot1)
 159
 160     def vtr_test(self, swif, tags):
 161         p = self.create_packet(swif, self.pg0)
 162         swif.add_stream(p)
 163         self.pg_enable_capture(self.pg_interfaces)
 164         self.pg_start()
 165         rx = self.pg0.get_capture(1)
 166
 167         if tags:
 168             self.remove_tags(rx[0], tags)
 169         self.assertTrue(Dot1Q not in rx[0])
 170
 171         if not tags:
 172             return
 173
 174         i = VppDot1QSubint(self, self.pg0, tags[0].vlan)
 175         self.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=i.sw_if_index,
 176                                              bd_id=self.bd_id, enable=1)
 177         i.admin_up()
 178
 179         p = self.create_packet(self.pg0, swif, do_dot1=False)
 180         self.add_tags(p, tags)
 181         self.pg0.add_stream(p)
 182         self.pg_enable_capture(self.pg_interfaces)
 183         self.pg_start()
 184         rx = swif.get_capture(1)
 185         swif.sub_if.remove_dot1_layer(rx[0])
 186         self.assertTrue(Dot1Q not in rx[0])
 187
 188         self.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=i.sw_if_index,
 189                                              bd_id=self.bd_id, enable=0)
 190         i.remove_vpp_config()
 191
 192     def test_1ad_vtr_pop_1(self):
 193         """ 1AD VTR pop 1 test
 194         """
 195         self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_POP_1)
 196         self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=100)])
 197
 198     def test_1ad_vtr_pop_2(self):
 199         """ 1AD VTR pop 2 test
 200         """
 201         self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_POP_2)
 202         self.vtr_test(self.pg1, [])
 203
 204     def test_1ad_vtr_push_1ad(self):
 205         """ 1AD VTR push 1 1AD test
 206         """
 207         self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_1, tag=300)
 208         self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=300),
 209                                  Tag(dot1=DOT1AD, vlan=200),
 210                                  Tag(dot1=DOT1Q, vlan=100)])
 211
 212     def test_1ad_vtr_push_2ad(self):
 213         """ 1AD VTR push 2 1AD test
 214         """
 215         self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_2, outer=400, inner=300)
 216         self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=400),
 217                                  Tag(dot1=DOT1Q, vlan=300),
 218                                  Tag(dot1=DOT1AD, vlan=200),
 219                                  Tag(dot1=DOT1Q, vlan=100)])
 220
 221     def test_1ad_vtr_push_1q(self):
 222         """ 1AD VTR push 1 1Q test
 223         """
 224         self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_1, tag=300, push1q=1)
 225         self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=300),
 226                                  Tag(dot1=DOT1AD, vlan=200),
 227                                  Tag(dot1=DOT1Q, vlan=100)])
 228
 229     def test_1ad_vtr_push_2q(self):
 230         """ 1AD VTR push 2 1Q test
 231         """
 232         self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_2,
 233                                 outer=400, inner=300, push1q=1)
 234         self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=400),
 235                                  Tag(dot1=DOT1Q, vlan=300),
 236                                  Tag(dot1=DOT1AD, vlan=200),
 237                                  Tag(dot1=DOT1Q, vlan=100)])
 238
 239     def test_1ad_vtr_translate_1_1ad(self):
 240         """ 1AD VTR translate 1 -> 1 1AD test
 241         """
 242         self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_1_1, tag=300)
 243         self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=300),
 244                                  Tag(dot1=DOT1Q, vlan=100)])
 245
 246     def test_1ad_vtr_translate_1_2ad(self):
 247         """ 1AD VTR translate 1 -> 2 1AD test
 248         """
 249         self.pg1.sub_if.set_vtr(
 250             L2_VTR_OP.L2_TRANSLATE_1_2, inner=300, outer=400)
 251         self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=400),
 252                                  Tag(dot1=DOT1Q, vlan=300),
 253                                  Tag(dot1=DOT1Q, vlan=100)])
 254
 255     def test_1ad_vtr_translate_2_1ad(self):
 256         """ 1AD VTR translate 2 -> 1 1AD test
 257         """
 258         self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_2_1, tag=300)
 259         self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=300)])
 260
 261     def test_1ad_vtr_translate_2_2ad(self):
 262         """ 1AD VTR translate 2 -> 2 1AD test
 263         """
 264         self.pg1.sub_if.set_vtr(
 265             L2_VTR_OP.L2_TRANSLATE_2_2, inner=300, outer=400)
 266         self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=400),
 267                                  Tag(dot1=DOT1Q, vlan=300)])
 268
 269     def test_1ad_vtr_translate_1_1q(self):
 270         """ 1AD VTR translate 1 -> 1 1Q test
 271         """
 272         self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_1_1, tag=300, push1q=1)
 273         self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=300),
 274                                  Tag(dot1=DOT1Q, vlan=100)])
 275
 276     def test_1ad_vtr_translate_1_2q(self):
 277         """ 1AD VTR translate 1 -> 2 1Q test
 278         """
 279         self.pg1.sub_if.set_vtr(
 280             L2_VTR_OP.L2_TRANSLATE_1_2, inner=300, outer=400, push1q=1)
 281         self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=400),
 282                                  Tag(dot1=DOT1Q, vlan=300),
 283                                  Tag(dot1=DOT1Q, vlan=100)])
 284
 285     def test_1ad_vtr_translate_2_1q(self):
 286         """ 1AD VTR translate 2 -> 1 1Q test
 287         """
 288         self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_2_1, tag=300, push1q=1)
 289         self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=300)])
 290
 291     def test_1ad_vtr_translate_2_2q(self):
 292         """ 1AD VTR translate 2 -> 2 1Q test
 293         """
 294         self.pg1.sub_if.set_vtr(
 295             L2_VTR_OP.L2_TRANSLATE_2_2, inner=300, outer=400, push1q=1)
 296         self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=400),
 297                                  Tag(dot1=DOT1Q, vlan=300)])
 298
 299     def test_1q_vtr_pop_1(self):
 300         """ 1Q VTR pop 1 test
 301         """
 302         self.pg2.sub_if.set_vtr(L2_VTR_OP.L2_POP_1)
 303         self.vtr_test(self.pg2, [])
 304
 305     def test_1q_vtr_push_1(self):
 306         """ 1Q VTR push 1 test
 307         """
 308         self.pg2.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_1, tag=300)
 309         self.vtr_test(self.pg2, [Tag(dot1=DOT1AD, vlan=300),
 310                                  Tag(dot1=DOT1Q, vlan=200)])
 311
 312     def test_1q_vtr_push_2(self):
 313         """ 1Q VTR push 2 test
 314         """
 315         self.pg2.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_2, outer=400, inner=300)
 316         self.vtr_test(self.pg2, [Tag(dot1=DOT1AD, vlan=400),
 317                                  Tag(dot1=DOT1Q, vlan=300),
 318                                  Tag(dot1=DOT1Q, vlan=200)])
 319
 320     def test_1q_vtr_translate_1_1(self):
 321         """ 1Q VTR translate 1 -> 1 test
 322         """
 323         self.pg2.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_1_1, tag=300)
 324         self.vtr_test(self.pg2, [Tag(dot1=DOT1AD, vlan=300)])
 325
 326     def test_1q_vtr_translate_1_2(self):
 327         """ 1Q VTR translate 1 -> 2 test
 328         """
 329         self.pg2.sub_if.set_vtr(
 330             L2_VTR_OP.L2_TRANSLATE_1_2, inner=300, outer=400)
 331         self.vtr_test(self.pg2, [Tag(dot1=DOT1AD, vlan=400),
 332                                  Tag(dot1=DOT1Q, vlan=300)])
 333
 334     def test_if_vtr_disable(self):
 335         """ Disable VTR on non-sub-interfaces
 336         """
 337         # First set the VTR fields to junk
 338         self.vapi.l2_interface_vlan_tag_rewrite(
 339             sw_if_index=self.pg0.sw_if_index, vtr_op=L2_VTR_OP.L2_PUSH_2,
 340             push_dot1q=1, tag1=19, tag2=630)
 341
 342         if_state = self.vapi.sw_interface_dump(
 343             sw_if_index=self.pg0.sw_if_index)
 344         self.assertEqual(if_state[0].sw_if_index, self.pg0.sw_if_index)
 345         self.assertNotEqual(if_state[0].vtr_op, L2_VTR_OP.L2_DISABLED)
 346
 347         # Then ensure that a request to disable VTR is honored.
 348         self.vapi.l2_interface_vlan_tag_rewrite(
 349             sw_if_index=self.pg0.sw_if_index, vtr_op=L2_VTR_OP.L2_DISABLED)
 350
 351         if_state = self.vapi.sw_interface_dump(
 352             sw_if_index=self.pg0.sw_if_index)
 353         self.assertEqual(if_state[0].sw_if_index, self.pg0.sw_if_index)
 354         self.assertEqual(if_state[0].vtr_op, L2_VTR_OP.L2_DISABLED)
 355
 356     def test_if_vtr_push_1q(self):
 357         """ 1Q VTR push 1 on non-sub-interfaces
 358         """
 359         self.vapi.l2_interface_vlan_tag_rewrite(
 360             sw_if_index=self.pg0.sw_if_index, vtr_op=L2_VTR_OP.L2_PUSH_1,
 361             push_dot1q=1, tag1=150)
 362
 363         if_state = self.vapi.sw_interface_dump(
 364             sw_if_index=self.pg0.sw_if_index)
 365         self.assertEqual(if_state[0].sw_if_index, self.pg0.sw_if_index)
 366         self.assertEqual(if_state[0].vtr_op, L2_VTR_OP.L2_PUSH_1)
 367         self.assertEqual(if_state[0].vtr_tag1, 150)
 368         self.assertNotEqual(if_state[0].vtr_push_dot1q, 0)
 369
 370     def test_if_vtr_push_2ad(self):
 371         """ 1AD VTR push 2 on non-sub-interfaces
 372         """
 373         self.vapi.l2_interface_vlan_tag_rewrite(
 374             sw_if_index=self.pg0.sw_if_index, vtr_op=L2_VTR_OP.L2_PUSH_2,
 375             push_dot1q=0, tag1=450, tag2=350)
 376
 377         if_state = self.vapi.sw_interface_dump(
 378             sw_if_index=self.pg0.sw_if_index)
 379         self.assertEqual(if_state[0].sw_if_index, self.pg0.sw_if_index)
 380         self.assertEqual(if_state[0].vtr_op, L2_VTR_OP.L2_PUSH_2)
 381         self.assertEqual(if_state[0].vtr_tag1, 450)         # outer
 382         self.assertEqual(if_state[0].vtr_tag2, 350)         # inner
 383         self.assertEqual(if_state[0].vtr_push_dot1q, 0)
 384
 385 if __name__ == '__main__':
 386     unittest.main(testRunner=VppTestRunner)