+def find_mpls_route(test, table_id, label, eos_bit, paths=None):
+ dump = test.vapi.mpls_route_dump(table_id)
+ for e in dump:
+ if (
+ label == e.mr_route.mr_label
+ and eos_bit == e.mr_route.mr_eos
+ and table_id == e.mr_route.mr_table_id
+ ):
+ if not paths:
+ return True
+ else:
+ if len(paths) != len(e.mr_route.mr_paths):
+ return False
+ for i in range(len(paths)):
+ if paths[i] != e.mr_route.mr_paths[i]:
+ return False
+ return True
+ return False
+
+
+def fib_interface_ip_prefix(test, addr, len, sw_if_index):
+ # can't use python net here since we need the host bits in the prefix
+ prefix = "%s/%d" % (addr, len)
+ addrs = test.vapi.ip_address_dump(
+ sw_if_index, is_ipv6=(6 == ip_address(addr).version)
+ )
+
+ for a in addrs:
+ if a.sw_if_index == sw_if_index and str(a.prefix) == prefix:
+ return True
+ return False
+
+
+class VppIpTable(VppObject):
+ def __init__(self, test, table_id, is_ip6=0, register=True):
+ self._test = test
+ self.table_id = table_id
+ self.is_ip6 = is_ip6
+ self.register = register
+
+ def add_vpp_config(self):
+ self._test.vapi.ip_table_add_del(
+ is_add=1, table={"is_ip6": self.is_ip6, "table_id": self.table_id}
+ )
+ if self.register:
+ self._test.registry.register(self, self._test.logger)
+ return self
+
+ def remove_vpp_config(self):
+ self._test.vapi.ip_table_add_del(
+ is_add=0, table={"is_ip6": self.is_ip6, "table_id": self.table_id}
+ )
+
+ def replace_begin(self):
+ self._test.vapi.ip_table_replace_begin(
+ table={"is_ip6": self.is_ip6, "table_id": self.table_id}
+ )
+
+ def replace_end(self):
+ self._test.vapi.ip_table_replace_end(
+ table={"is_ip6": self.is_ip6, "table_id": self.table_id}
+ )
+
+ def flush(self):
+ self._test.vapi.ip_table_flush(
+ table={"is_ip6": self.is_ip6, "table_id": self.table_id}
+ )
+
+ def dump(self):
+ return self._test.vapi.ip_route_dump(self.table_id, self.is_ip6)
+
+ def mdump(self):
+ return self._test.vapi.ip_mroute_dump(self.table_id, self.is_ip6)
+
+ def query_vpp_config(self):
+ if self.table_id == 0:
+ # the default table always exists
+ return False
+ # find the default route
+ return find_route(
+ self._test, "::" if self.is_ip6 else "0.0.0.0", 0, self.table_id
+ )
+
+ def object_id(self):
+ return "table-%s-%d" % ("v6" if self.is_ip6 == 1 else "v4", self.table_id)
+
+
+class VppIpInterfaceAddress(VppObject):
+ def __init__(self, test, intf, addr, len, bind=None):
+ self._test = test
+ self.intf = intf
+ self.addr = addr
+ self.len = len
+ self.prefix = "%s/%d" % (addr, len)
+ self.host_len = ip_network(self.prefix, strict=False).max_prefixlen
+ self.table_id = 0
+ if bind:
+ self.table_id = bind.table.table_id
+
+ def add_vpp_config(self):
+ self._test.vapi.sw_interface_add_del_address(
+ sw_if_index=self.intf.sw_if_index, prefix=self.prefix, is_add=1
+ )
+ self._test.registry.register(self, self._test.logger)
+ return self
+
+ def remove_vpp_config(self):
+ self._test.vapi.sw_interface_add_del_address(
+ sw_if_index=self.intf.sw_if_index, prefix=self.prefix, is_add=0
+ )
+
+ def query_vpp_config(self):
+ # search for the IP address mapping and the two expected
+ # FIB entries
+ v = ip_address(self.addr).version
+
+ if (v == 4 and self.len < 31) or (v == 6 and self.len < 127):
+ return (
+ fib_interface_ip_prefix(
+ self._test, self.addr, self.len, self.intf.sw_if_index
+ )
+ & find_route(
+ self._test,
+ self.addr,
+ self.len,
+ table_id=self.table_id,
+ sw_if_index=self.intf.sw_if_index,
+ )
+ & find_route(
+ self._test,
+ self.addr,
+ self.host_len,
+ table_id=self.table_id,
+ sw_if_index=self.intf.sw_if_index,
+ )
+ )
+ else:
+ return fib_interface_ip_prefix(
+ self._test, self.addr, self.len, self.intf.sw_if_index
+ ) & find_route(
+ self._test,
+ self.addr,
+ self.host_len,
+ table_id=self.table_id,
+ sw_if_index=self.intf.sw_if_index,
+ )
+
+ def object_id(self):
+ return "interface-ip-%s-%d-%s" % (self.intf, self.table_id, self.prefix)
+
+
+class VppIp6LinkLocalAddress(VppObject):
+ def __init__(self, test, intf, addr):
+ self._test = test
+ self.intf = intf
+ self.addr = addr
+
+ def add_vpp_config(self):
+ self._test.vapi.sw_interface_ip6_set_link_local_address(
+ sw_if_index=self.intf.sw_if_index, ip=self.addr
+ )
+ self._test.registry.register(self, self._test.logger)
+ return self
+
+ def remove_vpp_config(self):
+ # link locals can't be removed, only changed
+ pass
+
+ def query_vpp_config(self):
+ # no API to query
+ return False
+
+ def object_id(self):
+ return "ip6-link-local-%s-%s" % (self.intf, self.addr)
+