def load_topo_from_yaml():
- """Loads topology from file defined in "${TOPOLOGY_PATH}" variable
+ """Load topology from file defined in "${TOPOLOGY_PATH}" variable
:return: nodes from loaded topology
"""
# Traffic Generator (this node has traffic generator on it)
TG = 'TG'
+
class NodeSubTypeTG(object):
#T-Rex traffic generator
TREX = 'TREX'
# Moongen
MOONGEN = 'MOONGEN'
- #IxNetwork
+ # IxNetwork
IXNET = 'IXNET'
DICT__nodes = load_topo_from_yaml()
the used topology.
"""
- def __init__(self):
- pass
-
@staticmethod
def get_node_by_hostname(nodes, hostname):
"""Get node from nodes of the topology by hostname.
@staticmethod
def _get_interface_by_key_value(node, key, value):
- """ Return node interface name according to key and value
+ """Return node interface name according to key and value
:param node: :param node: the node dictionary
:param key: key by which to select the interface.
return list_mac
def _extract_vpp_interface_by_mac(self, interfaces_list, mac_address):
- """Returns interface dictionary from interface_list by mac address.
+ """Return interface dictionary from interface_list by mac address.
Extracts interface dictionary from all of the interfaces in interfaces
list parsed from json according to mac_address of the interface
interface_dict = {}
list_mac_address = self.convert_mac_to_number_list(mac_address)
- logger.trace(list_mac_address.__str__())
+ logger.trace(str(list_mac_address))
for interface in interfaces_list:
# TODO: create vat json integrity checking and move there
if "l2_address" not in interface:
return interface_name
def _update_node_interface_data_from_json(self, node, interface_dump_json):
- """ Update node vpp data in node__DICT from json interface dump.
+ """Update node vpp data in node__DICT from json interface dump.
This method updates vpp interface names and sw indexexs according to
interface mac addresses found in interface_dump_json
if_mac = ifc['mac_address']
interface_dict = self._extract_vpp_interface_by_mac(interface_list,
if_mac)
+ if not interface_dict:
+ raise Exception('Interface {0} not found by MAC {1}'.
+ format(ifc, if_mac))
ifc['name'] = interface_dict["interface_name"]
ifc['vpp_sw_index'] = interface_dict["sw_if_index"]
InterfaceSetup.tg_set_interfaces_udev_rules(node)
def update_all_interface_data_on_all_nodes(self, nodes):
- """ Update interface names on all nodes in DICT__nodes
+ """Update interface names on all nodes in DICT__nodes
:param nodes: Nodes in the topology.
:type nodes: dict
This method updates the topology dictionary by querying interface lists
of all nodes mentioned in the topology dictionary.
It does this by dumping interface list to json output from all devices
- using vpe_api_test, and pairing known information from topology
+ using vpp_api_test, and pairing known information from topology
(mac address/pci address of interface) to state from VPP.
For TG/linux nodes add interface name only.
"""
:rtype: (dict, dict)
"""
link_name = None
- # get link name where the interface belongs to
+ # get link name where the interface belongs to
for port_name, port_data in node['interfaces'].iteritems():
if port_name == 'mgmt':
continue
@staticmethod
def _get_node_active_link_names(node):
- """Returns list of link names that are other than mgmt links
+ """Return list of link names that are other than mgmt links
:param node: node topology dictionary
:return: list of strings that represent link names occupied by the node
@keyword('Get active links connecting "${node1}" and "${node2}"')
def get_active_connecting_links(self, node1, node2):
- """Returns list of link names that connect together node1 and node2
+ """Return list of link names that connect together node1 and node2
:param node1: node topology dictionary
:param node2: node topology dictionary
:param node2: Second node.
:type node1: dict
:type node2: dict
- :return: Engress interfaces.
+ :return: Egress interfaces.
:rtype: list
"""
interfaces = []
:param node2: Second node.
:type node1: dict
:type node2: dict
- :return: Engress interface.
+ :return: Egress interface.
:rtype: str
"""
interfaces = self.get_egress_interfaces_for_nodes(node1, node2)
if not interfaces:
- raise RuntimeError('No engress interface for nodes')
+ raise RuntimeError('No egress interface for nodes')
return interfaces[0]
@keyword('Get link data useful in circular topology test from tg "${tgen}"'
' dut1 "${dut1}" dut2 "${dut2}"')
def get_links_dict_from_nodes(self, tgen, dut1, dut2):
- """Returns link combinations used in tests in circular topology.
+ """Return link combinations used in tests in circular topology.
For the time being it returns links from the Node path:
TG->DUT1->DUT2->TG
- :param tg: traffic generator node data
+ :param tgen: traffic generator node data
:param dut1: DUT1 node data
:param dut2: DUT2 node data
- :type tg: dict
+ :type tgen: dict
:type dut1: dict
:type dut2: dict
:return: dictionary of possible link combinations
'DUT1_BD_LINKS': dut1_bd_links,
'DUT2_BD_LINKS': dut2_bd_links}
return topology_links
+
+ @staticmethod
+ def is_tg_node(node):
+ """Find out whether the node is TG
+
+ :param node: node to examine
+ :return: True if node is type of TG; False otherwise
+ """
+ return node['type'] == NodeType.TG
+
+ @staticmethod
+ def get_node_hostname(node):
+ """Return host (hostname/ip address) of the node.
+
+ :param node: Node created from topology.
+ :type node: dict
+ :return: host as 'str' type
+ """
+ return node['host']
Code Review / csit.git / blobdiff