resources/libraries/python/Tap.py

   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:
   5 #
   6 #     http://www.apache.org/licenses/LICENSE-2.0
   7 #
   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.
  13
  14 """Tap utilities library."""
  15
  16 from robot.api import logger
  17
  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
  23
  24
  25 class Tap:
  26     """Tap utilities."""
  27
  28     @staticmethod
  29     def add_tap_interface(node, tap_name, mac=None, num_rx_queues=1):
  30         """Add tap interface with name and optionally with MAC.
  31
  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.
  36         :type node: dict
  37         :type tap_name: str
  38         :type mac: str
  39         :type num_rx_queues: int
  40         :returns: Returns a interface index.
  41         :rtype: int
  42         """
  43         cmd = u"tap_create_v2"
  44         args = dict(
  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),
  49             host_mtu_set=False,
  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,
  58             host_bridge_set=False
  59         )
  60         err_msg = f"Failed to create tap interface {tap_name} " \
  61             f"on host {node[u'host']}"
  62
  63         with PapiSocketExecutor(node) as papi_exec:
  64             sw_if_index = papi_exec.add(cmd, **args).get_sw_if_index(err_msg)
  65
  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)
  69         if mac is None:
  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)
  74
  75         return sw_if_index
  76
  77     @staticmethod
  78     def vpp_get_tap_dev_name(node, host_if_name):
  79         """Get VPP tap interface name from hardware interfaces dump.
  80
  81         :param node: DUT node.
  82         :param host_if_name: Tap host interface name.
  83         :type node: dict
  84         :type host_if_name: str
  85         :returns: VPP tap interface dev_name.
  86         :rtype: str
  87         """
  88         return Tap.tap_dump(node, host_if_name).get(u"dev_name")
  89
  90     @staticmethod
  91     def vpp_get_tap_interface_mac(node, interface_name):
  92         """Get tap interface MAC address from interfaces dump.
  93
  94         :param node: DUT node.
  95         :param interface_name: Tap interface name.
  96         :type node: dict
  97         :type interface_name: str
  98         :returns: Tap interface MAC address.
  99         :rtype: str
 100         """
 101         return InterfaceUtil.vpp_get_interface_mac(node, interface_name)
 102
 103     @staticmethod
 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.
 107
 108         :param node: VPP node to get data from.
 109         :param name: Optional name of a specific TAP interface.
 110         :type node: dict
 111         :type name: str
 112         :returns: Dictionary of information about a specific TAP interface, or
 113             a List of dictionaries containing all TAP data for the given node.
 114         :rtype: dict or list
 115         """
 116         def process_tap_dump(tap_dump):
 117             """Process tap dump.
 118
 119             :param tap_dump: Tap interface dump.
 120             :type tap_dump: dict
 121             :returns: Processed tap interface dump.
 122             :rtype: dict
 123             """
 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"]
 136
 137             return tap_dump
 138
 139         cmd = u"sw_interface_tap_v2_dump"
 140         err_msg = f"Failed to get TAP dump on host {node[u'host']}"
 141
 142         with PapiSocketExecutor(node) as papi_exec:
 143             details = papi_exec.add(cmd).get_details(err_msg)
 144
 145         data = list() if name is None else dict()
 146         for dump in details:
 147             if name is None:
 148                 data.append(process_tap_dump(dump))
 149             elif dump.get(u"host_if_name") == name:
 150                 data = process_tap_dump(dump)
 151                 break
 152
 153         logger.debug(f"TAP data:\n{data}")
 154         return data