test/vpp_ip_route.py

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