test/test_l3xc.py

   1 #!/usr/bin/env python
   2
   3 from socket import inet_pton, inet_ntop, AF_INET, AF_INET6
   4 import unittest
   5
   6 from framework import VppTestCase, VppTestRunner
   7 from vpp_ip import DpoProto
   8 from vpp_ip_route import VppIpRoute, VppRoutePath, VppMplsLabel, VppIpTable
   9
  10 from scapy.packet import Raw
  11 from scapy.layers.l2 import Ether
  12 from scapy.layers.inet import IP, UDP
  13 from scapy.layers.inet6 import IPv6
  14
  15 from vpp_object import VppObject
  16
  17 NUM_PKTS = 67
  18
  19
  20 def find_l3xc(test, sw_if_index, dump_sw_if_index=None):
  21     if not dump_sw_if_index:
  22         dump_sw_if_index = sw_if_index
  23     xcs = test.vapi.l3xc_dump(dump_sw_if_index)
  24     for xc in xcs:
  25         if sw_if_index == xc.l3xc.sw_if_index:
  26             return True
  27     return False
  28
  29
  30 class VppL3xc(VppObject):
  31
  32     def __init__(self,  test, intf, paths, is_ip6=False):
  33         self._test = test
  34         self.intf = intf
  35         self.is_ip6 = is_ip6
  36         self.paths = paths
  37
  38     def encode_paths(self):
  39         br_paths = []
  40         for p in self.paths:
  41             lstack = []
  42             for l in p.nh_labels:
  43                 if type(l) == VppMplsLabel:
  44                     lstack.append(l.encode())
  45                 else:
  46                     lstack.append({'label': l, 'ttl': 255})
  47             n_labels = len(lstack)
  48             while (len(lstack) < 16):
  49                 lstack.append({})
  50             br_paths.append({'next_hop': p.nh_addr,
  51                              'weight': 1,
  52                              'afi': p.proto,
  53                              'sw_if_index': p.nh_itf,
  54                              'preference': 0,
  55                              'table_id': p.nh_table_id,
  56                              'next_hop_id': p.next_hop_id,
  57                              'is_udp_encap': p.is_udp_encap,
  58                              'n_labels': n_labels,
  59                              'label_stack': lstack})
  60         return br_paths
  61
  62     def add_vpp_config(self):
  63         self._test.vapi.l3xc_update(
  64             l3xc={
  65                 'is_ip6': self.is_ip6,
  66                 'sw_if_index': self.intf.sw_if_index,
  67                 'n_paths': len(self.paths),
  68                 'paths': self.encode_paths()
  69             })
  70         self._test.registry.register(self, self._test.logger)
  71
  72     def remove_vpp_config(self):
  73         self._test.vapi.l3xc_del(
  74             is_ip6=self.is_ip6,
  75             sw_if_index=self.intf.sw_if_index)
  76
  77     def query_vpp_config(self):
  78         return find_l3xc(self._test, self.intf.sw_if_index)
  79
  80     def object_id(self):
  81         return ("l3xc-%d" % self.intf.sw_if_index)
  82
  83
  84 class TestL3xc(VppTestCase):
  85     """ L3XC Test Case """
  86
  87     @classmethod
  88     def setUpClass(cls):
  89         super(TestL3xc, cls).setUpClass()
  90
  91     @classmethod
  92     def tearDownClass(cls):
  93         super(TestL3xc, cls).tearDownClass()
  94
  95     def setUp(self):
  96         super(TestL3xc, self).setUp()
  97
  98         self.create_pg_interfaces(range(6))
  99
 100         for i in self.pg_interfaces:
 101             i.admin_up()
 102             i.config_ip4()
 103             i.resolve_arp()
 104             i.config_ip6()
 105             i.resolve_ndp()
 106
 107     def tearDown(self):
 108         for i in self.pg_interfaces:
 109             i.unconfig_ip4()
 110             i.unconfig_ip6()
 111             i.ip6_disable()
 112             i.admin_down()
 113         super(TestL3xc, self).tearDown()
 114
 115     def send_and_expect_load_balancing(self, input, pkts, outputs):
 116         self.pg_send(input, pkts)
 117         rxs = []
 118         for oo in outputs:
 119             rx = oo._get_capture(1)
 120             self.assertNotEqual(0, len(rx))
 121             for r in rx:
 122                 rxs.append(r)
 123         return rxs
 124
 125     def test_l3xc4(self):
 126         """ IPv4 X-Connect """
 127
 128         #
 129         # x-connect pg0 to pg1 and pg2 to pg3->5
 130         #
 131         l3xc_1 = VppL3xc(self, self.pg0,
 132                          [VppRoutePath(self.pg1.remote_ip4,
 133                                        self.pg1.sw_if_index)])
 134         l3xc_1.add_vpp_config()
 135         l3xc_2 = VppL3xc(self, self.pg2,
 136                          [VppRoutePath(self.pg3.remote_ip4,
 137                                        self.pg3.sw_if_index),
 138                           VppRoutePath(self.pg4.remote_ip4,
 139                                        self.pg4.sw_if_index),
 140                           VppRoutePath(self.pg5.remote_ip4,
 141                                        self.pg5.sw_if_index)])
 142         l3xc_2.add_vpp_config()
 143
 144         self.assertTrue(find_l3xc(self, self.pg2.sw_if_index, 0xffffffff))
 145
 146         self.logger.info(self.vapi.cli("sh l3xc"))
 147
 148         #
 149         # fire in packets. If it's forwarded then the L3XC was successful,
 150         # since default routing will drop it
 151         #
 152         p_1 = (Ether(src=self.pg0.remote_mac,
 153                      dst=self.pg0.local_mac) /
 154                IP(src="1.1.1.1", dst="1.1.1.2") /
 155                UDP(sport=1234, dport=1234) /
 156                Raw('\xa5' * 100))
 157         # self.send_and_expect(self.pg0, p_1*NUM_PKTS, self.pg1)
 158
 159         p_2 = []
 160         for ii in range(NUM_PKTS):
 161             p_2.append(Ether(src=self.pg0.remote_mac,
 162                              dst=self.pg0.local_mac) /
 163                        IP(src="1.1.1.1", dst="1.1.1.2") /
 164                        UDP(sport=1000 + ii, dport=1234) /
 165                        Raw('\xa5' * 100))
 166         self.send_and_expect_load_balancing(self.pg2, p_2,
 167                                             [self.pg3, self.pg4, self.pg5])
 168
 169         l3xc_2.remove_vpp_config()
 170         self.send_and_assert_no_replies(self.pg2, p_2)
 171
 172
 173 if __name__ == '__main__':
 174     unittest.main(testRunner=VppTestRunner)