from robot.api import logger
from resources.libraries.python.PapiExecutor import PapiExecutor
-from resources.libraries.python.PapiErrors import PapiError
from resources.libraries.python.topology import Topology
from resources.libraries.python.VatExecutor import VatExecutor
from resources.libraries.python.VatJsonUtil import VatJsonUtil
:type node: dict
:type protocol: IPsecProto
:type index: int
+ :raises RuntimeError: If failed to select IPsec backend or if no API
+ reply received.
"""
- # TODO: move composition of api data to separate method
- api_data = list()
- api = dict(api_name='ipsec_select_backend')
- api_args = dict(protocol=protocol)
- api_args['index'] = index
- api['api_args'] = api_args
- api_data.append(api)
-
- api_reply = None
- with PapiExecutor(node) as papi_executor:
- papi_executor.execute_papi(api_data)
- try:
- papi_executor.papi_should_have_passed()
- except AssertionError:
- raise PapiError('Failed to select IPsec backend on host {host}'.
- format(host=node['host']))
- api_reply = papi_executor.get_papi_reply()
-
- if api_reply is not None:
- api_r = api_reply[0]['api_reply']['ipsec_select_backend_reply']
- if api_r['retval'] == 0:
- logger.trace('IPsec backend successfully selected on host '
- '{host}'.format(host=node['host']))
- else:
- raise PapiError('Failed to select IPsec backend on host {host}'.
- format(host=node['host']))
- else:
- raise PapiError('No reply received for ipsec_select_backend API '
- 'command on host {host}'.format(host=node['host']))
+
+ cmd = 'ipsec_select_backend'
+ cmd_reply = 'ipsec_select_backend_reply'
+ err_msg = 'Failed to select IPsec backend on host {host}'.format(
+ host=node['host'])
+ args = dict(protocol=protocol, index=index)
+ with PapiExecutor(node) as papi_exec:
+ papi_resp = papi_exec.add(cmd, **args).execute_should_pass(err_msg)
+ data = papi_resp.reply[0]['api_reply'][cmd_reply]
+ if data['retval'] != 0:
+ raise RuntimeError('Failed to select IPsec backend on host {host}'.
+ format(host=node['host']))
@staticmethod
def vpp_ipsec_backend_dump(node):
:param node: VPP node to dump IPsec backend on.
:type node: dict
"""
- # TODO: move composition of api data to separate method
- api_data = list()
- api = dict(api_name='ipsec_backend_dump')
- api_args = dict()
- api['api_args'] = api_args
- api_data.append(api)
-
- api_reply = None
- with PapiExecutor(node) as papi_executor:
- papi_executor.execute_papi(api_data)
- try:
- papi_executor.papi_should_have_passed()
- except AssertionError:
- raise PapiError('Failed to dump IPsec backends on host {host}'.
- format(host=node['host']))
- # After API change there is returned VPP internal enum object
- # representing VPP IPSEC protocol instead of integer representation
- # so JSON fails to decode it - we need to check if it is Python API
- # bug or we need to adapt vpp_papi_provider to correctly encode
- # such object into JSON
- # api_reply = papi_executor.get_papi_reply()
- api_reply = papi_executor.get_papi_stdout()
-
- if api_reply is not None:
- logger.trace('IPsec backend dump\n{dump}'.format(dump=api_reply))
- else:
- raise PapiError('No reply received for ipsec_select_backend API '
- 'command on host {host}'.format(host=node['host']))
+
+ err_msg = 'Failed to dump IPsec backends on host {host}'.format(
+ host=node['host'])
+ with PapiExecutor(node) as papi_exec:
+ papi_resp = papi_exec.add('ipsec_backend_dump').execute_should_pass(
+ err_msg, process_reply=False)
+ # After API change there is returned VPP internal enum object
+ # representing VPP IPSEC protocol instead of integer representation
+ # so JSON fails to decode it - we need to check if it is Python API
+ # bug or we need to adapt vpp_papi_provider to correctly encode
+ # such object into JSON
+ # logger.trace('IPsec backend dump\n{dump}'.
+ # format(dump=papi_resp.reply))
+ logger.trace('IPsec backend dump\n{dump}'.format(dump=papi_resp.stdout))
@staticmethod
def vpp_ipsec_add_sad_entry(node, sad_id, spi, crypto_alg, crypto_key,
from resources.libraries.python.CpuUtils import CpuUtils
from resources.libraries.python.DUTSetup import DUTSetup
from resources.libraries.python.PapiExecutor import PapiExecutor
-from resources.libraries.python.PapiErrors import PapiError
from resources.libraries.python.IPUtil import convert_ipv4_netmask_prefix
from resources.libraries.python.IPUtil import IPUtil
from resources.libraries.python.parsers.JsonParser import JsonParser
:type node: dict
:returns: Thread mapping information as a list of dictionaries.
:rtype: list
- :raises RuntimeError: If failed to run command on host.
- :raises PapiError: If no API reply received.
"""
- api_data = list()
- for ifc in node['interfaces'].values():
- if ifc['vpp_sw_index'] is not None:
- api = dict(api_name='sw_interface_rx_placement_dump')
- api_args = dict(sw_if_index=ifc['vpp_sw_index'])
- api['api_args'] = api_args
- api_data.append(api)
-
- with PapiExecutor(node) as papi_executor:
- papi_executor.execute_papi(api_data)
- try:
- papi_executor.papi_should_have_passed()
- api_reply = papi_executor.get_papi_reply()
- except AssertionError:
- raise RuntimeError('Failed to run {api_name} on host '
- '{host}!'.format(host=node['host'], **api))
-
- if api_reply:
- thr_mapping = [s['sw_interface_rx_placement_details'] \
- for r in api_reply for s in r['api_reply']]
- return sorted(thr_mapping, key=lambda k: k['sw_if_index'])
- else:
- raise PapiError('No reply received for {api_name} on host {host}!'.
- format(host=node['host'], **api))
+
+ cmd = 'sw_interface_rx_placement_dump'
+ cmd_reply = 'sw_interface_rx_placement_details'
+ err_msg = "Failed to run '{cmd}' PAPI command on host {host}!".format(
+ cmd=cmd, host=node['host'])
+ with PapiExecutor(node) as papi_exec:
+ for ifc in node['interfaces'].values():
+ papi_exec.add(cmd, sw_if_index=ifc['vpp_sw_index'])
+ papi_resp = papi_exec.execute_should_pass(err_msg)
+ thr_mapping = [s[cmd_reply] for r in papi_resp.reply
+ for s in r['api_reply']]
+ return sorted(thr_mapping, key=lambda k: k['sw_if_index'])
@staticmethod
def vpp_sw_interface_set_rx_placement(node, sw_if_index, queue_id,
:type sw_if_index: int
:type queue_id: int
:type worker_id: int
- :raises RuntimeError: If failed to run command on host.
- :raises PapiError: If no API reply received.
+ :raises RuntimeError: If failed to run command on host or if no API
+ reply received.
"""
- api_data = list()
- api = dict(api_name='sw_interface_set_rx_placement')
- api_args = dict(sw_if_index=sw_if_index, queue_id=queue_id,
- worker_id=worker_id)
- api['api_args'] = api_args
- api_data.append(api)
-
- with PapiExecutor(node) as papi_executor:
- papi_executor.execute_papi(api_data)
- try:
- papi_executor.papi_should_have_passed()
- api_reply = papi_executor.get_papi_reply()
- except AssertionError:
- raise RuntimeError('Failed to run {api_name} on host '
- '{host}!'.format(host=node['host'], **api))
-
- if not api_reply:
- raise PapiError('No reply received for {api_name} on host {host}!'.
- format(host=node['host'], **api))
+
+ cmd = 'sw_interface_set_rx_placement'
+ cmd_reply = 'sw_interface_set_rx_placement_reply'
+ err_msg = "Failed to run '{cmd}' PAPI command on host {host}!".format(
+ host=node['host'], cmd=cmd)
+ args = dict(sw_if_index=sw_if_index, queue_id=queue_id,
+ worker_id=worker_id)
+ with PapiExecutor(node) as papi_exec:
+ papi_resp = papi_exec.add(cmd, **args).execute_should_pass(err_msg)
+ data = papi_resp.reply[0]['api_reply'][cmd_reply]
+ if data['retval'] != 0:
+ raise RuntimeError("Failed to set interface RX placement "
+ "to worker on host {host}".
+ format(host=node['host']))
@staticmethod
def vpp_round_robin_rx_placement(node, prefix):
+++ /dev/null
-# Copyright (c) 2018 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:
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-"""PAPI Errors class file."""
-
-__all__ = ['PapiError', 'PapiInitError', 'PapiJsonFileError',
- 'PapiCommandError', 'PapiCommandInputError']
-
-
-class PapiError(Exception):
- """Python API error."""
- pass
-
-
-class PapiInitError(PapiError):
- """This exception is raised when construction of VPP instance failed."""
- pass
-
-
-class PapiJsonFileError(PapiError):
- """This exception is raised in case of JSON API file error."""
- pass
-
-
-class PapiCommandError(PapiError):
- """This exception is raised when PAPI command(s) execution failed."""
- pass
-
-
-class PapiCommandInputError(PapiCommandError):
- """This exception is raised when incorrect input of Python API is used."""
- pass
-# Copyright (c) 2018 Cisco and/or its affiliates.
+# Copyright (c) 2019 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:
import binascii
import json
-from paramiko.ssh_exception import SSHException
from robot.api import logger
from resources.libraries.python.Constants import Constants
-from resources.libraries.python.PapiErrors import PapiInitError, \
- PapiJsonFileError, PapiCommandError, PapiCommandInputError
-# TODO: from resources.libraries.python.PapiHistory import PapiHistory
from resources.libraries.python.ssh import SSH, SSHTimeout
-__all__ = ['PapiExecutor']
+__all__ = ["PapiExecutor", "PapiResponse"]
+
+# TODO: Implement Papi History
+# from resources.libraries.python.PapiHistory import PapiHistory
+
+
+class PapiResponse(object):
+ """Class for metadata specifying the Papi reply, stdout, stderr and return
+ code.
+ """
+
+ def __init__(self, papi_reply=None, stdout="", stderr="", ret_code=None):
+ """Construct the Papi response by setting the values needed.
+
+ :param papi_reply: API reply from last executed PAPI command(s).
+ :param stdout: stdout from last executed PAPI command(s).
+ :param stderr: stderr from last executed PAPI command(s).
+ :param ret_code: ret_code from last executed PAPI command(s).
+ :type papi_reply: list
+ :type stdout: str
+ :type stderr: str
+ :type ret_code: int
+ """
+
+ # API reply from last executed PAPI command(s)
+ self.reply = papi_reply
+
+ # stdout from last executed PAPI command(s)
+ self.stdout = stdout
+
+ # stderr from last executed PAPI command(s).
+ self.stderr = stderr
+
+ # return code from last executed PAPI command(s)
+ self.ret_code = ret_code
+
+ def __str__(self):
+ """Return string with human readable description of the group.
+
+ :returns: Readable description.
+ :rtype: str
+ """
+ return ("papi_reply={papi_reply} "
+ "stdout={stdout} "
+ "stderr={stderr} "
+ "ret_code={ret_code}".
+ format(papi_reply=self.reply,
+ stdout=self.stdout,
+ stderr=self.stderr,
+ ret_code=self.ret_code))
+
+ def __repr__(self):
+ """Return string executable as Python constructor call.
+
+ :returns: Executable constructor call.
+ :rtype: str
+ """
+ return ("PapiResponse(papi_reply={papi_reply} "
+ "stdout={stdout} "
+ "stderr={stderr} "
+ "ret_code={ret_code})".
+ format(papi_reply=self.reply,
+ stdout=self.stdout,
+ stderr=self.stderr,
+ ret_code=self.ret_code))
class PapiExecutor(object):
- """Contains methods for executing Python API commands on DUTs."""
+ """Contains methods for executing Python API commands on DUTs.
+
+ Use only with "with" statement, e.g.:
+
+ with PapiExecutor(node) as papi_exec:
+ papi_resp = papi_exec.add('show_version').execute_should_pass(err_msg)
+ """
def __init__(self, node):
- self._stdout = None
- self._stderr = None
- self._ret_code = None
+ """Initialization.
+
+ :param node: Node to run command(s) on.
+ :type node: dict
+ """
+
+ # Node to run command(s) on.
self._node = node
- self._json_data = None
- self._api_reply = list()
- self._api_data = None
+
+ # The list of PAPI commands to be executed on the node.
+ self._api_command_list = list()
+
+ # The response on the PAPI commands.
+ self.response = PapiResponse()
self._ssh = SSH()
- try:
- self._ssh.connect(node)
- except:
- raise SSHException('Cannot open SSH connection to host {host} to '
- 'execute PAPI command(s)'.
- format(host=self._node['host']))
def __enter__(self):
+ try:
+ self._ssh.connect(self._node)
+ except IOError:
+ raise RuntimeError("Cannot open SSH connection to host {host} to "
+ "execute PAPI command(s)".
+ format(host=self._node["host"]))
return self
def __exit__(self, exc_type, exc_val, exc_tb):
- pass
+ self._ssh.disconnect(self._node)
+
+ def clear(self):
+ """Empty the internal command list; return self.
+
+ Use when not sure whether previous usage has left something in the list.
+
+ :returns: self, so that method chaining is possible.
+ :rtype: PapiExecutor
+ """
+ self._api_command_list = list()
+ return self
+
+ def add(self, command, **kwargs):
+ """Add next command to internal command list; return self.
+
+ :param command: VPP API command.
+ :param kwargs: Optional key-value arguments.
+ :type command: str
+ :type kwargs: dict
+ :returns: self, so that method chaining is possible.
+ :rtype: PapiExecutor
+ """
+ self._api_command_list.append(dict(api_name=command, api_args=kwargs))
+ return self
+
+ def execute(self, process_reply=True, ignore_errors=False, timeout=120):
+ """Turn internal command list into proper data and execute; return
+ PAPI response.
+
+ This method also clears the internal command list.
+
+ :param process_reply: Process PAPI reply if True.
+ :param ignore_errors: If true, the errors in the reply are ignored.
+ :param timeout: Timeout in seconds.
+ :type process_reply: bool
+ :type ignore_errors: bool
+ :type timeout: int
+ :returns: Papi response including: papi reply, stdout, stderr and
+ return code.
+ :rtype: PapiResponse
+ :raises KeyError: If the reply is not correct.
+ """
+
+ local_list = self._api_command_list
+
+ # Clear first as execution may fail.
+ self.clear()
+
+ ret_code, stdout, stderr = self._execute_papi(local_list, timeout)
+
+ papi_reply = list()
+ if process_reply:
+ json_data = json.loads(stdout)
+ for data in json_data:
+ try:
+ api_reply_processed = dict(
+ api_name=data["api_name"],
+ api_reply=self._process_reply(data["api_reply"]))
+ except KeyError:
+ if ignore_errors:
+ continue
+ else:
+ raise
+ papi_reply.append(api_reply_processed)
+
+ return PapiResponse(papi_reply=papi_reply,
+ stdout=stdout,
+ stderr=stderr,
+ ret_code=ret_code)
+
+ def execute_should_pass(self, err_msg="Failed to execute PAPI command.",
+ process_reply=True, ignore_errors=False,
+ timeout=120):
+ """Execute the PAPI commands and check the return code.
+ Raise exception if the PAPI command(s) failed.
+
+ Note: There are two exceptions raised to distinguish two situations. If
+ not needed, re-implement using only RuntimeError.
+
+ :param err_msg: The message used if the PAPI command(s) execution fails.
+ :param process_reply: Indicate whether or not to process PAPI reply.
+ :param ignore_errors: If true, the errors in the reply are ignored.
+ :param timeout: Timeout in seconds.
+ :type err_msg: str
+ :type process_reply: bool
+ :type ignore_errors: bool
+ :type timeout: int
+ :returns: Papi response including: papi reply, stdout, stderr and
+ return code.
+ :rtype: PapiResponse
+ :raises RuntimeError: If no PAPI command(s) executed.
+ :raises AssertionError: If PAPI command(s) execution passed.
+ """
+
+ response = self.execute(process_reply=process_reply,
+ ignore_errors=ignore_errors,
+ timeout=timeout)
+
+ if response.ret_code != 0:
+ raise AssertionError(err_msg)
+ return response
+
+ def execute_should_fail(self,
+ err_msg="Execution of PAPI command did not fail.",
+ process_reply=False, ignore_errors=False,
+ timeout=120):
+ """Execute the PAPI commands and check the return code.
+ Raise exception if the PAPI command(s) did not fail.
+
+ It does not return anything as we expect it fails.
+
+ Note: There are two exceptions raised to distinguish two situations. If
+ not needed, re-implement using only RuntimeError.
+
+ :param err_msg: The message used if the PAPI command(s) execution fails.
+ :param process_reply: Indicate whether or not to process PAPI reply.
+ :param ignore_errors: If true, the errors in the reply are ignored.
+ :param timeout: Timeout in seconds.
+ :type err_msg: str
+ :type process_reply: bool
+ :type ignore_errors: bool
+ :type timeout: int
+ :raises RuntimeError: If no PAPI command(s) executed.
+ :raises AssertionError: If PAPI command(s) execution passed.
+ """
+
+ response = self.execute(process_reply=process_reply,
+ ignore_errors=ignore_errors,
+ timeout=timeout)
+
+ if response.ret_code == 0:
+ raise AssertionError(err_msg)
@staticmethod
def _process_api_data(api_d):
api_data_processed = list()
for api in api_d:
- api_name = api['api_name']
- api_args = api['api_args']
- api_processed = dict(api_name=api_name)
api_args_processed = dict()
- for a_k, a_v in api_args.iteritems():
+ for a_k, a_v in api["api_args"].iteritems():
value = binascii.hexlify(a_v) if isinstance(a_v, str) else a_v
api_args_processed[str(a_k)] = value
- api_processed['api_args'] = api_args_processed
- api_data_processed.append(api_processed)
+ api_data_processed.append(dict(api_name=api["api_name"],
+ api_args=api_args_processed))
return api_data_processed
@staticmethod
Apply binascii.unhexlify() method for unicode values.
+ TODO: Remove the disabled code when definitely not needed.
+
:param api_r: API reply.
:type api_r: dict
:returns: Processed API reply / a part of API reply.
"""
if isinstance(api_reply, list):
- reverted_reply = list()
- for a_r in api_reply:
- reverted_reply.append(self._revert_api_reply(a_r))
+ reverted_reply = [self._revert_api_reply(a_r) for a_r in api_reply]
else:
reverted_reply = self._revert_api_reply(api_reply)
return reverted_reply
- def _process_json_data(self):
- """Process received JSON data."""
-
- for data in self._json_data:
- api_name = data['api_name']
- api_reply = data['api_reply']
- api_reply_processed = dict(
- api_name=api_name, api_reply=self._process_reply(api_reply))
- self._api_reply.append(api_reply_processed)
-
- def execute_papi(self, api_data, timeout=120):
+ def _execute_papi(self, api_data, timeout=120):
"""Execute PAPI command(s) on remote node and store the result.
:param api_data: List of APIs with their arguments.
:param timeout: Timeout in seconds.
:type api_data: list
:type timeout: int
- :raises SSHTimeout: If PAPI command(s) execution is timed out.
- :raises PapiInitError: If PAPI initialization failed.
- :raises PapiJsonFileError: If no api.json file found.
- :raises PapiCommandError: If PAPI command(s) execution failed.
- :raises PapiCommandInputError: If invalid attribute name or invalid
- value is used in API call.
+ :raises SSHTimeout: If PAPI command(s) execution has timed out.
:raises RuntimeError: If PAPI executor failed due to another reason.
"""
- self._api_data = api_data
+
+ if not api_data:
+ RuntimeError("No API data provided.")
+
api_data_processed = self._process_api_data(api_data)
json_data = json.dumps(api_data_processed)
ret_code, stdout, stderr = self._ssh.exec_command_sudo(
cmd=cmd, timeout=timeout)
except SSHTimeout:
- logger.error('PAPI command(s) execution timeout on host {host}:'
- '\n{apis}'.format(host=self._node['host'],
- apis=self._api_data))
- raise
- except (PapiInitError, PapiJsonFileError, PapiCommandError,
- PapiCommandInputError):
- logger.error('PAPI command(s) execution failed on host {host}'.
- format(host=self._node['host']))
+ logger.error("PAPI command(s) execution timeout on host {host}:"
+ "\n{apis}".format(host=self._node["host"],
+ apis=api_data))
raise
- except:
- raise RuntimeError('PAPI command(s) execution on host {host} '
- 'failed: {apis}'.format(host=self._node['host'],
- apis=self._api_data))
-
- self._ret_code = ret_code
- self._stdout = stdout
- self._stderr = stderr
-
- def papi_should_have_failed(self):
- """Read return code from last executed script and raise exception if the
- PAPI command(s) didn't fail.
-
- :raises RuntimeError: When no PAPI command executed.
- :raises AssertionError: If PAPI command(s) execution passed.
- """
-
- if self._ret_code is None:
- raise RuntimeError("First execute the PAPI command(s)!")
- if self._ret_code == 0:
- raise AssertionError(
- "PAPI command(s) execution passed, but failure was expected: "
- "{apis}".format(apis=self._api_data))
-
- def papi_should_have_passed(self):
- """Read return code from last executed script and raise exception if the
- PAPI command(s) failed.
-
- :raises RuntimeError: When no PAPI command executed.
- :raises AssertionError: If PAPI command(s) execution failed.
- """
-
- if self._ret_code is None:
- raise RuntimeError("First execute the PAPI command(s)!")
- if self._ret_code != 0:
- raise AssertionError(
- "PAPI command(s) execution failed, but success was expected: "
- "{apis}".format(apis=self._api_data))
-
- def get_papi_stdout(self):
- """Returns value of stdout from last executed PAPI command(s)."""
-
- return self._stdout
-
- def get_papi_stderr(self):
- """Returns value of stderr from last executed PAPI command(s)."""
-
- return self._stderr
-
- def get_papi_reply(self):
- """Returns api reply from last executed PAPI command(s)."""
-
- self._json_data = json.loads(self._stdout)
- self._process_json_data()
-
- return self._api_reply
+ except Exception:
+ raise RuntimeError("PAPI command(s) execution on host {host} "
+ "failed: {apis}".format(host=self._node["host"],
+ apis=api_data))
+ return ret_code, stdout, stderr
from resources.libraries.python.Constants import Constants
from resources.libraries.python.DUTSetup import DUTSetup
from resources.libraries.python.PapiExecutor import PapiExecutor
-from resources.libraries.python.PapiErrors import PapiError
from resources.libraries.python.ssh import exec_cmd, exec_cmd_no_error
from resources.libraries.python.topology import NodeType
from resources.libraries.python.VatExecutor import VatExecutor
@staticmethod
def vpp_show_version(node, verbose=False):
- """Run "show_version" API command.
+ """Run "show_version" PAPI command.
:param node: Node to run command on.
:param verbose: Show version, compile date and compile location if True
otherwise show only version.
:type node: dict
:type verbose: bool
- :raises PapiError: If no reply received for show_version API command.
"""
- # TODO: move composition of api data to separate method
- api_data = list()
- api = dict(api_name='show_version')
- api_args = dict()
- api['api_args'] = api_args
- api_data.append(api)
-
- api_reply = None
- with PapiExecutor(node) as papi_executor:
- papi_executor.execute_papi(api_data)
- try:
- papi_executor.papi_should_have_passed()
- except AssertionError:
- raise RuntimeError('Failed to get VPP version on host: {host}'.
- format(host=node['host']))
- api_reply = papi_executor.get_papi_reply()
-
- if api_reply is not None:
- version_data = api_reply[0]['api_reply']['show_version_reply']
- ver = version_data['version'].rstrip('\0x00')
- if verbose:
- date = version_data['build_date'].rstrip('\0x00')
- loc = version_data['build_directory'].rstrip('\0x00')
- version = \
- 'VPP Version: {ver}\n' \
- 'Compile date: {date}\n' \
- 'Compile location: {loc}\n '\
- .format(ver=ver, date=date, loc=loc)
- else:
- version = 'VPP version:{ver}'.format(ver=ver)
- logger.info(version)
- else:
- raise PapiError('No reply received for show_version API command on '
- 'host {host}'.format(host=node['host']))
+
+ with PapiExecutor(node) as papi_exec:
+ papi_resp = papi_exec.add('show_version').execute_should_pass()
+ data = papi_resp.reply[0]['api_reply']['show_version_reply']
+ version = ('VPP version: {ver}\n'.
+ format(ver=data['version'].rstrip('\0x00')))
+ if verbose:
+ version += ('Compile date: {date}\n'
+ 'Compile location: {loc}\n '.
+ format(date=data['build_date'].rstrip('\0x00'),
+ loc=data['build_directory'].rstrip('\0x00')))
+ logger.info(version)
@staticmethod
def vpp_show_version_verbose(node):
:type node: dict
:returns: VPP thread data.
:rtype: list
- :raises RuntimeError: If failed to run command on host.
- :raises PapiError: If no API reply received.
"""
- api_data = list()
- api = dict(api_name='show_threads')
- api_args = dict()
- api['api_args'] = api_args
- api_data.append(api)
-
- with PapiExecutor(node) as papi_executor:
- papi_executor.execute_papi(api_data)
- try:
- papi_executor.papi_should_have_passed()
- api_reply = papi_executor.get_papi_reply()
- except AssertionError:
- raise RuntimeError('Failed to run {api_name} on host '
- '{host}!'.format(host=node['host'], **api))
-
- if api_reply:
- return \
- api_reply[0]['api_reply']['show_threads_reply']['thread_data']
- else:
- raise PapiError('No reply received for {api_name} on host {host}!'.
- format(host=node['host'], **api))
+
+ with PapiExecutor(node) as papi_exec:
+ resp = papi_exec.add('show_threads').execute_should_pass()
+ return resp.reply[0]['api_reply']['show_threads_reply']['thread_data']
#!/usr/bin/env python
-# Copyright (c) 2018 Cisco and/or its affiliates.
+# Copyright (c) 2019 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:
import argparse
import binascii
-import fnmatch
import json
import os
import sys
-sys.path.append('/tmp/openvpp-testing')
-try:
- from resources.libraries.python.PapiErrors import *
-except:
- raise
-
# Sphinx creates auto-generated documentation by importing the python source
# files and collecting the docstrings from them. The NO_VPP_PAPI flag allows
# the vpp_papi_provider.py file to be importable without having to build
sys.path.append(modules_path)
from vpp_papi import VPP
else:
- raise PapiInitError('vpp_papi module not found')
+ raise RuntimeError('vpp_papi module not found')
# client name
CLIENT_NAME = 'csit_papi'
vpp = VPP()
return vpp
except Exception as err:
- raise PapiInitError('PAPI init failed:\n{exc}'.format(exc=repr(err)))
+ raise RuntimeError('PAPI init failed:\n{err}'.format(err=repr(err)))
def papi_connect(vpp_client, name='vpp_api'):
def papi_run(vpp_client, api_name, api_args):
- """api_name
+ """Run PAPI.
:param vpp_client: VPP instance.
:param api_name: VPP API name.
"""Process API reply / a part of API reply for smooth converting to
JSON string.
- # Apply binascii.hexlify() method for string values.
+ Apply binascii.hexlify() method for string values.
+
:param api_r: API reply.
:type api_r: Vpp_serializer reply object (named tuple)
:returns: Processed API reply / a part of API reply.
def main():
"""Main function for the Python API provider.
- :raises PapiCommandInputError: If invalid attribute name or invalid value is
- used in API call.
- :raises PapiCommandError: If PAPI command(s) execution failed.
+ :raises RuntimeError: If invalid attribute name or invalid value is
+ used in API call or if PAPI command(s) execution failed.
"""
parser = argparse.ArgumentParser()
reply.append(api_reply)
except (AttributeError, ValueError) as err:
papi_disconnect(vpp)
- raise PapiCommandInputError(
- 'PAPI command {api}({args}) input error:\n{exc}'.format(
- api=api_name, args=api_args), exc=repr(err))
+ raise RuntimeError('PAPI command {api}({args}) input error:\n{err}'.
+ format(api=api_name,
+ args=api_args,
+ err=repr(err)))
except Exception as err:
papi_disconnect(vpp)
- raise PapiCommandError(
- 'PAPI command {api}({args}) error:\n{exc}'.format(
- api=api_name, args=api_args), exc=repr(err))
+ raise RuntimeError('PAPI command {api}({args}) error:\n{exc}'.
+ format(api=api_name,
+ args=api_args,
+ exc=repr(err)))
papi_disconnect(vpp)
return json.dumps(reply)
Code Review / csit.git / commitdiff