+ @staticmethod
+ def vpp_interface_set_ip_address(
+ node, interface, address, prefix_length=None):
+ """Set IP address to VPP interface.
+
+ :param node: VPP node.
+ :param interface: Interface name.
+ :param address: IP address.
+ :param prefix_length: Prefix length.
+ :type node: dict
+ :type interface: str
+ :type address: str
+ :type prefix_length: int
+ """
+ ip_addr = ip_address(address)
+
+ cmd = u"sw_interface_add_del_address"
+ args = dict(
+ sw_if_index=InterfaceUtil.get_interface_index(node, interface),
+ is_add=True,
+ del_all=False,
+ prefix=IPUtil.create_prefix_object(
+ ip_addr,
+ prefix_length if prefix_length else 128
+ if ip_addr.version == 6 else 32
+ )
+ )
+ err_msg = f"Failed to add IP address on interface {interface}"
+
+ with PapiSocketExecutor(node) as papi_exec:
+ papi_exec.add(cmd, **args).get_reply(err_msg)
+
+ @staticmethod
+ def vpp_add_ip_neighbor(node, iface_key, ip_addr, mac_address):
+ """Add IP neighbor on DUT node.
+
+ :param node: VPP node.
+ :param iface_key: Interface key.
+ :param ip_addr: IP address of the interface.
+ :param mac_address: MAC address of the interface.
+ :type node: dict
+ :type iface_key: str
+ :type ip_addr: str
+ :type mac_address: str
+ """
+ dst_ip = ip_address(ip_addr)
+
+ neighbor = dict(
+ sw_if_index=Topology.get_interface_sw_index(node, iface_key),
+ flags=0,
+ mac_address=str(mac_address),
+ ip_address=str(dst_ip)
+ )
+ cmd = u"ip_neighbor_add_del"
+ args = dict(
+ is_add=1,
+ neighbor=neighbor
+ )
+ err_msg = f"Failed to add IP neighbor on interface {iface_key}"
+
+ with PapiSocketExecutor(node) as papi_exec:
+ papi_exec.add(cmd, **args).get_reply(err_msg)
+
+ @staticmethod
+ def union_addr(ip_addr):
+ """Creates union IP address.
+
+ :param ip_addr: IPv4 or IPv6 address.
+ :type ip_addr: IPv4Address or IPv6Address
+ :returns: Union IP address.
+ :rtype: dict
+ """
+ return dict(ip6=ip_addr.packed) if ip_addr.version == 6 \
+ else dict(ip4=ip_addr.packed)
+
+ @staticmethod
+ def create_ip_address_object(ip_addr):
+ """Create IP address object.
+
+ :param ip_addr: IPv4 or IPv6 address
+ :type ip_addr: IPv4Address or IPv6Address
+ :returns: IP address object.
+ :rtype: dict
+ """
+ return dict(
+ af=getattr(
+ AddressFamily, u"ADDRESS_IP6" if ip_addr.version == 6
+ else u"ADDRESS_IP4"
+ ).value,
+ un=IPUtil.union_addr(ip_addr)
+ )
+
+ @staticmethod
+ def create_prefix_object(ip_addr, addr_len):
+ """Create prefix object.
+
+ :param ip_addr: IPv4 or IPv6 address.
+ :para, addr_len: Length of IP address.
+ :type ip_addr: IPv4Address or IPv6Address
+ :type addr_len: int
+ :returns: Prefix object.
+ :rtype: dict
+ """
+ addr = IPUtil.create_ip_address_object(ip_addr)
+
+ return dict(
+ len=int(addr_len),
+ address=addr
+ )
+
+ @staticmethod
+ def compose_vpp_route_structure(node, network, prefix_len, **kwargs):
+ """Create route object for ip_route_add_del api call.
+
+ :param node: VPP node.
+ :param network: Route destination network address.
+ :param prefix_len: Route destination network prefix length.
+ :param kwargs: Optional key-value arguments:
+
+ gateway: Route gateway address. (str)
+ interface: Route interface. (str)
+ vrf: VRF table ID. (int)
+ count: number of IP addresses to add starting from network IP (int)
+ local: The route is local with same prefix (increment is 1).
+ If None, then is not used. (bool)
+ lookup_vrf: VRF table ID for lookup. (int)
+ multipath: Enable multipath routing. (bool)
+ weight: Weight value for unequal cost multipath routing. (int)
+
+ :type node: dict
+ :type network: str
+ :type prefix_len: int
+ :type kwargs: dict
+ :returns: route parameter basic structure
+ :rtype: dict
+ """
+ interface = kwargs.get(u"interface", u"")
+ gateway = kwargs.get(u"gateway", u"")
+
+ net_addr = ip_address(network)
+
+ prefix = IPUtil.create_prefix_object(net_addr, prefix_len)
+
+ paths = list()
+ n_hop = dict(
+ address=IPUtil.union_addr(ip_address(gateway)) if gateway else 0,
+ via_label=MPLS_LABEL_INVALID,
+ obj_id=Constants.BITWISE_NON_ZERO
+ )
+ path = dict(
+ sw_if_index=InterfaceUtil.get_interface_index(node, interface)
+ if interface else Constants.BITWISE_NON_ZERO,
+ table_id=int(kwargs.get(u"lookup_vrf", 0)),
+ rpf_id=Constants.BITWISE_NON_ZERO,
+ weight=int(kwargs.get(u"weight", 1)),
+ preference=1,
+ type=getattr(
+ FibPathType, u"FIB_PATH_TYPE_LOCAL"
+ if kwargs.get(u"local", False)
+ else u"FIB_PATH_TYPE_NORMAL"
+ ).value,
+ flags=getattr(FibPathFlags, u"FIB_PATH_FLAG_NONE").value,
+ proto=getattr(
+ FibPathNhProto, u"FIB_PATH_NH_PROTO_IP6"
+ if net_addr.version == 6
+ else u"FIB_PATH_NH_PROTO_IP4"
+ ).value,
+ nh=n_hop,
+ n_labels=0,
+ label_stack=list(0 for _ in range(16))
+ )
+ paths.append(path)
+
+ route = dict(
+ table_id=int(kwargs.get(u"vrf", 0)),
+ prefix=prefix,
+ n_paths=len(paths),
+ paths=paths
+ )
+ return route
+
+ @staticmethod
+ def vpp_route_add(node, network, prefix_len, **kwargs):
+ """Add route to the VPP node.
+
+ :param node: VPP node.
+ :param network: Route destination network address.
+ :param prefix_len: Route destination network prefix length.
+ :param kwargs: Optional key-value arguments:
+
+ gateway: Route gateway address. (str)
+ interface: Route interface. (str)
+ vrf: VRF table ID. (int)
+ count: number of IP addresses to add starting from network IP (int)
+ local: The route is local with same prefix (increment is 1).
+ If None, then is not used. (bool)
+ lookup_vrf: VRF table ID for lookup. (int)
+ multipath: Enable multipath routing. (bool)
+ weight: Weight value for unequal cost multipath routing. (int)