-# Copyright (c) 2020 Cisco and/or its affiliates.
+# Copyright (c) 2024 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:
PAPI = u"PAPI"
# VPP PAPI Stats (legacy option until stats are migrated to Socket PAPI)
STATS = u"STATS"
+ # VPP Socket CLI
+ CLI = u"CLI"
DICT__nodes = load_topo_from_yaml()
port_types = (
u"subinterface", u"vlan_subif", u"memif", u"tap", u"vhost",
u"loopback", u"gre_tunnel", u"vxlan_tunnel", u"eth_bond",
- u"eth_avf", u"eth_rdma"
+ u"eth_avf", u"eth_rdma", u"geneve_tunnel", u"eth_af_xdp",
+ u"gtpu_tunnel"
)
for node_data in nodes.values():
return links
@staticmethod
- def _get_interface_by_key_value(node, key, value):
+ def _get_interface_by_key_value(node, key, value, subsequent=False):
"""Return node interface key from topology file
according to key and value.
:param node: The node dictionary.
:param key: Key by which to select the interface.
:param value: Value that should be found using the key.
+ :param subsequent: Use second interface of the link. Useful for
+ back-to-back links. Default: False
:type node: dict
:type key: string
:type value: string
+ :type subsequent: bool
:returns: Interface key from topology file
:rtype: string
"""
k_val = if_val.get(key)
if k_val is not None:
if k_val == value:
- retval = if_key
- break
+ if subsequent:
+ subsequent = False
+ else:
+ retval = if_key
+ break
return retval
@staticmethod
return Topology._get_interface_by_key_value(node, u"name", iface_name)
@staticmethod
- def get_interface_by_link_name(node, link_name):
+ def get_interface_by_link_name(node, link_name, subsequent=False):
"""Return interface key of link on node.
This method returns the interface name associated with a given link
:param node: The node topology dictionary.
:param link_name: Name of the link that a interface is connected to.
+ :param subsequent: Use second interface of the link. Useful for
+ back-to-back links. Default: False
:type node: dict
:type link_name: string
+ :type subsequent: bool
:returns: Interface key of the interface connected to the given link.
:rtype: str
"""
- return Topology._get_interface_by_key_value(node, u"link", link_name)
+ return Topology._get_interface_by_key_value(
+ node, u"link", link_name, subsequent=subsequent
+ )
def get_interfaces_by_link_names(self, node, link_names):
"""Return dictionary of dictionaries {"interfaceN", interface name}.
# find link
for node_data in nodes_info.values():
# skip self
- if node_data[u"host"] == node[u"host"]:
+ l_hash = node_data[u"host"]+str(node_data[u"port"])
+ r_hash = node[u"host"]+str(node[u"port"])
+ if l_hash == r_hash:
continue
for if_key, if_val \
in node_data[u"interfaces"].items():
return None
@staticmethod
- def _get_node_active_link_names(node, filter_list=None):
+ def _get_node_active_link_names(node, filter_list=None, topo_has_dut=True):
"""Return list of link names that are other than mgmt links.
:param node: Node topology dictionary.
:param filter_list: Link filter criteria.
+ :param topo_has_dut: Whether we require back-to-back links.
:type node: dict
:type filter_list: list of strings
+ :type topo_has_dut: bool
:returns: List of link names occupied by the node.
:rtype: None or list of string
"""
link_names.append(interface[u"link"])
if not link_names:
link_names = None
+ if not topo_has_dut:
+ new_link_names = list()
+ for link_name in link_names:
+ count = 0
+ for interface in interfaces.values():
+ link = interface.get(u"link", None)
+ if link == link_name:
+ count += 1
+ if count == 2:
+ new_link_names.append(link_name)
+ link_names = new_link_names
return link_names
def get_active_connecting_links(
:rtype: list
"""
- node1_links = self._get_node_active_link_names(
- node1, filter_list=filter_list_node1
- )
- node2_links = self._get_node_active_link_names(
- node2, filter_list=filter_list_node2
- )
+ if node1 != node2:
+ node1_links = self._get_node_active_link_names(
+ node1, filter_list=filter_list_node1
+ )
+ node2_links = self._get_node_active_link_names(
+ node2, filter_list=filter_list_node2
+ )
+ else:
+ # Looking for back-to-back links.
+ node1_links = self._get_node_active_link_names(
+ node1, filter_list=filter_list_node1, topo_has_dut=False
+ )
+ node2_links = node1_links
connecting_links = None
if node1_links is None:
except KeyError:
return None
+ def get_bus(node):
+ """Return bus configuration of the node.
+
+ :param node: Node created from topology.
+ :type node: dict
+ :returns: bus configuration string.
+ :rtype: str
+ """
+ try:
+ return node[u"bus"]
+ except KeyError:
+ return None
+
@staticmethod
def get_uio_driver(node):
"""Return uio-driver configuration of the node.
"""
for node in nodes.values():
if u"sockets" in list(node.keys()):
+ # Containers are disconnected and destroyed already.
node.pop(u"sockets")
Code Review / csit.git / blobdiff