+ :type name: str
+ :returns: Dictionary of information about a specific TAP interface, or
+ a List of dictionaries containing all TAP data for the given node.
+ :rtype: dict or list
+ """
+ def process_tap_dump(tap_dump):
+ """Process tap dump.
+
+ :param tap_dump: Tap interface dump.
+ :type tap_dump: dict
+ :returns: Processed tap interface dump.
+ :rtype: dict
+ """
+ tap_dump[u"host_mac_addr"] = str(tap_dump[u"host_mac_addr"])
+ tap_dump[u"host_ip4_prefix"] = str(tap_dump[u"host_ip4_prefix"])
+ tap_dump[u"host_ip6_prefix"] = str(tap_dump[u"host_ip6_prefix"])
+ tap_dump[u"tap_flags"] = tap_dump[u"tap_flags"].value \
+ if hasattr(tap_dump[u"tap_flags"], u"value") \
+ else int(tap_dump[u"tap_flags"])
+ tap_dump[u"host_namespace"] = None \
+ if tap_dump[u"host_namespace"] == u"(nil)" \
+ else tap_dump[u"host_namespace"]
+ tap_dump[u"host_bridge"] = None \
+ if tap_dump[u"host_bridge"] == u"(nil)" \
+ else tap_dump[u"host_bridge"]
+
+ return tap_dump
+
+ cmd = u"sw_interface_tap_v2_dump"
+ err_msg = f"Failed to get TAP dump on host {node[u'host']}"
+
+ with PapiSocketExecutor(node) as papi_exec:
+ details = papi_exec.add(cmd).get_details(err_msg)
+
+ data = list() if name is None else dict()
+ for dump in details:
+ if name is None:
+ data.append(process_tap_dump(dump))
+ elif dump.get(u"host_if_name") == name:
+ data = process_tap_dump(dump)
+ break
+
+ logger.debug(f"TAP data:\n{data}")
+ return data
+
+
+class TapFeatureMask:
+ """Tap features utilities"""
+
+ @staticmethod
+ def create_tap_feature_mask(**kwargs):
+ """Create tap feature mask with feature bits set according to kwargs.
+ :param kwargs: Key-value pairs of feature names and it's state
+ :type kwargs: dict
+ """
+ tap_feature_mask = 0
+
+ if u"all" in kwargs and kwargs[u"all"] is True:
+ for tap_feature_flag in TapFeaturesFlags:
+ tap_feature_mask |= 1 << (tap_feature_flag.value - 1)
+ else:
+ for feature_name, enabled in kwargs.items():
+ tap_feature_name = u"TAP_API_FLAG_" + feature_name.upper()
+ if tap_feature_name not in TapFeaturesFlags.__members__:
+ raise ValueError(u"Unsupported tap feature flag name")
+ if enabled:
+ tap_feature_mask |= \
+ 1 << (TapFeaturesFlags[tap_feature_name].value - 1)
+
+ return tap_feature_mask
+
+ @staticmethod
+ def is_feature_enabled(tap_feature_mask, tap_feature_flag):
+ """Checks if concrete tap feature is enabled within
+ tap_feature_mask
+ :param tap_feature_mask: Mask of enabled tap features
+ :param tap_feature_flag: Checked tap feature
+ :type tap_feature_mask: int
+ :type tap_feature_flag: TapFeaturesFlags