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