-# 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.Constants import Constants
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
Code Review / csit.git / blobdiff