+ :param node: VPP node.
+ :param interface: Interface name.
+ :param ip_addr_list: IP addresses.
+ :param prefix_length: Prefix length.
+ :type node: dict
+ :type interface: str
+ :type ip_addr_list: list
+ :type prefix_length: int
+ """
+ for ip_addr in ip_addr_list:
+ IPUtil.vpp_interface_set_ip_address(node, interface, ip_addr,
+ prefix_length)
+
+ @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=True,
+ 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 create_prefix_object(ip_addr, addr_len):
+ """Create prefix object.
+
+ :param ip_addr: IPv4 or IPv6 address.
+ :param addr_len: Length of IP address.
+ :type ip_addr: IPv4Address or IPv6Address
+ :type addr_len: int
+ :returns: Prefix object.
+ :rtype: dict
+ """
+ addr = IPAddress.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=IPAddress.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)
+
+ :type node: dict
+ :type network: str
+ :type prefix_len: int
+ :type kwargs: dict
+ """
+ count = kwargs.get(u"count", 1)
+
+ if count > 100:
+ gateway = kwargs.get(u"gateway", '')
+ interface = kwargs.get(u"interface", '')
+ vrf = kwargs.get(u"vrf", None)
+ multipath = kwargs.get(u"multipath", False)
+
+ with VatTerminal(node, json_param=False) as vat:
+
+ vat.vat_terminal_exec_cmd_from_template(
+ u"vpp_route_add.vat",
+ network=network,
+ prefix_length=prefix_len,
+ via=f"via {gateway}" if gateway else u"",
+ sw_if_index=f"sw_if_index "
+ f"{InterfaceUtil.get_interface_index(node, interface)}"
+ if interface else u"",
+ vrf=f"vrf {vrf}" if vrf else u"",
+ count=f"count {count}" if count else u"",
+ multipath=u"multipath" if multipath else u""
+ )
+ return
+
+ net_addr = ip_address(network)
+ cmd = u"ip_route_add_del"
+ args = dict(
+ is_add=True,
+ is_multipath=kwargs.get(u"multipath", False),
+ route=None
+ )
+ err_msg = f"Failed to add route(s) on host {node[u'host']}"
+
+ with PapiSocketExecutor(node) as papi_exec:
+ for i in range(kwargs.get(u"count", 1)):
+ args[u"route"] = IPUtil.compose_vpp_route_structure(
+ node, net_addr + i, prefix_len, **kwargs
+ )
+ history = bool(not 1 < i < kwargs.get(u"count", 1))
+ papi_exec.add(cmd, history=history, **args)
+ papi_exec.get_replies(err_msg)
+
+ @staticmethod
+ def flush_ip_addresses(node, interface):
+ """Flush all IP addresses from specified interface.
+
+ :param node: VPP node.
+ :param interface: Interface name.
+ :type node: dict
+ :type interface: str
+ """
+ cmd = u"sw_interface_add_del_address"
+ args = dict(
+ sw_if_index=InterfaceUtil.get_interface_index(node, interface),
+ is_add=False,
+ del_all=True
+ )
+ err_msg = f"Failed to flush IP address on interface {interface}"
+
+ with PapiSocketExecutor(node) as papi_exec:
+ papi_exec.add(cmd, **args).get_reply(err_msg)
+
+ @staticmethod
+ def add_fib_table(node, table_id, ipv6=False):
+ """Create new FIB table according to ID.
+
+ :param node: Node to add FIB on.
+ :param table_id: FIB table ID.
+ :param ipv6: Is this an IPv6 table
+ :type node: dict
+ :type table_id: int
+ :type ipv6: bool
+ """
+ cmd = u"ip_table_add_del"
+ table = dict(
+ table_id=int(table_id),
+ is_ip6=ipv6
+ )
+ args = dict(
+ table=table,
+ is_add=True
+ )
+ err_msg = f"Failed to add FIB table on host {node[u'host']}"
+
+ with PapiSocketExecutor(node) as papi_exec:
+ papi_exec.add(cmd, **args).get_reply(err_msg)
Code Review / csit.git / blobdiff