+class VppIpTable(VppObject):
+
+ def __init__(self,
+ test,
+ table_id,
+ is_ip6=0):
+ self._test = test
+ self.table_id = table_id
+ self.is_ip6 = is_ip6
+
+ 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})
+ 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):
+ self._test = test
+ self.intf = intf
+ self.addr = addr
+ self.len = len
+ self.prefix = "%s/%d" % (addr, len)
+
+ 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):
+ return fib_interface_ip_prefix(self._test,
+ self.addr,
+ self.len,
+ self.intf.sw_if_index)
+
+ def object_id(self):
+ return "interface-ip-%s-%s" % (self.intf, self.prefix)
+
+
+class VppIpInterfaceBind(VppObject):
+
+ def __init__(self, test, intf, table):
+ self._test = test
+ self.intf = intf
+ self.table = table
+
+ def add_vpp_config(self):
+ if self.table.is_ip6:
+ self.intf.set_table_ip6(self.table.table_id)
+ else:
+ self.intf.set_table_ip4(self.table.table_id)
+ self._test.registry.register(self, self._test.logger)
+
+ def remove_vpp_config(self):
+ if 0 == self.table.table_id:
+ return
+ if self.table.is_ip6:
+ self.intf.set_table_ip6(0)
+ else:
+ self.intf.set_table_ip4(0)
+
+ def query_vpp_config(self):
+ if 0 == self.table.table_id:
+ return False
+ return self._test.vapi.sw_interface_get_table(
+ self.intf.sw_if_index,
+ self.table.is_ip6).vrf_id == self.table.table_id
+
+ def object_id(self):
+ return "interface-bind-%s-%s" % (self.intf, self.table)
+
+
+class VppMplsLabel(object):
+ def __init__(self, value, mode=MplsLspMode.PIPE, ttl=64, exp=0):
+ self.value = value
+ self.mode = mode
+ self.ttl = ttl
+ self.exp = exp
+
+ def encode(self):
+ is_uniform = 0 if self.mode is MplsLspMode.PIPE else 1
+ return {'label': self.value,
+ 'ttl': self.ttl,
+ 'exp': self.exp,
+ 'is_uniform': is_uniform}
+
+ def __eq__(self, other):
+ if isinstance(other, self.__class__):
+ return (self.value == other.value and
+ self.ttl == other.ttl and
+ self.exp == other.exp and
+ self.mode == other.mode)
+ elif hasattr(other, 'label'):
+ return (self.value == other.label and
+ self.ttl == other.ttl and
+ self.exp == other.exp and
+ (self.mode == MplsLspMode.UNIFORM) == other.is_uniform)
+ else:
+ return False
+
+ def __ne__(self, other):
+ return not (self == other)
+
+
+class VppFibPathNextHop(object):
+ def __init__(self, addr,
+ via_label=MPLS_LABEL_INVALID,
+ next_hop_id=INVALID_INDEX):
+ self.addr = VppIpAddressUnion(addr)
+ self.via_label = via_label
+ self.obj_id = next_hop_id
+
+ def encode(self):
+ if self.via_label is not MPLS_LABEL_INVALID:
+ return {'via_label': self.via_label}
+ if self.obj_id is not INVALID_INDEX:
+ return {'obj_id': self.obj_id}
+ else:
+ return {'address': self.addr.encode()}
+
+ def proto(self):
+ if self.via_label is MPLS_LABEL_INVALID:
+ return address_proto(self.addr)
+ else:
+ return FibPathProto.FIB_PATH_NH_PROTO_MPLS
+
+ def __eq__(self, other):
+ if not isinstance(other, self.__class__):
+ # try the other instance's __eq__.
+ return NotImplemented
+ return (self.addr == other.addr and
+ self.via_label == other.via_label and
+ self.obj_id == other.obj_id)
+
+