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 modify(self, paths, is_local=0,
  83                is_unreach=0, is_prohibit=0):
  84         self.paths = paths
  85         self.is_local = is_local
  86         self.is_unreach = is_unreach
  87         self.is_prohibit = is_prohibit
  88
  89     def add_vpp_config(self):
  90         if self.is_local or self.is_unreach or self.is_prohibit:
  91             self._test.vapi.ip_add_del_route(
  92                 self.dest_addr,
  93                 self.dest_addr_len,
  94                 inet_pton(AF_INET6, "::"),
  95                 0xffffffff,
  96                 is_local=self.is_local,
  97                 is_unreach=self.is_unreach,
  98                 is_prohibit=self.is_prohibit,
  99                 table_id=self.table_id,
 100                 is_ipv6=self.is_ip6)
 101         else:
 102             for path in self.paths:
 103                 self._test.vapi.ip_add_del_route(
 104                     self.dest_addr,
 105                     self.dest_addr_len,
 106                     path.nh_addr,
 107                     path.nh_itf,
 108                     table_id=self.table_id,
 109                     next_hop_out_label_stack=path.nh_labels,
 110                     next_hop_n_out_labels=len(
 111                         path.nh_labels),
 112                     next_hop_via_label=path.nh_via_label,
 113                     is_ipv6=self.is_ip6)
 114         self._test.registry.register(self, self._test.logger)
 115
 116     def remove_vpp_config(self):
 117         if self.is_local or self.is_unreach or self.is_prohibit:
 118             self._test.vapi.ip_add_del_route(
 119                 self.dest_addr,
 120                 self.dest_addr_len,
 121                 inet_pton(AF_INET6, "::"),
 122                 0xffffffff,
 123                 is_local=self.is_local,
 124                 is_unreach=self.is_unreach,
 125                 is_prohibit=self.is_prohibit,
 126                 is_add=0,
 127                 table_id=self.table_id,
 128                 is_ipv6=self.is_ip6)
 129         else:
 130             for path in self.paths:
 131                 self._test.vapi.ip_add_del_route(self.dest_addr,
 132                                                  self.dest_addr_len,
 133                                                  path.nh_addr,
 134                                                  path.nh_itf,
 135                                                  table_id=self.table_id,
 136                                                  is_add=0,
 137                                                  is_ipv6=self.is_ip6)
 138
 139     def query_vpp_config(self):
 140         return find_route(self._test,
 141                           self.dest_addr_p,
 142                           self.dest_addr_len,
 143                           self.table_id,
 144                           inet=AF_INET6 if self.is_ip6 == 1 else AF_INET)
 145
 146     def __str__(self):
 147         return self.object_id()
 148
 149     def object_id(self):
 150         return ("%d:%s/%d"
 151                 % (self.table_id,
 152                    self.dest_addr_p,
 153                    self.dest_addr_len))
 154
 155
 156 class VppIpMRoute(VppObject):
 157     """
 158     IP Multicast Route
 159     """
 160
 161     def __init__(self, test, src_addr, grp_addr,
 162                  grp_addr_len, e_flags, paths, table_id=0, is_ip6=0):
 163         self._test = test
 164         self.paths = paths
 165         self.grp_addr_len = grp_addr_len
 166         self.table_id = table_id
 167         self.e_flags = e_flags
 168         self.is_ip6 = is_ip6
 169
 170         if is_ip6:
 171             self.grp_addr = inet_pton(AF_INET6, grp_addr)
 172             self.src_addr = inet_pton(AF_INET6, src_addr)
 173         else:
 174             self.grp_addr = inet_pton(AF_INET, grp_addr)
 175             self.src_addr = inet_pton(AF_INET, src_addr)
 176
 177     def add_vpp_config(self):
 178         for path in self.paths:
 179             self._test.vapi.ip_mroute_add_del(self.src_addr,
 180                                               self.grp_addr,
 181                                               self.grp_addr_len,
 182                                               self.e_flags,
 183                                               path.nh_itf,
 184                                               path.nh_i_flags,
 185                                               table_id=self.table_id,
 186                                               is_ipv6=self.is_ip6)
 187         self._test.registry.register(self, self._test.logger)
 188
 189     def remove_vpp_config(self):
 190         for path in self.paths:
 191             self._test.vapi.ip_mroute_add_del(self.src_addr,
 192                                               self.grp_addr,
 193                                               self.grp_addr_len,
 194                                               self.e_flags,
 195                                               path.nh_itf,
 196                                               path.nh_i_flags,
 197                                               table_id=self.table_id,
 198                                               is_add=0,
 199                                               is_ipv6=self.is_ip6)
 200
 201     def update_entry_flags(self, flags):
 202         self.e_flags = flags
 203         self._test.vapi.ip_mroute_add_del(self.src_addr,
 204                                           self.grp_addr,
 205                                           self.grp_addr_len,
 206                                           self.e_flags,
 207                                           0xffffffff,
 208                                           0,
 209                                           table_id=self.table_id,
 210                                           is_ipv6=self.is_ip6)
 211
 212     def update_path_flags(self, itf, flags):
 213         for path in self.paths:
 214             if path.nh_itf == itf:
 215                 path.nh_i_flags = flags
 216                 break
 217         self._test.vapi.ip_mroute_add_del(self.src_addr,
 218                                           self.grp_addr,
 219                                           self.grp_addr_len,
 220                                           self.e_flags,
 221                                           path.nh_itf,
 222                                           path.nh_i_flags,
 223                                           table_id=self.table_id,
 224                                           is_ipv6=self.is_ip6)
 225
 226     def query_vpp_config(self):
 227         dump = self._test.vapi.ip_fib_dump()
 228         for e in dump:
 229             if self.grp_addr == e.address \
 230                and self.grp_addr_len == e.address_length \
 231                and self.table_id == e.table_id:
 232                 return True
 233         return False
 234
 235     def __str__(self):
 236         return self.object_id()
 237
 238     def object_id(self):
 239         if self.is_ip6:
 240             return ("%d:(%s,%s/%d)"
 241                     % (self.table_id,
 242                        inet_ntop(AF_INET6, self.src_addr),
 243                        inet_ntop(AF_INET6, self.grp_addr),
 244                        self.grp_addr_len))
 245         else:
 246             return ("%d:(%s,%s/%d)"
 247                     % (self.table_id,
 248                        inet_ntop(AF_INET, self.src_addr),
 249                        inet_ntop(AF_INET, self.grp_addr),
 250                        self.grp_addr_len))
 251
 252
 253 class VppMFibSignal(object):
 254     def __init__(self, test, route, interface, packet):
 255         self.route = route
 256         self.interface = interface
 257         self.packet = packet
 258         self.test = test
 259
 260     def compare(self, signal):
 261         self.test.assertEqual(self.interface, signal.sw_if_index)
 262         self.test.assertEqual(self.route.table_id, signal.table_id)
 263         self.test.assertEqual(self.route.grp_addr_len,
 264                               signal.grp_address_len)
 265         for i in range(self.route.grp_addr_len / 8):
 266             self.test.assertEqual(self.route.grp_addr[i],
 267                                   signal.grp_address[i])
 268         if (self.route.grp_addr_len > 32):
 269             for i in range(4):
 270                 self.test.assertEqual(self.route.src_addr[i],
 271                                       signal.src_address[i])
 272
 273
 274 class VppMplsIpBind(VppObject):
 275     """
 276     MPLS to IP Binding
 277     """
 278
 279     def __init__(self, test, local_label, dest_addr, dest_addr_len,
 280                  table_id=0, ip_table_id=0):
 281         self._test = test
 282         self.dest_addr = inet_pton(AF_INET, dest_addr)
 283         self.dest_addr_len = dest_addr_len
 284         self.local_label = local_label
 285         self.table_id = table_id
 286         self.ip_table_id = ip_table_id
 287
 288     def add_vpp_config(self):
 289         self._test.vapi.mpls_ip_bind_unbind(self.local_label,
 290                                             self.dest_addr,
 291                                             self.dest_addr_len,
 292                                             table_id=self.table_id,
 293                                             ip_table_id=self.ip_table_id)
 294         self._test.registry.register(self, self._test.logger)
 295
 296     def remove_vpp_config(self):
 297         self._test.vapi.mpls_ip_bind_unbind(self.local_label,
 298                                             self.dest_addr,
 299                                             self.dest_addr_len,
 300                                             is_bind=0)
 301
 302     def query_vpp_config(self):
 303         dump = self._test.vapi.mpls_fib_dump()
 304         for e in dump:
 305             if self.local_label == e.label \
 306                and self.eos_bit == e.eos_bit \
 307                and self.table_id == e.table_id:
 308                 return True
 309         return False
 310
 311     def __str__(self):
 312         return self.object_id()
 313
 314     def object_id(self):
 315         return ("%d:%s binds %d:%s/%d"
 316                 % (self.table_id,
 317                    self.local_label,
 318                    self.ip_table_id,
 319                    inet_ntop(AF_INET, self.dest_addr),
 320                    self.dest_addr_len))
 321
 322
 323 class VppMplsRoute(VppObject):
 324     """
 325     MPLS Route/LSP
 326     """
 327
 328     def __init__(self, test, local_label, eos_bit, paths, table_id=0):
 329         self._test = test
 330         self.paths = paths
 331         self.local_label = local_label
 332         self.eos_bit = eos_bit
 333         self.table_id = table_id
 334
 335     def add_vpp_config(self):
 336         for path in self.paths:
 337             self._test.vapi.mpls_route_add_del(
 338                 self.local_label,
 339                 self.eos_bit,
 340                 1,
 341                 path.nh_addr,
 342                 path.nh_itf,
 343                 table_id=self.table_id,
 344                 next_hop_out_label_stack=path.nh_labels,
 345                 next_hop_n_out_labels=len(
 346                     path.nh_labels),
 347                 next_hop_via_label=path.nh_via_label,
 348                 next_hop_table_id=path.nh_table_id)
 349         self._test.registry.register(self, self._test.logger)
 350
 351     def remove_vpp_config(self):
 352         for path in self.paths:
 353             self._test.vapi.mpls_route_add_del(self.local_label,
 354                                                self.eos_bit,
 355                                                1,
 356                                                path.nh_addr,
 357                                                path.nh_itf,
 358                                                table_id=self.table_id,
 359                                                is_add=0)
 360
 361     def query_vpp_config(self):
 362         dump = self._test.vapi.mpls_fib_dump()
 363         for e in dump:
 364             if self.local_label == e.label \
 365                and self.eos_bit == e.eos_bit \
 366                and self.table_id == e.table_id:
 367                 return True
 368         return False
 369
 370     def __str__(self):
 371         return self.object_id()
 372
 373     def object_id(self):
 374         return ("%d:%s/%d"
 375                 % (self.table_id,
 376                    self.local_label,
 377                    20+self.eos_bit))