-# Copyright (c) 2019 Cisco and/or its affiliates.
+# Copyright (c) 2021 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:
"""Memif interface library."""
-import logging
from enum import IntEnum
+from robot.api import logger
+
from resources.libraries.python.topology import NodeType, Topology
-from resources.libraries.python.PapiExecutor import PapiExecutor
-from resources.libraries.python.L2Util import L2Util
+from resources.libraries.python.PapiExecutor import PapiSocketExecutor
class MemifRole(IntEnum):
SLAVE = 1
-class Memif(object):
+class Memif:
"""Memif interface class"""
def __init__(self):
pass
@staticmethod
- def _memif_dump(node):
- """Get the memif dump on the given node.
+ def _memif_details(node):
+ """Get the memif dump details on the given node.
:param node: Given node to get Memif dump from.
:type node: dict
:returns: List of memif interfaces extracted from Papi response.
:rtype: list
"""
- with PapiExecutor(node) as papi_exec:
- dump = papi_exec.add("memif_dump").get_dump()
+ cmd = u"memif_dump"
+ with PapiSocketExecutor(node) as papi_exec:
+ details = papi_exec.add(cmd).get_details()
- data = list()
- for item in dump.reply[0]["api_reply"]:
- item["memif_details"]["if_name"] = \
- item["memif_details"]["if_name"].rstrip('\x00')
- item["memif_details"]["hw_addr"] = \
- L2Util.bin_to_mac(item["memif_details"]["hw_addr"])
- data.append(item)
+ for memif in details:
+ memif[u"hw_addr"] = str(memif[u"hw_addr"])
+ memif[u"role"] = memif[u"role"].value
+ memif[u"mode"] = memif[u"mode"].value
+ memif[u"flags"] = memif[u"flags"].value \
+ if hasattr(memif[u"flags"], u"value") else int(memif[u"flags"])
- logging.debug("MEMIF data:\n{data}".format(data=data))
+ logger.debug(f"MEMIF details:\n{details}")
- return data
+ return details
@staticmethod
def _memif_socket_filename_add_del(node, is_add, filename, sid):
includes only retval.
:rtype: dict
"""
- cmd = 'memif_socket_filename_add_del'
- err_msg = 'Failed to create memif socket on host {host}'.format(
- host=node['host'])
+ cmd = u"memif_socket_filename_add_del"
+ err_msg = f"Failed to create memif socket on host {node[u'host']}"
args = dict(
- is_add=int(is_add),
+ is_add=is_add,
socket_id=int(sid),
- socket_filename=str('/tmp/' + filename)
+ socket_filename=str(u"/tmp/" + filename)
)
- with PapiExecutor(node) as papi_exec:
- data = papi_exec.add(cmd, **args).get_replies(err_msg).\
- verify_reply(err_msg=err_msg)
- return data
+ with PapiSocketExecutor(node) as papi_exec:
+ return papi_exec.add(cmd, **args).get_reply(err_msg)
@staticmethod
def _memif_create(node, mid, sid, rxq=1, txq=1, role=1):
- """Create Memif interface on the given node.
+ """Create Memif interface on the given node, return its sw_if_index.
:param node: Given node to create Memif interface on.
:param mid: Memif interface ID.
:type rxq: int
:type txq: int
:type role: int
- :returns: Verified data from PAPI response.
- :rtype: dict
+ :returns: sw_if_index
+ :rtype: int
"""
- cmd = 'memif_create'
- err_msg = 'Failed to create memif interface on host {host}'.format(
- host=node['host'])
+ cmd = u"memif_create"
+ err_msg = f"Failed to create memif interface on host {node[u'host']}"
args = dict(
role=role,
rx_queues=int(rxq),
tx_queues=int(txq),
socket_id=int(sid),
- id=int(mid)
+ id=int(mid),
+ secret=u""
)
- with PapiExecutor(node) as papi_exec:
- data = papi_exec.add(cmd, **args).get_replies(err_msg).\
- verify_reply(err_msg=err_msg)
- return data
+
+ with PapiSocketExecutor(node) as papi_exec:
+ return papi_exec.add(cmd, **args).get_sw_if_index(err_msg)
@staticmethod
- def create_memif_interface(node, filename, mid, sid, rxq=1, txq=1,
- role="SLAVE"):
+ def create_memif_interface(
+ node, filename, mid, sid, rxq=1, txq=1, role=u"SLAVE"):
"""Create Memif interface on the given node.
:param node: Given node to create Memif interface on.
:rtype: int
:raises ValueError: If command 'create memif' fails.
"""
-
role = getattr(MemifRole, role.upper()).value
# Create socket
Memif._memif_socket_filename_add_del(node, True, filename, sid)
# Create memif
- rsp = Memif._memif_create(node, mid, sid, rxq=rxq, txq=txq, role=role)
+ sw_if_index = Memif._memif_create(
+ node, mid, sid, rxq=rxq, txq=txq, role=role
+ )
# Update Topology
- if_key = Topology.add_new_port(node, 'memif')
- Topology.update_interface_sw_if_index(node, if_key, rsp["sw_if_index"])
+ if_key = Topology.add_new_port(node, u"memif")
+ Topology.update_interface_sw_if_index(node, if_key, sw_if_index)
- ifc_name = Memif.vpp_get_memif_interface_name(node, rsp["sw_if_index"])
+ ifc_name = Memif.vpp_get_memif_interface_name(node, sw_if_index)
Topology.update_interface_name(node, if_key, ifc_name)
- ifc_mac = Memif.vpp_get_memif_interface_mac(node, rsp["sw_if_index"])
+ ifc_mac = Memif.vpp_get_memif_interface_mac(node, sw_if_index)
Topology.update_interface_mac_address(node, if_key, ifc_mac)
- Topology.update_interface_memif_socket(node, if_key, '/tmp/' + filename)
+ Topology.update_interface_memif_socket(
+ node, if_key, u"/tmp/" + filename
+ )
Topology.update_interface_memif_id(node, if_key, mid)
Topology.update_interface_memif_role(node, if_key, str(role))
- return rsp["sw_if_index"]
+ return sw_if_index
@staticmethod
def show_memif(node):
:param node: Given node to show Memif data on.
:type node: dict
"""
-
- Memif._memif_dump(node)
+ Memif._memif_details(node)
@staticmethod
def show_memif_on_all_duts(nodes):
:type nodes: dict
"""
for node in nodes.values():
- if node['type'] == NodeType.DUT:
+ if node[u"type"] == NodeType.DUT:
Memif.show_memif(node)
@staticmethod
- def vpp_get_memif_interface_name(node, sw_if_idx):
+ def vpp_get_memif_interface_name(node, sw_if_index):
"""Get Memif interface name from Memif interfaces dump.
:param node: DUT node.
- :param sw_if_idx: DUT node.
+ :param sw_if_index: DUT node.
:type node: dict
- :type sw_if_idx: int
+ :type sw_if_index: int
:returns: Memif interface name, or None if not found.
:rtype: str
"""
+ details = Memif._memif_details(node)
- dump = Memif._memif_dump(node)
-
- for item in dump:
- if item["memif_details"]["sw_if_index"] == sw_if_idx:
- return item["memif_details"]["if_name"]
+ for memif in details:
+ if memif[u"sw_if_index"] == sw_if_index:
+ return memif[u"if_name"]
return None
@staticmethod
- def vpp_get_memif_interface_mac(node, sw_if_idx):
+ def vpp_get_memif_interface_mac(node, sw_if_index):
"""Get Memif interface MAC address from Memif interfaces dump.
:param node: DUT node.
- :param sw_if_idx: DUT node.
+ :param sw_if_index: DUT node.
:type node: dict
- :type sw_if_idx: int
+ :type sw_if_index: int
:returns: Memif interface MAC address, or None if not found.
:rtype: str
"""
+ details = Memif._memif_details(node)
- dump = Memif._memif_dump(node)
-
- for item in dump:
- if item["memif_details"]["sw_if_index"] == sw_if_idx:
- return item["memif_details"]["hw_addr"]
+ for memif in details:
+ if memif[u"sw_if_index"] == sw_if_index:
+ return memif[u"hw_addr"]
return None
Code Review / csit.git / blobdiff