1 # Copyright (c) 2019 Cisco and/or its affiliates.
2 # Licensed under the Apache License, Version 2.0 (the "License");
3 # you may not use this file except in compliance with the License.
4 # You may obtain a copy of the License at:
6 # http://www.apache.org/licenses/LICENSE-2.0
8 # Unless required by applicable law or agreed to in writing, software
9 # distributed under the License is distributed on an "AS IS" BASIS,
10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11 # See the License for the specific language governing permissions and
12 # limitations under the License.
14 """Tap utilities library."""
16 from robot.api import logger
18 from resources.libraries.python.Constants import Constants
19 from resources.libraries.python.InterfaceUtil import InterfaceUtil
20 from resources.libraries.python.L2Util import L2Util
21 from resources.libraries.python.PapiExecutor import PapiSocketExecutor
22 from resources.libraries.python.topology import Topology
29 def add_tap_interface(node, tap_name, mac=None, num_rx_queues=1):
30 """Add tap interface with name and optionally with MAC.
32 :param node: Node to add tap on.
33 :param tap_name: Tap interface name for linux tap.
34 :param mac: Optional MAC address for VPP tap.
35 :param num_rx_queues: Number of RX queues.
39 :type num_rx_queues: int
40 :returns: Returns a interface index.
43 cmd = u"tap_create_v2"
45 id=Constants.BITWISE_NON_ZERO,
46 use_random_mac=bool(mac is None),
47 mac_address=L2Util.mac_to_bin(mac) if mac else None,
48 num_rx_queues=int(num_rx_queues),
50 host_mac_addr_set=False,
51 host_ip4_prefix_set=False,
52 host_ip6_prefix_set=False,
53 host_ip4_gw_set=False,
54 host_ip6_gw_set=False,
55 host_namespace_set=False,
56 host_if_name_set=True,
57 host_if_name=tap_name,
60 err_msg = f"Failed to create tap interface {tap_name} " \
61 f"on host {node[u'host']}"
63 with PapiSocketExecutor(node) as papi_exec:
64 sw_if_index = papi_exec.add(cmd, **args).get_sw_if_index(err_msg)
66 if_key = Topology.add_new_port(node, u"tap")
67 Topology.update_interface_sw_if_index(node, if_key, sw_if_index)
68 Topology.update_interface_name(node, if_key, tap_name)
70 mac = Tap.vpp_get_tap_interface_mac(node, tap_name)
71 Topology.update_interface_mac_address(node, if_key, mac)
72 tap_dev_name = Tap.vpp_get_tap_dev_name(node, tap_name)
73 Topology.update_interface_tap_dev_name(node, if_key, tap_dev_name)
78 def vpp_get_tap_dev_name(node, host_if_name):
79 """Get VPP tap interface name from hardware interfaces dump.
81 :param node: DUT node.
82 :param host_if_name: Tap host interface name.
84 :type host_if_name: str
85 :returns: VPP tap interface dev_name.
88 return Tap.tap_dump(node, host_if_name).get(u"dev_name")
91 def vpp_get_tap_interface_mac(node, interface_name):
92 """Get tap interface MAC address from interfaces dump.
94 :param node: DUT node.
95 :param interface_name: Tap interface name.
97 :type interface_name: str
98 :returns: Tap interface MAC address.
101 return InterfaceUtil.vpp_get_interface_mac(node, interface_name)
104 def tap_dump(node, name=None):
105 """Get all TAP interface data from the given node, or data about
106 a specific TAP interface.
108 :param node: VPP node to get data from.
109 :param name: Optional name of a specific TAP interface.
112 :returns: Dictionary of information about a specific TAP interface, or
113 a List of dictionaries containing all TAP data for the given node.
116 def process_tap_dump(tap_dump):
119 :param tap_dump: Tap interface dump.
121 :returns: Processed tap interface dump.
124 tap_dump[u"host_mac_addr"] = str(tap_dump[u"host_mac_addr"])
125 tap_dump[u"host_ip4_prefix"] = str(tap_dump[u"host_ip4_prefix"])
126 tap_dump[u"host_ip6_prefix"] = str(tap_dump[u"host_ip6_prefix"])
127 tap_dump[u"tap_flags"] = tap_dump[u"tap_flags"].value \
128 if hasattr(tap_dump[u"tap_flags"], u"value") \
129 else int(tap_dump[u"tap_flags"])
130 tap_dump[u"host_namespace"] = None \
131 if tap_dump[u"host_namespace"] == u"(nil)" \
132 else tap_dump[u"host_namespace"]
133 tap_dump[u"host_bridge"] = None \
134 if tap_dump[u"host_bridge"] == u"(nil)" \
135 else tap_dump[u"host_bridge"]
139 cmd = u"sw_interface_tap_v2_dump"
140 err_msg = f"Failed to get TAP dump on host {node[u'host']}"
142 with PapiSocketExecutor(node) as papi_exec:
143 details = papi_exec.add(cmd).get_details(err_msg)
145 data = list() if name is None else dict()
148 data.append(process_tap_dump(dump))
149 elif dump.get(u"host_if_name") == name:
150 data = process_tap_dump(dump)
153 logger.debug(f"TAP data:\n{data}")