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