+ if ret_code != 0:
+ raise AssertionError(err_msg)
+
+ return stdout, stderr
+
+ def _execute(self, method='request', process_reply=True,
+ ignore_errors=False, err_msg="", timeout=120):
+ """Turn internal command list into proper data and execute; return
+ PAPI response.
+
+ This method also clears the internal command list.
+
+ IMPORTANT!
+ Do not use this method in L1 keywords. Use:
+ - get_stats()
+ - get_replies()
+ - get_dump()
+
+ :param method: VPP Python API method. Supported methods are: 'request',
+ 'dump' and 'stats'.
+ :param process_reply: Process PAPI reply if True.
+ :param ignore_errors: If true, the errors in the reply are ignored.
+ :param err_msg: The message used if the PAPI command(s) execution fails.
+ :param timeout: Timeout in seconds.
+ :type method: str
+ :type process_reply: bool
+ :type ignore_errors: bool
+ :type err_msg: str
+ :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._api_command_list = list()
+
+ stdout, stderr = self._execute_papi(
+ local_list, method=method, err_msg=err_msg, timeout=timeout)
+ papi_reply = list()
+ if process_reply:
+ try:
+ json_data = json.loads(stdout)
+ except ValueError:
+ logger.error("An error occured while processing the PAPI "
+ "request:\n{rqst}".format(rqst=local_list))
+ raise
+ 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)
+
+ # Log processed papi reply to be able to check API replies changes
+ logger.debug("Processed PAPI reply: {reply}".format(reply=papi_reply))
+
+ return PapiResponse(
+ papi_reply=papi_reply, stdout=stdout, stderr=stderr,
+ requests=[rqst["api_name"] for rqst in local_list])