test/vpp_ip_route.py

   1 """
   2   IP Routes
   3
   4   object abstractions for representing IP routes in VPP
   5 """
   6
   7 from vpp_object import *
   8 from socket import inet_pton, inet_ntop, AF_INET, AF_INET6
   9
  10 # from vnet/vnet/mpls/mpls_types.h
  11 MPLS_IETF_MAX_LABEL = 0xfffff
  12 MPLS_LABEL_INVALID = MPLS_IETF_MAX_LABEL + 1
  13
  14
  15 def find_route(test, ip_addr, len, table_id=0, inet=AF_INET):
  16     if inet == AF_INET:
  17         s = 4
  18         routes = test.vapi.ip_fib_dump()
  19     else:
  20         s = 16
  21         routes = test.vapi.ip6_fib_dump()
  22
  23     route_addr = inet_pton(inet, ip_addr)
  24     for e in routes:
  25         if route_addr == e.address[:s] \
  26            and len == e.address_length \
  27            and table_id == e.table_id:
  28             return True
  29     return False
  30
  31
  32 class VppRoutePath(object):
  33
  34     def __init__(
  35             self,
  36             nh_addr,
  37             nh_sw_if_index,
  38             nh_table_id=0,
  39             labels=[],
  40             nh_via_label=MPLS_LABEL_INVALID,
  41             is_ip6=0):
  42         self.nh_itf = nh_sw_if_index
  43         self.nh_table_id = nh_table_id
  44         self.nh_via_label = nh_via_label
  45         self.nh_labels = labels
  46         if is_ip6:
  47             self.nh_addr = inet_pton(AF_INET6, nh_addr)
  48         else:
  49             self.nh_addr = inet_pton(AF_INET, nh_addr)
  50
  51
  52 class VppMRoutePath(VppRoutePath):
  53
  54     def __init__(self, nh_sw_if_index, flags):
  55         super(VppMRoutePath, self).__init__("0.0.0.0",
  56                                             nh_sw_if_index)
  57         self.nh_i_flags = flags
  58
  59
  60 class VppIpRoute(VppObject):
  61     """
  62     IP Route
  63     """
  64
  65     def __init__(self, test, dest_addr,
  66                  dest_addr_len, paths, table_id=0, is_ip6=0, is_local=0,
  67                  is_unreach=0, is_prohibit=0):
  68         self._test = test
  69         self.paths = paths
  70         self.dest_addr_len = dest_addr_len
  71         self.table_id = table_id
  72         self.is_ip6 = is_ip6
  73         self.is_local = is_local
  74         self.is_unreach = is_unreach
  75         self.is_prohibit = is_prohibit
  76         self.dest_addr_p = dest_addr
  77         if is_ip6:
  78             self.dest_addr = inet_pton(AF_INET6, dest_addr)
  79         else:
  80             self.dest_addr = inet_pton(AF_INET, dest_addr)
  81
  82     def add_vpp_config(self):
  83         if self.is_local or self.is_unreach or self.is_prohibit:
  84             self._test.vapi.ip_add_del_route(
  85                 self.dest_addr,
  86                 self.dest_addr_len,
  87                 inet_pton(AF_INET6, "::"),
  88                 0xffffffff,
  89                 is_local=self.is_local,
  90                 is_unreach=self.is_unreach,
  91                 is_prohibit=self.is_prohibit,
  92                 table_id=self.table_id,
  93                 is_ipv6=self.is_ip6)
  94         else:
  95             for path in self.paths:
  96                 self._test.vapi.ip_add_del_route(
  97                     self.dest_addr,
  98                     self.dest_addr_len,
  99                     path.nh_addr,
 100                     path.nh_itf,
 101                     table_id=self.table_id,
 102                     next_hop_out_label_stack=path.nh_labels,
 103                     next_hop_n_out_labels=len(
 104                         path.nh_labels),
 105                     next_hop_via_label=path.nh_via_label,
 106                     is_ipv6=self.is_ip6)
 107         self._test.registry.register(self, self._test.logger)
 108
 109     def remove_vpp_config(self):
 110         if self.is_local or self.is_unreach or self.is_prohibit:
 111             self._test.vapi.ip_add_del_route(
 112                 self.dest_addr,
 113                 self.dest_addr_len,
 114                 inet_pton(AF_INET6, "::"),
 115                 0xffffffff,
 116                 is_local=self.is_local,
 117                 is_unreach=self.is_unreach,
 118                 is_prohibit=self.is_prohibit,
 119                 is_add=0,
 120                 table_id=self.table_id,
 121                 is_ipv6=self.is_ip6)
 122         else:
 123             for path in self.paths:
 124                 self._test.vapi.ip_add_del_route(self.dest_addr,
 125                                                  self.dest_addr_len,
 126                                                  path.nh_addr,
 127                                                  path.nh_itf,
 128                                                  table_id=self.table_id,
 129                                                  is_add=0)
 130
 131     def query_vpp_config(self):
 132         return find_route(self._test,
 133                           self.dest_addr_p,
 134                           self.dest_addr_len,
 135                           self.table_id,
 136                           inet=AF_INET6 if self.is_ip6 == 1 else AF_INET)
 137
 138     def __str__(self):
 139         return self.object_id()
 140
 141     def object_id(self):
 142         return ("%d:%s/%d"
 143                 % (self.table_id,
 144                    self.dest_addr_p,
 145                    self.dest_addr_len))
 146
 147
 148 class VppIpMRoute(VppObject):
 149     """
 150     IP Multicast Route
 151     """
 152
 153     def __init__(self, test, src_addr, grp_addr,
 154                  grp_addr_len, e_flags, paths, table_id=0, is_ip6=0):
 155         self._test = test
 156         self.paths = paths
 157         self.grp_addr_len = grp_addr_len
 158         self.table_id = table_id
 159         self.e_flags = e_flags
 160         self.is_ip6 = is_ip6
 161
 162         if is_ip6:
 163             self.grp_addr = inet_pton(AF_INET6, grp_addr)
 164             self.src_addr = inet_pton(AF_INET6, src_addr)
 165         else:
 166             self.grp_addr = inet_pton(AF_INET, grp_addr)
 167             self.src_addr = inet_pton(AF_INET, src_addr)
 168
 169     def add_vpp_config(self):
 170         for path in self.paths:
 171             self._test.vapi.ip_mroute_add_del(self.src_addr,
 172                                               self.grp_addr,
 173                                               self.grp_addr_len,
 174                                               self.e_flags,
 175                                               path.nh_itf,
 176                                               path.nh_i_flags,
 177                                               table_id=self.table_id,
 178                                               is_ipv6=self.is_ip6)
 179         self._test.registry.register(self, self._test.logger)
 180
 181     def remove_vpp_config(self):
 182         for path in self.paths:
 183             self._test.vapi.ip_mroute_add_del(self.src_addr,
 184                                               self.grp_addr,
 185                                               self.grp_addr_len,
 186                                               self.e_flags,
 187                                               path.nh_itf,
 188                                               path.nh_i_flags,
 189                                               table_id=self.table_id,
 190                                               is_add=0,
 191                                               is_ipv6=self.is_ip6)
 192
 193     def update_entry_flags(self, flags):
 194         self.e_flags = flags
 195         self._test.vapi.ip_mroute_add_del(self.src_addr,
 196                                           self.grp_addr,
 197                                           self.grp_addr_len,
 198                                           self.e_flags,
 199                                           0xffffffff,
 200                                           0,
 201                                           table_id=self.table_id,
 202                                           is_ipv6=self.is_ip6)
 203
 204     def update_path_flags(self, itf, flags):
 205         for path in self.paths:
 206             if path.nh_itf == itf:
 207                 path.nh_i_flags = flags
 208                 break
 209         self._test.vapi.ip_mroute_add_del(self.src_addr,
 210                                           self.grp_addr,
 211                                           self.grp_addr_len,
 212                                           self.e_flags,
 213                                           path.nh_itf,
 214                                           path.nh_i_flags,
 215                                           table_id=self.table_id,
 216                                           is_ipv6=self.is_ip6)
 217
 218     def query_vpp_config(self):
 219         dump = self._test.vapi.ip_fib_dump()
 220         for e in dump:
 221             if self.grp_addr == e.address \
 222                and self.grp_addr_len == e.address_length \
 223                and self.table_id == e.table_id:
 224                 return True
 225         return False
 226
 227     def __str__(self):
 228         return self.object_id()
 229
 230     def object_id(self):
 231         if self.is_ip6:
 232             return ("%d:(%s,%s/%d)"
 233                     % (self.table_id,
 234                        inet_ntop(AF_INET6, self.src_addr),
 235                        inet_ntop(AF_INET6, self.grp_addr),
 236                        self.grp_addr_len))
 237         else:
 238             return ("%d:(%s,%s/%d)"
 239                     % (self.table_id,
 240                        inet_ntop(AF_INET, self.src_addr),
 241                        inet_ntop(AF_INET, self.grp_addr),
 242                        self.grp_addr_len))
 243
 244
 245 class VppMFibSignal(object):
 246     def __init__(self, test, route, interface, packet):
 247         self.route = route
 248         self.interface = interface
 249         self.packet = packet
 250         self.test = test
 251
 252     def compare(self, signal):
 253         self.test.assertEqual(self.interface, signal.sw_if_index)
 254         self.test.assertEqual(self.route.table_id, signal.table_id)
 255         self.test.assertEqual(self.route.grp_addr_len,
 256                               signal.grp_address_len)
 257         for i in range(self.route.grp_addr_len / 8):
 258             self.test.assertEqual(self.route.grp_addr[i],
 259                                   signal.grp_address[i])
 260         if (self.route.grp_addr_len > 32):
 261             for i in range(4):
 262                 self.test.assertEqual(self.route.src_addr[i],
 263                                       signal.src_address[i])
 264
 265
 266 class VppMplsIpBind(VppObject):
 267     """
 268     MPLS to IP Binding
 269     """
 270
 271     def __init__(self, test, local_label, dest_addr, dest_addr_len,
 272                  table_id=0, ip_table_id=0):
 273         self._test = test
 274         self.dest_addr = inet_pton(AF_INET, dest_addr)
 275         self.dest_addr_len = dest_addr_len
 276         self.local_label = local_label
 277         self.table_id = table_id
 278         self.ip_table_id = ip_table_id
 279
 280     def add_vpp_config(self):
 281         self._test.vapi.mpls_ip_bind_unbind(self.local_label,
 282                                             self.dest_addr,
 283                                             self.dest_addr_len,
 284                                             table_id=self.table_id,
 285                                             ip_table_id=self.ip_table_id)
 286         self._test.registry.register(self, self._test.logger)
 287
 288     def remove_vpp_config(self):
 289         self._test.vapi.mpls_ip_bind_unbind(self.local_label,
 290                                             self.dest_addr,
 291                                             self.dest_addr_len,
 292                                             is_bind=0)
 293
 294     def query_vpp_config(self):
 295         dump = self._test.vapi.mpls_fib_dump()
 296         for e in dump:
 297             if self.local_label == e.label \
 298                and self.eos_bit == e.eos_bit \
 299                and self.table_id == e.table_id:
 300                 return True
 301         return False
 302
 303     def __str__(self):
 304         return self.object_id()
 305
 306     def object_id(self):
 307         return ("%d:%s binds %d:%s/%d"
 308                 % (self.table_id,
 309                    self.local_label,
 310                    self.ip_table_id,
 311                    inet_ntop(AF_INET, self.dest_addr),
 312                    self.dest_addr_len))
 313
 314
 315 class VppMplsRoute(VppObject):
 316     """
 317     MPLS Route/LSP
 318     """
 319
 320     def __init__(self, test, local_label, eos_bit, paths, table_id=0):
 321         self._test = test
 322         self.paths = paths
 323         self.local_label = local_label
 324         self.eos_bit = eos_bit
 325         self.table_id = table_id
 326
 327     def add_vpp_config(self):
 328         for path in self.paths:
 329             self._test.vapi.mpls_route_add_del(
 330                 self.local_label,
 331                 self.eos_bit,
 332                 1,
 333                 path.nh_addr,
 334                 path.nh_itf,
 335                 table_id=self.table_id,
 336                 next_hop_out_label_stack=path.nh_labels,
 337                 next_hop_n_out_labels=len(
 338                     path.nh_labels),
 339                 next_hop_via_label=path.nh_via_label,
 340                 next_hop_table_id=path.nh_table_id)
 341         self._test.registry.register(self, self._test.logger)
 342
 343     def remove_vpp_config(self):
 344         for path in self.paths:
 345             self._test.vapi.mpls_route_add_del(self.local_label,
 346                                                self.eos_bit,
 347                                                1,
 348                                                path.nh_addr,
 349                                                path.nh_itf,
 350                                                table_id=self.table_id,
 351                                                is_add=0)
 352
 353     def query_vpp_config(self):
 354         dump = self._test.vapi.mpls_fib_dump()
 355         for e in dump:
 356             if self.local_label == e.label \
 357                and self.eos_bit == e.eos_bit \
 358                and self.table_id == e.table_id:
 359                 return True
 360         return False
 361
 362     def __str__(self):
 363         return self.object_id()
 364
 365     def object_id(self):
 366         return ("%d:%s/%d"
 367                 % (self.table_id,
 368                    self.local_label,
 369                    20+self.eos_bit))