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 VppObject
   8 from vpp_ip import DpoProto, INVALID_INDEX, VppIpAddressUnion, VppIpMPrefix
   9 from ipaddress import ip_network, ip_address, IPv4Network, IPv6Network
  10 from vpp_papi_exceptions import UnexpectedApiReturnValueError
  11
  12 # from vnet/vnet/mpls/mpls_types.h
  13 MPLS_IETF_MAX_LABEL = 0xFFFFF
  14 MPLS_LABEL_INVALID = MPLS_IETF_MAX_LABEL + 1
  15
  16 try:
  17     text_type = unicode
  18 except NameError:
  19     text_type = str
  20
  21
  22 class FibPathProto:
  23     FIB_PATH_NH_PROTO_IP4 = 0
  24     FIB_PATH_NH_PROTO_IP6 = 1
  25     FIB_PATH_NH_PROTO_MPLS = 2
  26     FIB_PATH_NH_PROTO_ETHERNET = 3
  27     FIB_PATH_NH_PROTO_BIER = 4
  28     FIB_PATH_NH_PROTO_NSH = 5
  29
  30
  31 class FibPathType:
  32     FIB_PATH_TYPE_NORMAL = 0
  33     FIB_PATH_TYPE_LOCAL = 1
  34     FIB_PATH_TYPE_DROP = 2
  35     FIB_PATH_TYPE_UDP_ENCAP = 3
  36     FIB_PATH_TYPE_BIER_IMP = 4
  37     FIB_PATH_TYPE_ICMP_UNREACH = 5
  38     FIB_PATH_TYPE_ICMP_PROHIBIT = 6
  39     FIB_PATH_TYPE_SOURCE_LOOKUP = 7
  40     FIB_PATH_TYPE_DVR = 8
  41     FIB_PATH_TYPE_INTERFACE_RX = 9
  42     FIB_PATH_TYPE_CLASSIFY = 10
  43
  44
  45 class FibPathFlags:
  46     FIB_PATH_FLAG_NONE = 0
  47     FIB_PATH_FLAG_RESOLVE_VIA_ATTACHED = 1
  48     FIB_PATH_FLAG_RESOLVE_VIA_HOST = 2
  49     FIB_PATH_FLAG_POP_PW_CW = 4
  50
  51
  52 class MplsLspMode:
  53     PIPE = 0
  54     UNIFORM = 1
  55
  56
  57 def mk_network(addr, len):
  58     if ip_address(text_type(addr)).version == 4:
  59         return IPv4Network("%s/%d" % (addr, len), strict=False)
  60     else:
  61         return IPv6Network("%s/%d" % (addr, len), strict=False)
  62
  63
  64 def ip_to_dpo_proto(addr):
  65     if addr.version == 6:
  66         return DpoProto.DPO_PROTO_IP6
  67     else:
  68         return DpoProto.DPO_PROTO_IP4
  69
  70
  71 def address_proto(ip_addr):
  72     if ip_addr.ip_addr.version == 4:
  73         return FibPathProto.FIB_PATH_NH_PROTO_IP4
  74     else:
  75         return FibPathProto.FIB_PATH_NH_PROTO_IP6
  76
  77
  78 def find_route(
  79     test, addr, len, table_id=0, sw_if_index=None, ignore_default_route=False
  80 ):
  81     prefix = mk_network(addr, len)
  82
  83     if 4 == prefix.version:
  84         routes = test.vapi.ip_route_dump(table_id, False)
  85     else:
  86         routes = test.vapi.ip_route_dump(table_id, True)
  87
  88     for e in routes:
  89         if table_id == e.route.table_id and str(e.route.prefix) == str(prefix):
  90             if not sw_if_index:
  91                 # if the route is a default one of the table:
  92                 # 0.0.0.0/0, 0.0.0.0/32, 240.0.0.0/4, 255.255.255.255/32
  93                 return not (
  94                     ignore_default_route
  95                     and e.route.n_paths == 1
  96                     and e.route.paths[0].type == FibPathType.FIB_PATH_TYPE_DROP
  97                 )
  98             else:
  99                 # should be only one path if the user is looking
 100                 # for the interface the route is reachable through
 101                 if e.route.n_paths != 1:
 102                     return False
 103                 else:
 104                     return e.route.paths[0].sw_if_index == sw_if_index
 105
 106     return False
 107
 108
 109 def find_route_in_dump(dump, route, table):
 110     for r in dump:
 111         if table.table_id == r.route.table_id and route.prefix == r.route.prefix:
 112             if len(route.paths) == r.route.n_paths:
 113                 return True
 114     return False
 115
 116
 117 def find_mroute_in_dump(dump, route, table):
 118     for r in dump:
 119         if table.table_id == r.route.table_id and route.prefix == r.route.prefix:
 120             return True
 121     return False
 122
 123
 124 def find_mroute(test, grp_addr, src_addr, grp_addr_len, table_id=0):
 125     ip_mprefix = VppIpMPrefix(text_type(src_addr), text_type(grp_addr), grp_addr_len)
 126
 127     if 4 == ip_mprefix.version:
 128         routes = test.vapi.ip_mroute_dump(table_id, False)
 129     else:
 130         routes = test.vapi.ip_mroute_dump(table_id, True)
 131
 132     for e in routes:
 133         if table_id == e.route.table_id and ip_mprefix == e.route.prefix:
 134             return True
 135     return False
 136
 137
 138 def find_mpls_route(test, table_id, label, eos_bit, paths=None):
 139     dump = test.vapi.mpls_route_dump(table_id)
 140     for e in dump:
 141         if (
 142             label == e.mr_route.mr_label
 143             and eos_bit == e.mr_route.mr_eos
 144             and table_id == e.mr_route.mr_table_id
 145         ):
 146             if not paths:
 147                 return True
 148             else:
 149                 if len(paths) != len(e.mr_route.mr_paths):
 150                     return False
 151                 for i in range(len(paths)):
 152                     if paths[i] != e.mr_route.mr_paths[i]:
 153                         return False
 154                 return True
 155     return False
 156
 157
 158 def fib_interface_ip_prefix(test, addr, len, sw_if_index):
 159     # can't use python net here since we need the host bits in the prefix
 160     prefix = "%s/%d" % (addr, len)
 161     addrs = test.vapi.ip_address_dump(
 162         sw_if_index, is_ipv6=(6 == ip_address(addr).version)
 163     )
 164
 165     for a in addrs:
 166         if a.sw_if_index == sw_if_index and str(a.prefix) == prefix:
 167             return True
 168     return False
 169
 170
 171 class VppIpTable(VppObject):
 172     def __init__(self, test, table_id, is_ip6=0, register=True):
 173         self._test = test
 174         self.table_id = table_id
 175         self.is_ip6 = is_ip6
 176         self.register = register
 177
 178     def add_vpp_config(self):
 179         self._test.vapi.ip_table_add_del(
 180             is_add=1, table={"is_ip6": self.is_ip6, "table_id": self.table_id}
 181         )
 182         if self.register:
 183             self._test.registry.register(self, self._test.logger)
 184         return self
 185
 186     def remove_vpp_config(self):
 187         self._test.vapi.ip_table_add_del(
 188             is_add=0, table={"is_ip6": self.is_ip6, "table_id": self.table_id}
 189         )
 190
 191     def replace_begin(self):
 192         self._test.vapi.ip_table_replace_begin(
 193             table={"is_ip6": self.is_ip6, "table_id": self.table_id}
 194         )
 195
 196     def replace_end(self):
 197         self._test.vapi.ip_table_replace_end(
 198             table={"is_ip6": self.is_ip6, "table_id": self.table_id}
 199         )
 200
 201     def flush(self):
 202         self._test.vapi.ip_table_flush(
 203             table={"is_ip6": self.is_ip6, "table_id": self.table_id}
 204         )
 205
 206     def dump(self):
 207         return self._test.vapi.ip_route_dump(self.table_id, self.is_ip6)
 208
 209     def mdump(self):
 210         return self._test.vapi.ip_mroute_dump(self.table_id, self.is_ip6)
 211
 212     def query_vpp_config(self):
 213         if self.table_id == 0:
 214             # the default table always exists
 215             return False
 216         # find the default route
 217         return find_route(
 218             self._test, "::" if self.is_ip6 else "0.0.0.0", 0, self.table_id
 219         )
 220
 221     def object_id(self):
 222         return "table-%s-%d" % ("v6" if self.is_ip6 == 1 else "v4", self.table_id)
 223
 224
 225 class VppIpInterfaceAddress(VppObject):
 226     def __init__(self, test, intf, addr, len, bind=None):
 227         self._test = test
 228         self.intf = intf
 229         self.addr = addr
 230         self.len = len
 231         self.prefix = "%s/%d" % (addr, len)
 232         self.host_len = ip_network(self.prefix, strict=False).max_prefixlen
 233         self.table_id = 0
 234         if bind:
 235             self.table_id = bind.table.table_id
 236
 237     def add_vpp_config(self):
 238         self._test.vapi.sw_interface_add_del_address(
 239             sw_if_index=self.intf.sw_if_index, prefix=self.prefix, is_add=1
 240         )
 241         self._test.registry.register(self, self._test.logger)
 242         return self
 243
 244     def remove_vpp_config(self):
 245         self._test.vapi.sw_interface_add_del_address(
 246             sw_if_index=self.intf.sw_if_index, prefix=self.prefix, is_add=0
 247         )
 248
 249     def query_vpp_config(self):
 250         # search for the IP address mapping and the two expected
 251         # FIB entries
 252         v = ip_address(self.addr).version
 253
 254         if (v == 4 and self.len < 31) or (v == 6 and self.len < 127):
 255             return (
 256                 fib_interface_ip_prefix(
 257                     self._test, self.addr, self.len, self.intf.sw_if_index
 258                 )
 259                 & find_route(
 260                     self._test,
 261                     self.addr,
 262                     self.len,
 263                     table_id=self.table_id,
 264                     sw_if_index=self.intf.sw_if_index,
 265                 )
 266                 & find_route(
 267                     self._test,
 268                     self.addr,
 269                     self.host_len,
 270                     table_id=self.table_id,
 271                     sw_if_index=self.intf.sw_if_index,
 272                 )
 273             )
 274         else:
 275             return fib_interface_ip_prefix(
 276                 self._test, self.addr, self.len, self.intf.sw_if_index
 277             ) & find_route(
 278                 self._test,
 279                 self.addr,
 280                 self.host_len,
 281                 table_id=self.table_id,
 282                 sw_if_index=self.intf.sw_if_index,
 283             )
 284
 285     def object_id(self):
 286         return "interface-ip-%s-%d-%s" % (self.intf, self.table_id, self.prefix)
 287
 288
 289 class VppIp6LinkLocalAddress(VppObject):
 290     def __init__(self, test, intf, addr):
 291         self._test = test
 292         self.intf = intf
 293         self.addr = addr
 294
 295     def add_vpp_config(self):
 296         self._test.vapi.sw_interface_ip6_set_link_local_address(
 297             sw_if_index=self.intf.sw_if_index, ip=self.addr
 298         )
 299         self._test.registry.register(self, self._test.logger)
 300         return self
 301
 302     def remove_vpp_config(self):
 303         # link locals can't be removed, only changed
 304         pass
 305
 306     def query_vpp_config(self):
 307         # no API to query
 308         return False
 309
 310     def object_id(self):
 311         return "ip6-link-local-%s-%s" % (self.intf, self.addr)
 312
 313
 314 class VppIpInterfaceBind(VppObject):
 315     def __init__(self, test, intf, table):
 316         self._test = test
 317         self.intf = intf
 318         self.table = table
 319
 320     def add_vpp_config(self):
 321         if self.table.is_ip6:
 322             self.intf.set_table_ip6(self.table.table_id)
 323         else:
 324             self.intf.set_table_ip4(self.table.table_id)
 325         self._test.registry.register(self, self._test.logger)
 326         return self
 327
 328     def remove_vpp_config(self):
 329         if 0 == self.table.table_id:
 330             return
 331         if self.table.is_ip6:
 332             self.intf.set_table_ip6(0)
 333         else:
 334             self.intf.set_table_ip4(0)
 335
 336     def query_vpp_config(self):
 337         if 0 == self.table.table_id:
 338             return False
 339         try:
 340             return (
 341                 self._test.vapi.sw_interface_get_table(
 342                     self.intf.sw_if_index, self.table.is_ip6
 343                 ).vrf_id
 344                 == self.table.table_id
 345             )
 346         except UnexpectedApiReturnValueError as e:
 347             if e.retval == -2:  # INVALID_SW_IF_INDEX
 348                 return False
 349             raise
 350
 351     def object_id(self):
 352         return "interface-bind-%s-%s" % (self.intf, self.table)
 353
 354
 355 class VppMplsLabel:
 356     def __init__(self, value, mode=MplsLspMode.PIPE, ttl=64, exp=0):
 357         self.value = value
 358         self.mode = mode
 359         self.ttl = ttl
 360         self.exp = exp
 361
 362     def encode(self):
 363         is_uniform = 0 if self.mode is MplsLspMode.PIPE else 1
 364         return {
 365             "label": self.value,
 366             "ttl": self.ttl,
 367             "exp": self.exp,
 368             "is_uniform": is_uniform,
 369         }
 370
 371     def __eq__(self, other):
 372         if isinstance(other, self.__class__):
 373             return (
 374                 self.value == other.value
 375                 and self.ttl == other.ttl
 376                 and self.exp == other.exp
 377                 and self.mode == other.mode
 378             )
 379         elif hasattr(other, "label"):
 380             return (
 381                 self.value == other.label
 382                 and self.ttl == other.ttl
 383                 and self.exp == other.exp
 384                 and (self.mode == MplsLspMode.UNIFORM) == other.is_uniform
 385             )
 386         else:
 387             return False
 388
 389     def __ne__(self, other):
 390         return not (self == other)
 391
 392
 393 class VppFibPathNextHop:
 394     def __init__(self, addr, via_label=MPLS_LABEL_INVALID, next_hop_id=INVALID_INDEX):
 395         self.addr = VppIpAddressUnion(addr)
 396         self.via_label = via_label
 397         self.obj_id = next_hop_id
 398
 399     def encode(self):
 400         if self.via_label is not MPLS_LABEL_INVALID:
 401             return {"via_label": self.via_label}
 402         if self.obj_id is not INVALID_INDEX:
 403             return {"obj_id": self.obj_id}
 404         else:
 405             return {"address": self.addr.encode()}
 406
 407     def proto(self):
 408         if self.via_label is MPLS_LABEL_INVALID:
 409             return address_proto(self.addr)
 410         else:
 411             return FibPathProto.FIB_PATH_NH_PROTO_MPLS
 412
 413     def __eq__(self, other):
 414         if not isinstance(other, self.__class__):
 415             # try the other instance's __eq__.
 416             return NotImplemented
 417         return (
 418             self.addr == other.addr
 419             and self.via_label == other.via_label
 420             and self.obj_id == other.obj_id
 421         )
 422
 423
 424 class VppRoutePath:
 425     def __init__(
 426         self,
 427         nh_addr,
 428         nh_sw_if_index,
 429         nh_table_id=0,
 430         labels=[],
 431         nh_via_label=MPLS_LABEL_INVALID,
 432         rpf_id=0,
 433         next_hop_id=INVALID_INDEX,
 434         proto=None,
 435         flags=FibPathFlags.FIB_PATH_FLAG_NONE,
 436         type=FibPathType.FIB_PATH_TYPE_NORMAL,
 437     ):
 438         self.nh_itf = nh_sw_if_index
 439         self.nh_table_id = nh_table_id
 440         self.nh_labels = labels
 441         self.weight = 1
 442         self.rpf_id = rpf_id
 443         self.proto = proto
 444         self.flags = flags
 445         self.type = type
 446         self.nh = VppFibPathNextHop(nh_addr, nh_via_label, next_hop_id)
 447         if proto is None:
 448             self.proto = self.nh.proto()
 449         else:
 450             self.proto = proto
 451         self.next_hop_id = next_hop_id
 452
 453     def encode_labels(self):
 454         lstack = []
 455         for l in self.nh_labels:
 456             if type(l) == VppMplsLabel:
 457                 lstack.append(l.encode())
 458             else:
 459                 lstack.append({"label": l, "ttl": 255})
 460         while len(lstack) < 16:
 461             lstack.append({})
 462
 463         return lstack
 464
 465     def encode(self):
 466         return {
 467             "weight": 1,
 468             "preference": 0,
 469             "table_id": self.nh_table_id,
 470             "nh": self.nh.encode(),
 471             "next_hop_id": self.next_hop_id,
 472             "sw_if_index": self.nh_itf,
 473             "rpf_id": self.rpf_id,
 474             "proto": self.proto,
 475             "type": self.type,
 476             "flags": self.flags,
 477             "n_labels": len(self.nh_labels),
 478             "label_stack": self.encode_labels(),
 479         }
 480
 481     def __eq__(self, other):
 482         if isinstance(other, self.__class__):
 483             return self.nh == other.nh
 484         elif hasattr(other, "sw_if_index"):
 485             # vl_api_fib_path_t
 486             if len(self.nh_labels) != other.n_labels:
 487                 return False
 488             for i in range(len(self.nh_labels)):
 489                 if self.nh_labels[i] != other.label_stack[i]:
 490                     return False
 491             return self.nh_itf == other.sw_if_index
 492         else:
 493             return False
 494
 495     def __ne__(self, other):
 496         return not (self == other)
 497
 498
 499 class VppMRoutePath(VppRoutePath):
 500     def __init__(
 501         self,
 502         nh_sw_if_index,
 503         flags,
 504         nh=None,
 505         proto=FibPathProto.FIB_PATH_NH_PROTO_IP4,
 506         type=FibPathType.FIB_PATH_TYPE_NORMAL,
 507         bier_imp=INVALID_INDEX,
 508     ):
 509         if not nh:
 510             nh = "::" if proto is FibPathProto.FIB_PATH_NH_PROTO_IP6 else "0.0.0.0"
 511         super(VppMRoutePath, self).__init__(
 512             nh, nh_sw_if_index, proto=proto, type=type, next_hop_id=bier_imp
 513         )
 514         self.nh_i_flags = flags
 515         self.bier_imp = bier_imp
 516
 517     def encode(self):
 518         return {
 519             "path": super(VppMRoutePath, self).encode(),
 520             "itf_flags": self.nh_i_flags,
 521         }
 522
 523
 524 class VppIpRoute(VppObject):
 525     """
 526     IP Route
 527     """
 528
 529     def __init__(
 530         self, test, dest_addr, dest_addr_len, paths, table_id=0, register=True
 531     ):
 532         self._test = test
 533         self.paths = paths
 534         self.table_id = table_id
 535         self.prefix = mk_network(dest_addr, dest_addr_len)
 536         self.register = register
 537         self.stats_index = None
 538         self.modified = False
 539
 540         self.encoded_paths = []
 541         for path in self.paths:
 542             self.encoded_paths.append(path.encode())
 543
 544     def __eq__(self, other):
 545         if self.table_id == other.table_id and self.prefix == other.prefix:
 546             return True
 547         return False
 548
 549     def modify(self, paths):
 550         self.paths = paths
 551         self.encoded_paths = []
 552         for path in self.paths:
 553             self.encoded_paths.append(path.encode())
 554         self.modified = True
 555
 556         self._test.vapi.ip_route_add_del(
 557             route={
 558                 "table_id": self.table_id,
 559                 "prefix": self.prefix,
 560                 "n_paths": len(self.encoded_paths),
 561                 "paths": self.encoded_paths,
 562             },
 563             is_add=1,
 564             is_multipath=0,
 565         )
 566
 567     def add_vpp_config(self):
 568         r = self._test.vapi.ip_route_add_del(
 569             route={
 570                 "table_id": self.table_id,
 571                 "prefix": self.prefix,
 572                 "n_paths": len(self.encoded_paths),
 573                 "paths": self.encoded_paths,
 574             },
 575             is_add=1,
 576             is_multipath=0,
 577         )
 578         self.stats_index = r.stats_index
 579         if self.register:
 580             self._test.registry.register(self, self._test.logger)
 581         return self
 582
 583     def remove_vpp_config(self):
 584         # there's no need to issue different deletes for modified routes
 585         # we do this only to test the two different ways to delete routes
 586         # eiter by passing all the paths to remove and mutlipath=1 or
 587         # passing no paths and multipath=0
 588         if self.modified:
 589             self._test.vapi.ip_route_add_del(
 590                 route={
 591                     "table_id": self.table_id,
 592                     "prefix": self.prefix,
 593                     "n_paths": len(self.encoded_paths),
 594                     "paths": self.encoded_paths,
 595                 },
 596                 is_add=0,
 597                 is_multipath=1,
 598             )
 599         else:
 600             self._test.vapi.ip_route_add_del(
 601                 route={"table_id": self.table_id, "prefix": self.prefix, "n_paths": 0},
 602                 is_add=0,
 603                 is_multipath=0,
 604             )
 605
 606     def query_vpp_config(self):
 607         return find_route(
 608             self._test,
 609             self.prefix.network_address,
 610             self.prefix.prefixlen,
 611             self.table_id,
 612             ignore_default_route=True,
 613         )
 614
 615     def object_id(self):
 616         return "%s:table-%d-%s" % (
 617             "ip6-route" if self.prefix.version == 6 else "ip-route",
 618             self.table_id,
 619             self.prefix,
 620         )
 621
 622     def get_stats_to(self):
 623         c = self._test.statistics.get_counter("/net/route/to")
 624         return c[0][self.stats_index]
 625
 626     def get_stats_via(self):
 627         c = self._test.statistics.get_counter("/net/route/via")
 628         return c[0][self.stats_index]
 629
 630
 631 class VppIpRouteV2(VppObject):
 632     """
 633     IP Route V2
 634     """
 635
 636     def __init__(
 637         self, test, dest_addr, dest_addr_len, paths, table_id=0, register=True, src=0
 638     ):
 639         self._test = test
 640         self.paths = paths
 641         self.table_id = table_id
 642         self.prefix = mk_network(dest_addr, dest_addr_len)
 643         self.register = register
 644         self.stats_index = None
 645         self.modified = False
 646         self.src = src
 647
 648         self.encoded_paths = []
 649         for path in self.paths:
 650             self.encoded_paths.append(path.encode())
 651
 652     def __eq__(self, other):
 653         if self.table_id == other.table_id and self.prefix == other.prefix:
 654             return True
 655         return False
 656
 657     def modify(self, paths):
 658         self.paths = paths
 659         self.encoded_paths = []
 660         for path in self.paths:
 661             self.encoded_paths.append(path.encode())
 662         self.modified = True
 663
 664         self._test.vapi.ip_route_add_del_v2(
 665             route={
 666                 "table_id": self.table_id,
 667                 "prefix": self.prefix,
 668                 "src": self.src,
 669                 "n_paths": len(self.encoded_paths),
 670                 "paths": self.encoded_paths,
 671             },
 672             is_add=1,
 673             is_multipath=0,
 674         )
 675
 676     def add_vpp_config(self):
 677         r = self._test.vapi.ip_route_add_del_v2(
 678             route={
 679                 "table_id": self.table_id,
 680                 "prefix": self.prefix,
 681                 "n_paths": len(self.encoded_paths),
 682                 "paths": self.encoded_paths,
 683                 "src": self.src,
 684             },
 685             is_add=1,
 686             is_multipath=0,
 687         )
 688         self.stats_index = r.stats_index
 689         if self.register:
 690             self._test.registry.register(self, self._test.logger)
 691         return self
 692
 693     def remove_vpp_config(self):
 694         # there's no need to issue different deletes for modified routes
 695         # we do this only to test the two different ways to delete routes
 696         # eiter by passing all the paths to remove and mutlipath=1 or
 697         # passing no paths and multipath=0
 698         if self.modified:
 699             self._test.vapi.ip_route_add_del_v2(
 700                 route={
 701                     "table_id": self.table_id,
 702                     "prefix": self.prefix,
 703                     "src": self.src,
 704                     "n_paths": len(self.encoded_paths),
 705                     "paths": self.encoded_paths,
 706                 },
 707                 is_add=0,
 708                 is_multipath=1,
 709             )
 710         else:
 711             self._test.vapi.ip_route_add_del_v2(
 712                 route={
 713                     "table_id": self.table_id,
 714                     "prefix": self.prefix,
 715                     "src": self.src,
 716                     "n_paths": 0,
 717                 },
 718                 is_add=0,
 719                 is_multipath=0,
 720             )
 721
 722     def query_vpp_config(self):
 723         return find_route(
 724             self._test,
 725             self.prefix.network_address,
 726             self.prefix.prefixlen,
 727             self.table_id,
 728             ignore_default_route=True,
 729         )
 730
 731     def object_id(self):
 732         return "%s:table-%d-%s" % (
 733             "ip6-route" if self.prefix.version == 6 else "ip-route",
 734             self.table_id,
 735             self.prefix,
 736         )
 737
 738     def get_stats_to(self):
 739         c = self._test.statistics.get_counter("/net/route/to")
 740         return c[0][self.stats_index]
 741
 742     def get_stats_via(self):
 743         c = self._test.statistics.get_counter("/net/route/via")
 744         return c[0][self.stats_index]
 745
 746
 747 class VppIpMRoute(VppObject):
 748     """
 749     IP Multicast Route
 750     """
 751
 752     def __init__(
 753         self,
 754         test,
 755         src_addr,
 756         grp_addr,
 757         grp_addr_len,
 758         e_flags,
 759         paths,
 760         table_id=0,
 761         rpf_id=0,
 762     ):
 763         self._test = test
 764         self.paths = paths
 765         self.table_id = table_id
 766         self.e_flags = e_flags
 767         self.rpf_id = rpf_id
 768
 769         self.prefix = VppIpMPrefix(src_addr, grp_addr, grp_addr_len)
 770         self.encoded_paths = []
 771         for path in self.paths:
 772             self.encoded_paths.append(path.encode())
 773
 774     def encode(self, paths=None):
 775         _paths = self.encoded_paths if paths is None else paths
 776         return {
 777             "table_id": self.table_id,
 778             "entry_flags": self.e_flags,
 779             "rpf_id": self.rpf_id,
 780             "prefix": self.prefix.encode(),
 781             "n_paths": len(_paths),
 782             "paths": _paths,
 783         }
 784
 785     def add_vpp_config(self):
 786         r = self._test.vapi.ip_mroute_add_del(
 787             route=self.encode(), is_multipath=1, is_add=1
 788         )
 789         self.stats_index = r.stats_index
 790         self._test.registry.register(self, self._test.logger)
 791         return self
 792
 793     def remove_vpp_config(self):
 794         self._test.vapi.ip_mroute_add_del(route=self.encode(), is_multipath=1, is_add=0)
 795
 796     def update_entry_flags(self, flags):
 797         self.e_flags = flags
 798         self._test.vapi.ip_mroute_add_del(
 799             route=self.encode(paths=[]), is_multipath=1, is_add=1
 800         )
 801
 802     def update_rpf_id(self, rpf_id):
 803         self.rpf_id = rpf_id
 804         self._test.vapi.ip_mroute_add_del(
 805             route=self.encode(paths=[]), is_multipath=1, is_add=1
 806         )
 807
 808     def update_path_flags(self, itf, flags):
 809         for p in range(len(self.paths)):
 810             if self.paths[p].nh_itf == itf:
 811                 self.paths[p].nh_i_flags = flags
 812                 self.encoded_paths[p] = self.paths[p].encode()
 813                 break
 814
 815         self._test.vapi.ip_mroute_add_del(
 816             route=self.encode(paths=[self.encoded_paths[p]]), is_add=1, is_multipath=0
 817         )
 818
 819     def query_vpp_config(self):
 820         return find_mroute(
 821             self._test,
 822             self.prefix.gaddr,
 823             self.prefix.saddr,
 824             self.prefix.length,
 825             self.table_id,
 826         )
 827
 828     def object_id(self):
 829         return "%d:(%s,%s/%d)" % (
 830             self.table_id,
 831             self.prefix.saddr,
 832             self.prefix.gaddr,
 833             self.prefix.length,
 834         )
 835
 836     def get_stats(self):
 837         c = self._test.statistics.get_counter("/net/mroute")
 838         return c[0][self.stats_index]
 839
 840
 841 class VppMFibSignal:
 842     def __init__(self, test, route, interface, packet):
 843         self.route = route
 844         self.interface = interface
 845         self.packet = packet
 846         self.test = test
 847
 848     def compare(self, signal):
 849         self.test.assertEqual(self.interface, signal.sw_if_index)
 850         self.test.assertEqual(self.route.table_id, signal.table_id)
 851         self.test.assertEqual(self.route.prefix, signal.prefix)
 852
 853
 854 class VppMplsIpBind(VppObject):
 855     """
 856     MPLS to IP Binding
 857     """
 858
 859     def __init__(
 860         self,
 861         test,
 862         local_label,
 863         dest_addr,
 864         dest_addr_len,
 865         table_id=0,
 866         ip_table_id=0,
 867         is_ip6=0,
 868     ):
 869         self._test = test
 870         self.dest_addr_len = dest_addr_len
 871         self.dest_addr = dest_addr
 872         self.ip_addr = ip_address(text_type(dest_addr))
 873         self.local_label = local_label
 874         self.table_id = table_id
 875         self.ip_table_id = ip_table_id
 876         self.prefix = mk_network(dest_addr, dest_addr_len)
 877
 878     def add_vpp_config(self):
 879         self._test.vapi.mpls_ip_bind_unbind(
 880             self.local_label,
 881             self.prefix,
 882             table_id=self.table_id,
 883             ip_table_id=self.ip_table_id,
 884         )
 885         self._test.registry.register(self, self._test.logger)
 886
 887     def remove_vpp_config(self):
 888         self._test.vapi.mpls_ip_bind_unbind(
 889             self.local_label,
 890             self.prefix,
 891             table_id=self.table_id,
 892             ip_table_id=self.ip_table_id,
 893             is_bind=0,
 894         )
 895
 896     def query_vpp_config(self):
 897         dump = self._test.vapi.mpls_route_dump(self.table_id)
 898         for e in dump:
 899             if (
 900                 self.local_label == e.mr_route.mr_label
 901                 and self.table_id == e.mr_route.mr_table_id
 902             ):
 903                 return True
 904         return False
 905
 906     def object_id(self):
 907         return "%d:%s binds %d:%s/%d" % (
 908             self.table_id,
 909             self.local_label,
 910             self.ip_table_id,
 911             self.dest_addr,
 912             self.dest_addr_len,
 913         )
 914
 915
 916 class VppMplsTable(VppObject):
 917     def __init__(self, test, table_id):
 918         self._test = test
 919         self.table_id = table_id
 920
 921     def add_vpp_config(self):
 922         self._test.vapi.mpls_table_add_del(self.table_id, is_add=1)
 923         self._test.registry.register(self, self._test.logger)
 924
 925     def remove_vpp_config(self):
 926         self._test.vapi.mpls_table_add_del(self.table_id, is_add=0)
 927
 928     def query_vpp_config(self):
 929         dump = self._test.vapi.mpls_table_dump()
 930         for d in dump:
 931             if d.mt_table.mt_table_id == self.table_id:
 932                 return True
 933         return False
 934
 935     def object_id(self):
 936         return "table-mpls-%d" % (self.table_id)
 937
 938
 939 class VppMplsRoute(VppObject):
 940     """
 941     MPLS Route/LSP
 942     """
 943
 944     def __init__(
 945         self,
 946         test,
 947         local_label,
 948         eos_bit,
 949         paths,
 950         table_id=0,
 951         is_multicast=0,
 952         eos_proto=FibPathProto.FIB_PATH_NH_PROTO_IP4,
 953     ):
 954         self._test = test
 955         self.paths = paths
 956         self.local_label = local_label
 957         self.eos_bit = eos_bit
 958         self.eos_proto = eos_proto
 959         self.table_id = table_id
 960         self.is_multicast = is_multicast
 961
 962     def add_vpp_config(self):
 963         paths = []
 964         for path in self.paths:
 965             paths.append(path.encode())
 966
 967         r = self._test.vapi.mpls_route_add_del(
 968             self.table_id,
 969             self.local_label,
 970             self.eos_bit,
 971             self.eos_proto,
 972             self.is_multicast,
 973             paths,
 974             1,
 975             0,
 976         )
 977         self.stats_index = r.stats_index
 978         self._test.registry.register(self, self._test.logger)
 979
 980     def remove_vpp_config(self):
 981         paths = []
 982         for path in self.paths:
 983             paths.append(path.encode())
 984
 985         self._test.vapi.mpls_route_add_del(
 986             self.table_id,
 987             self.local_label,
 988             self.eos_bit,
 989             self.eos_proto,
 990             self.is_multicast,
 991             paths,
 992             0,
 993             0,
 994         )
 995
 996     def query_vpp_config(self):
 997         return find_mpls_route(
 998             self._test, self.table_id, self.local_label, self.eos_bit
 999         )
1000
1001     def object_id(self):
1002         return "mpls-route-%d:%s/%d" % (
1003             self.table_id,
1004             self.local_label,
1005             20 + self.eos_bit,
1006         )
1007
1008     def get_stats_to(self):
1009         c = self._test.statistics.get_counter("/net/route/to")
1010         return c[0][self.stats_index]
1011
1012     def get_stats_via(self):
1013         c = self._test.statistics.get_counter("/net/route/via")
1014         return c[0][self.stats_index]