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