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 enum import IntEnum
  17
  18 from robot.api import logger
  19
  20 from resources.libraries.python.Constants import Constants
  21 from resources.libraries.python.InterfaceUtil import InterfaceUtil
  22 from resources.libraries.python.L2Util import L2Util
  23 from resources.libraries.python.PapiExecutor import PapiSocketExecutor
  24 from resources.libraries.python.topology import Topology
  25
  26
  27 class TapFlags(IntEnum):
  28     """TAP interface flags."""
  29     TAP_FLAG_GSO = 1
  30
  31
  32 class Tap(object):
  33     """Tap utilities."""
  34
  35     @staticmethod
  36     def add_tap_interface(node, tap_name, mac=None):
  37         """Add tap interface with name and optionally with MAC.
  38
  39         :param node: Node to add tap on.
  40         :param tap_name: Tap interface name for linux tap.
  41         :param mac: Optional MAC address for VPP tap.
  42         :type node: dict
  43         :type tap_name: str or unicode
  44         :type mac: str
  45         :returns: Returns a interface index.
  46         :rtype: int
  47         """
  48         if isinstance(tap_name, unicode):
  49             tap_name = str(tap_name)
  50         cmd = 'tap_create_v2'
  51         args = dict(
  52             id=Constants.BITWISE_NON_ZERO,
  53             use_random_mac=False if mac else True,
  54             mac_address=L2Util.mac_to_bin(mac) if mac else None,
  55             host_mtu_set=False,
  56             host_mac_addr_set=False,
  57             host_ip4_prefix_set=False,
  58             host_ip6_prefix_set=False,
  59             host_ip4_gw_set=False,
  60             host_ip6_gw_set=False,
  61             host_namespace_set=False,
  62             host_if_name_set=True,
  63             host_bridge_set=False,
  64             host_if_name=tap_name,
  65         )
  66         err_msg = 'Failed to create tap interface {tap} on host {host}'.format(
  67             tap=tap_name, host=node['host'])
  68
  69         with PapiSocketExecutor(node) as papi_exec:
  70             sw_if_index = papi_exec.add(cmd, **args).get_sw_if_index(err_msg)
  71
  72         if_key = Topology.add_new_port(node, 'tap')
  73         Topology.update_interface_sw_if_index(node, if_key, sw_if_index)
  74         Topology.update_interface_name(node, if_key, tap_name)
  75         if mac is None:
  76             mac = Tap.vpp_get_tap_interface_mac(node, tap_name)
  77         Topology.update_interface_mac_address(node, if_key, mac)
  78         tap_dev_name = Tap.vpp_get_tap_dev_name(node, tap_name)
  79         Topology.update_interface_tap_dev_name(node, if_key, tap_dev_name)
  80
  81         return sw_if_index
  82
  83     @staticmethod
  84     def vpp_get_tap_dev_name(node, host_if_name):
  85         """Get VPP tap interface name from hardware interfaces dump.
  86
  87         :param node: DUT node.
  88         :param host_if_name: Tap host interface name.
  89         :type node: dict
  90         :type host_if_name: str
  91         :returns: VPP tap interface dev_name.
  92         :rtype: str
  93         """
  94         return Tap.tap_dump(node, host_if_name).get('dev_name')
  95
  96     @staticmethod
  97     def vpp_get_tap_interface_mac(node, interface_name):
  98         """Get tap interface MAC address from interfaces dump.
  99
 100         :param node: DUT node.
 101         :param interface_name: Tap interface name.
 102         :type node: dict
 103         :type interface_name: str
 104         :returns: Tap interface MAC address.
 105         :rtype: str
 106         """
 107         return InterfaceUtil.vpp_get_interface_mac(node, interface_name)
 108
 109     @staticmethod
 110     def tap_dump(node, name=None):
 111         """Get all TAP interface data from the given node, or data about
 112         a specific TAP interface.
 113
 114         :param node: VPP node to get data from.
 115         :param name: Optional name of a specific TAP interface.
 116         :type node: dict
 117         :type name: str
 118         :returns: Dictionary of information about a specific TAP interface, or
 119             a List of dictionaries containing all TAP data for the given node.
 120         :rtype: dict or list
 121         """
 122         def process_tap_dump(tap_dump):
 123             """Process tap dump.
 124
 125             :param tap_dump: Tap interface dump.
 126             :type tap_dump: dict
 127             :returns: Processed tap interface dump.
 128             :rtype: dict
 129             """
 130             tap_dump['host_mac_addr'] = str(tap_dump['host_mac_addr'])
 131             tap_dump['host_ip4_prefix'] = str(tap_dump['host_ip4_prefix'])
 132             tap_dump['host_ip6_prefix'] = str(tap_dump['host_ip6_prefix'])
 133             tap_dump['tap_flags'] = tap_dump['tap_flags'].value \
 134                 if hasattr(tap_dump['tap_flags'], 'value') \
 135                 else int(tap_dump['tap_flags'])
 136             tap_dump['host_namespace'] = None \
 137                 if tap_dump['host_namespace'] == '(nil)' \
 138                 else tap_dump['host_namespace']
 139             tap_dump['host_bridge'] = None \
 140                 if tap_dump['host_bridge'] == '(nil)' \
 141                 else tap_dump['host_bridge']
 142
 143             return tap_dump
 144
 145         cmd = 'sw_interface_tap_v2_dump'
 146         err_msg = 'Failed to get TAP dump on host {host}'.format(
 147             host=node['host'])
 148         with PapiSocketExecutor(node) as papi_exec:
 149             details = papi_exec.add(cmd).get_details(err_msg)
 150
 151         data = list() if name is None else dict()
 152         for dump in details:
 153             if name is None:
 154                 data.append(process_tap_dump(dump))
 155             elif dump.get('host_if_name') == name:
 156                 data = process_tap_dump(dump)
 157                 break
 158
 159         logger.debug('TAP data:\n{tap_data}'.format(tap_data=data))
 160         return data