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