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