-# Copyright (c) 2016 Cisco and/or its affiliates.
+# Copyright (c) 2018 Cisco and/or its affiliates.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at:
from yaml import load
from robot.api import logger
-from robot.libraries.BuiltIn import BuiltIn
+from robot.libraries.BuiltIn import BuiltIn, RobotNotRunningError
from robot.api.deco import keyword
-__all__ = ["DICT__nodes", 'Topology']
+__all__ = ["DICT__nodes", 'Topology', 'NodeType']
def load_topo_from_yaml():
"""Load topology from file defined in "${TOPOLOGY_PATH}" variable.
- :return: Nodes from loaded topology.
+ :returns: Nodes from loaded topology.
"""
- topo_path = BuiltIn().get_variable_value("${TOPOLOGY_PATH}")
+ try:
+ topo_path = BuiltIn().get_variable_value("${TOPOLOGY_PATH}")
+ except RobotNotRunningError:
+ return ''
with open(topo_path) as work_file:
return load(work_file.read())['nodes']
# pylint: disable=invalid-name
+
class NodeType(object):
"""Defines node types used in topology dictionaries."""
# Device Under Test (this node has VPP running on it)
"Active topology" contains initially data from the topology file and can be
extended with additional data from the DUTs like internal interface indexes
or names. Additional data which can be filled to the active topology are
+
- additional internal representation (index, name, ...)
- operational data (dynamic ports)
:param ptype: Port type, used as key prefix.
:type node: dict
:type ptype: str
- :return: Port key or None
+ :returns: Port key or None
:rtype: string or None
"""
max_ports = 1000000
break
return iface
+ @staticmethod
+ def remove_port(node, iface_key):
+ """Remove required port from active topology.
+
+ :param node: Node to remove port on.
+ :param: iface_key: Topology key of the interface.
+ :type node: dict
+ :type iface_key: str
+ :returns: Nothing
+ """
+ try:
+ node['interfaces'].pop(iface_key)
+ except KeyError:
+ pass
+
@staticmethod
def remove_all_ports(node, ptype):
"""Remove all ports with ptype as prefix.
:param: ptype: Port type, used as key prefix.
:type node: dict
:type ptype: str
- :return: Nothing
+ :returns: Nothing
"""
for if_key in list(node['interfaces']):
if if_key.startswith(str(ptype)):
node['interfaces'].pop(if_key)
+ @staticmethod
+ def remove_all_added_ports_on_all_duts_from_topology(nodes):
+ """Remove all added ports on all DUT nodes in the topology.
+
+ :param nodes: Nodes in the topology.
+ :type nodes: dict
+ :returns: Nothing
+ """
+ port_types = ('subinterface', 'vlan_subif', 'memif', 'tap', 'vhost',
+ 'loopback', 'gre_tunnel', 'vxlan_tunnel', 'eth_bond')
+
+ for node_data in nodes.values():
+ if node_data['type'] == NodeType.DUT:
+ for ptype in port_types:
+ Topology.remove_all_ports(node_data, ptype)
+
@staticmethod
def update_interface_sw_if_index(node, iface_key, sw_if_index):
"""Update sw_if_index on the interface from the node.
"""
node['interfaces'][iface_key]['vpp_sw_index'] = int(sw_if_index)
+ @staticmethod
+ def update_interface_name(node, iface_key, name):
+ """Update name on the interface from the node.
+
+ :param node: Node to update name on.
+ :param iface_key: Topology key of the interface.
+ :param name: Interface name to store.
+ :type node: dict
+ :type iface_key: str
+ :type name: str
+ """
+ node['interfaces'][iface_key]['name'] = str(name)
+
@staticmethod
def update_interface_mac_address(node, iface_key, mac_address):
"""Update mac_address on the interface from the node.
"""
node['interfaces'][iface_key]['mac_address'] = str(mac_address)
+ @staticmethod
+ def update_interface_pci_address(node, iface_key, pci_address):
+ """Update pci_address on the interface from the node.
+
+ :param node: Node to update PCI on.
+ :param iface_key: Topology key of the interface.
+ :param pci_address: PCI address.
+ :type node: dict
+ :type iface_key: str
+ :type pci_address: str
+ """
+ node['interfaces'][iface_key]['pci_address'] = str(pci_address)
+
@staticmethod
def update_interface_vhost_socket(node, iface_key, vhost_socket):
"""Update vhost socket name on the interface from the node.
"""
node['interfaces'][iface_key]['vhost_socket'] = str(vhost_socket)
+ @staticmethod
+ def update_interface_memif_socket(node, iface_key, memif_socket):
+ """Update memif socket name on the interface from the node.
+
+ :param node: Node to update socket name on.
+ :param iface_key: Topology key of the interface.
+ :param memif_socket: Path to named socket on node.
+ :type node: dict
+ :type iface_key: str
+ :type memif_socket: str
+ """
+ node['interfaces'][iface_key]['memif_socket'] = str(memif_socket)
+
+ @staticmethod
+ def update_interface_memif_id(node, iface_key, memif_id):
+ """Update memif ID on the interface from the node.
+
+ :param node: Node to update memif ID on.
+ :param iface_key: Topology key of the interface.
+ :param memif_id: Memif interface ID.
+ :type node: dict
+ :type iface_key: str
+ :type memif_id: str
+ """
+ node['interfaces'][iface_key]['memif_id'] = str(memif_id)
+
+ @staticmethod
+ def update_interface_memif_role(node, iface_key, memif_role):
+ """Update memif role on the interface from the node.
+
+ :param node: Node to update memif role on.
+ :param iface_key: Topology key of the interface.
+ :param memif_role: Memif role.
+ :type node: dict
+ :type iface_key: str
+ :type memif_role: str
+ """
+ node['interfaces'][iface_key]['memif_role'] = str(memif_role)
+
+ @staticmethod
+ def update_interface_tap_dev_name(node, iface_key, dev_name):
+ """Update device name on the tap interface from the node.
+
+ :param node: Node to update tap device name on.
+ :param iface_key: Topology key of the interface.
+ :param dev_name: Device name of the tap interface.
+ :type node: dict
+ :type iface_key: str
+ :type dev_name: str
+ :returns: Nothing
+ """
+ node['interfaces'][iface_key]['dev_name'] = str(dev_name)
+
@staticmethod
def get_node_by_hostname(nodes, hostname):
"""Get node from nodes of the topology by hostname.
:param hostname: Host name.
:type nodes: dict
:type hostname: str
- :return: Node dictionary or None if not found.
+ :returns: Node dictionary or None if not found.
"""
for node in nodes.values():
if node['host'] == hostname:
:param nodes: Nodes of the test topology.
:type nodes: dict
- :return: Links in the topology.
+ :returns: Links in the topology.
:rtype: list
"""
links = []
:type node: dict
:type key: string
:type value: string
- :return: Interface key from topology file
+ :returns: Interface key from topology file
:rtype: string
"""
interfaces = node['interfaces']
:param iface_name: Interface name (string form).
:type node: dict
:type iface_name: string
- :return: Interface key.
+ :returns: Interface key.
:rtype: str
"""
return Topology._get_interface_by_key_value(node, "name", iface_name)
:param link_name: Name of the link that a interface is connected to.
:type node: dict
:type link_name: string
- :return: Interface key of the interface connected to the given link.
+ :returns: Interface key of the interface connected to the given link.
:rtype: str
"""
return Topology._get_interface_by_key_value(node, "link", link_name)
:param node: The node topology directory.
:param link_names: List of names of the link that a interface is
- connected to.
+ connected to.
:type node: dict
:type link_names: list
- :return: Dictionary of interface names that are connected to the given
- links.
+ :returns: Dictionary of interface names that are connected to the given
+ links.
:rtype: dict
"""
retval = {}
:param sw_index: Sw_index of the link that a interface is connected to.
:type node: dict
:type sw_index: int
- :return: Interface name of the interface connected to the given link.
+ :returns: Interface name of the interface connected to the given link.
:rtype: str
"""
return Topology._get_interface_by_key_value(node, "vpp_sw_index",
:param iface_key: Interface key from topology file, or sw_index.
:type node: dict
:type iface_key: str/int
- :return: Return sw_if_index or None if not found.
+ :returns: Return sw_if_index or None if not found.
+ :rtype: int or None
"""
try:
if isinstance(iface_key, basestring):
return node['interfaces'][iface_key].get('vpp_sw_index')
- #FIXME: use only iface_key, do not use integer
- else:
- return int(iface_key)
+ # TODO: use only iface_key, do not use integer
+ return int(iface_key)
except (KeyError, ValueError):
return None
:param iface_name: Interface name.
:type node: dict
:type iface_name: str
- :return: Return sw_if_index or None if not found.
+ :returns: Return sw_if_index or None if not found.
:raises TypeError: If provided interface name is not a string.
"""
try:
- if isinstance(iface_name, basestring):
- iface_key = Topology.get_interface_by_name(node, iface_name)
- return node['interfaces'][iface_key].get('vpp_sw_index')
- else:
+ if not isinstance(iface_name, basestring):
raise TypeError("Interface name must be a string.")
+ iface_key = Topology.get_interface_by_name(node, iface_name)
+ return node['interfaces'][iface_key].get('vpp_sw_index')
except (KeyError, ValueError):
return None
:param iface_key: Interface key from topology file.
:type node: dict
:type iface_key: str
- :return: MTU or None if not found.
+ :returns: MTU or None if not found.
:rtype: int
"""
try:
:param iface_key: Interface key from topology file.
:type node: dict
:type iface_key: str
- :return: Interface name or None if not found.
- :rtype: int
+ :returns: Interface name or None if not found.
+ :rtype: str
"""
try:
return node['interfaces'][iface_key].get('name')
except KeyError:
return None
+ @staticmethod
+ def convert_interface_reference_to_key(node, interface):
+ """Takes interface reference in any format
+ (name, link name, interface key or sw_if_index)
+ and converts to interface key using Topology methods.
+
+ :param node: Node in topology.
+ :param interface: Name, sw_if_index, link name or key of an interface
+ on the node.
+ :type node: dict
+ :type interface: str or int
+
+ :returns: Interface key.
+ :rtype: str
+
+ :raises TypeError: If provided with invalid interface argument.
+ :raises RuntimeError: If the interface does not exist in topology.
+ """
+
+ if isinstance(interface, int):
+ key = Topology.get_interface_by_sw_index(node, interface)
+ if key is None:
+ raise RuntimeError("Interface with sw_if_index={0} does not "
+ "exist in topology.".format(interface))
+ elif interface in Topology.get_node_interfaces(node):
+ key = interface
+ elif interface in Topology.get_links({"dut": node}):
+ key = Topology.get_interface_by_link_name(node, interface)
+ elif isinstance(interface, basestring):
+ key = Topology.get_interface_by_name(node, interface)
+ if key is None:
+ raise RuntimeError("Interface with key, name or link name "
+ "\"{0}\" does not exist in topology."
+ .format(interface))
+ else:
+ raise TypeError("Type of interface argument must be integer"
+ " or string.")
+ return key
+
+ @staticmethod
+ def convert_interface_reference(node, interface, wanted_format):
+ """Takes interface reference in any format
+ (name, link name, topology key or sw_if_index) and returns
+ its equivalent in the desired format.
+
+ :param node: Node in topology.
+ :param interface: Name, sw_if_index, link name or key of an interface
+ on the node.
+ :param wanted_format: Format of return value wanted.
+ Valid options are: sw_if_index, key, name.
+ :type node: dict
+ :type interface: str or int
+ :type wanted_format: str
+ :returns: Interface name, interface key or sw_if_index.
+ :rtype: str or int
+ :raises TypeError, ValueError: If provided with invalid arguments.
+ :raises RuntimeError: If the interface does not exist in topology.
+ """
+
+ key = Topology.convert_interface_reference_to_key(node, interface)
+
+ conversions = {
+ "key": lambda x, y: y,
+ "name": Topology.get_interface_name,
+ "sw_if_index": Topology.get_interface_sw_index
+ }
+
+ try:
+ return conversions[wanted_format](node, key)
+ except KeyError:
+ raise ValueError("Unrecognized return value wanted: {0}."
+ "Valid options are key, name, sw_if_index"
+ .format(wanted_format))
+
@staticmethod
def get_interface_numa_node(node, iface_key):
"""Get interface numa node.
:param iface_key: Interface key from topology file.
:type node: dict
:type iface_key: str
- :return: numa node id, None if not available.
+ :returns: numa node id, None if not available.
:rtype: int
"""
try:
:param iface_keys: Interface keys for lookup.
:type node: dict
:type iface_keys: strings
+ :returns: Numa node of most given interfaces or 0.
+ :rtype: int
"""
numa_list = []
for if_key in iface_keys:
numa_cnt_mc = Counter(numa_list).most_common()
- if len(numa_cnt_mc) > 0 and numa_cnt_mc[0][0] != -1:
+ if numa_cnt_mc and numa_cnt_mc[0][0] != -1:
return numa_cnt_mc[0][0]
- elif len(numa_cnt_mc) > 1 and numa_cnt_mc[0][0] == -1:
+ if len(numa_cnt_mc) > 1 and numa_cnt_mc[0][0] == -1:
return numa_cnt_mc[1][0]
- else:
- return 0
+ return 0
@staticmethod
def get_interface_mac(node, iface_key):
:param iface_key: Interface key from topology file.
:type node: dict
:type iface_key: str
- :return: Return MAC or None if not found.
+ :returns: Return MAC or None if not found.
"""
try:
return node['interfaces'][iface_key].get('mac_address')
except KeyError:
return None
+ @staticmethod
+ def get_interface_ip4(node, iface_key):
+ """Get IP4 address for the interface.
+
+ :param node: Node to get interface mac on.
+ :param iface_key: Interface key from topology file.
+ :type node: dict
+ :type iface_key: str
+ :returns: Return IP4 or None if not found.
+ """
+ try:
+ return node['interfaces'][iface_key].get('ip4_address', None)
+ except KeyError:
+ return None
+
@staticmethod
def get_adjacent_node_and_interface(nodes_info, node, iface_key):
"""Get node and interface adjacent to specified interface
on local network.
:param nodes_info: Dictionary containing information on all nodes
- in topology.
+ in topology.
:param node: Node that contains specified interface.
:param iface_key: Interface key from topology file.
:type nodes_info: dict
:type node: dict
:type iface_key: str
- :return: Return (node, interface_key) tuple or None if not found.
+ :returns: Return (node, interface_key) tuple or None if not found.
:rtype: (dict, str)
"""
link_name = None
continue
if if_val['link'] == link_name:
return node_data, if_key
+ return None
@staticmethod
def get_interface_pci_addr(node, iface_key):
:param iface_key: Interface key from topology file.
:type node: dict
:type iface_key: str
- :return: Return PCI address or None if not found.
+ :returns: Return PCI address or None if not found.
"""
try:
return node['interfaces'][iface_key].get('pci_address')
:param iface_key: Interface key from topology file.
:type node: dict
:type iface_key: str
- :return: Return interface driver or None if not found.
+ :returns: Return interface driver or None if not found.
"""
try:
return node['interfaces'][iface_key].get('driver')
:param node: Node to get list of interfaces from.
:type node: dict
- :return: Return list of keys of all interfaces.
+ :returns: Return list of keys of all interfaces.
:rtype: list
"""
return node['interfaces'].keys()
:param link_name: Link name.
:type node: dict
:type link_name: str
- :return: MAC address string.
+ :returns: MAC address string.
:rtype: str
"""
for port in node['interfaces'].values():
:param filter_list: Link filter criteria.
:type node: dict
:type filter_list: list of strings
- :return: List of strings that represent link names occupied by the node.
- :rtype: list
+ :returns: List of link names occupied by the node.
+ :rtype: None or list of string
"""
interfaces = node['interfaces']
link_names = []
if filt == interface['model']:
link_names.append(interface['link'])
elif (filter_list is not None) and ('model' not in interface):
- logger.trace("Cannot apply filter on interface: {}"
+ logger.trace('Cannot apply filter on interface: {}'
.format(str(interface)))
else:
link_names.append(interface['link'])
- if len(link_names) == 0:
+ if not link_names:
link_names = None
return link_names
:type node2: dict
:type filter_list_node1: list of strings
:type filter_list_node2: list of strings
- :return: List of strings that represent connecting link names.
+ :returns: List of strings that represent connecting link names.
:rtype: list
"""
:param node2: Connected node.
:type node1: dict
:type node2: dict
- :return: Name of link connecting the two nodes together.
+ :returns: Name of a link connecting the two nodes together.
:rtype: str
- :raises: RuntimeError
+ :raises RuntimeError: If no links are found.
"""
connecting_links = self.get_active_connecting_links(node1, node2)
- if len(connecting_links) == 0:
+ if not connecting_links:
raise RuntimeError("No links connecting the nodes were found")
- else:
- return connecting_links[0]
+ return connecting_links[0]
@keyword('Get egress interfaces name on "${node1}" for link with '
'"${node2}"')
:param node2: Second node.
:type node1: dict
:type node2: dict
- :return: Egress interfaces.
+ :returns: Egress interfaces.
:rtype: list
"""
interfaces = []
links = self.get_active_connecting_links(node1, node2)
- if len(links) == 0:
+ if not links:
raise RuntimeError('No link between nodes')
for interface in node1['interfaces'].values():
link = interface.get('link')
:param node2: Second node.
:type node1: dict
:type node2: dict
- :return: Egress interface name.
+ :returns: Egress interface name.
:rtype: str
"""
interfaces = self.get_egress_interfaces_name_for_nodes(node1, node2)
:type tgen: dict
:type dut1: dict
:type dut2: dict
- :return: Dictionary of possible link combinations.
+ :returns: Dictionary of possible link combinations.
:rtype: dict
"""
# TODO: replace with generic function.
:param node: Node to examine.
:type node: dict
- :return: True if node is type of TG, otherwise False.
+ :returns: True if node is type of TG, otherwise False.
:rtype: bool
"""
return node['type'] == NodeType.TG
:param node: Node created from topology.
:type node: dict
- :return: Hostname or IP address.
+ :returns: Hostname or IP address.
:rtype: str
"""
return node['host']
+ @staticmethod
+ def get_node_arch(node):
+ """Return arch of the node.
+ Default to x86_64 if no arch present
+
+ :param node: Node created from topology.
+ :type node: dict
+ :returns: Node architecture
+ :rtype: str
+ """
+ try:
+ return node['arch']
+ except KeyError:
+ node['arch'] = 'x86_64'
+ return 'x86_64'
+
+ @staticmethod
+ def get_cryptodev(node):
+ """Return Crytodev configuration of the node.
+
+ :param node: Node created from topology.
+ :type node: dict
+ :returns: Cryptodev configuration string.
+ :rtype: str
+ """
+ try:
+ return node['cryptodev']
+ except KeyError:
+ return None
+
+ @staticmethod
+ def get_uio_driver(node):
+ """Return uio-driver configuration of the node.
+
+ :param node: Node created from topology.
+ :type node: dict
+ :returns: uio-driver configuration string.
+ :rtype: str
+ """
+ try:
+ return node['uio_driver']
+ except KeyError:
+ return None
+
@staticmethod
def set_interface_numa_node(node, iface_key, numa_node_id):
"""Set interface numa_node location.
:param iface_key: Interface key from topology file.
:type node: dict
:type iface_key: str
- :return: Return iface_key or None if not found.
+ :returns: Return iface_key or None if not found.
"""
try:
node['interfaces'][iface_key]['numa_node'] = numa_node_id
Code Review / csit.git / blobdiff