- def get_stats_reply(self, err_msg="Failed to get statistics.", timeout=120):
- """Get VPP Stats reply from VPP Python API.
-
- :param err_msg: The message used if the PAPI command(s) execution fails.
- :param timeout: Timeout in seconds.
- :type err_msg: str
- :type timeout: int
- :returns: Requested VPP statistics.
- :rtype: list
- """
-
- args = self._api_command_list[0]['api_args']
- self._api_command_list = list()
-
- stdout, _ = self._execute_papi(
- args, method='stats_request', err_msg=err_msg, timeout=timeout)
-
- return json.loads(stdout)
-
- def get_replies(self, err_msg="Failed to get replies.",
- process_reply=True, ignore_errors=False, timeout=120):
- """Get reply/replies from VPP Python API.
-
- :param err_msg: The message used if the PAPI command(s) execution fails.
- :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 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
- """
- return self._execute(
- method='request', process_reply=process_reply,
- ignore_errors=ignore_errors, err_msg=err_msg, timeout=timeout)
-
- def get_dump(self, err_msg="Failed to get dump.",
- process_reply=True, ignore_errors=False, timeout=120):
- """Get dump from VPP Python API.
-
- :param err_msg: The message used if the PAPI command(s) execution fails.
- :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 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
- """
- return self._execute(
- method='dump', process_reply=process_reply,
- ignore_errors=ignore_errors, err_msg=err_msg, timeout=timeout)
-
- @staticmethod
- def dump_and_log(node, cmds):
- """Dump and log requested information.
-
- :param node: DUT node.
- :param cmds: Dump commands to be executed.
- :type node: dict
- :type cmds: list
- """
- with PapiExecutor(node) as papi_exec:
- for cmd in cmds:
- dump = papi_exec.add(cmd).get_dump()
- logger.debug("{cmd}:\n{data}".format(
- cmd=cmd, data=pformat(dump.reply[0]["api_reply"])))
-
- @staticmethod
- def run_cli_cmd(node, cmd, log=True):
- """Run a CLI command.
-
- :param node: Node to run command on.
- :param cmd: The CLI command to be run on the node.
- :param log: If True, the response is logged.
- :type node: dict
- :type cmd: str
- :type log: bool
- :returns: Verified data from PAPI response.
- :rtype: dict
- """
-
- cli = 'cli_inband'
- args = dict(cmd=cmd)
- err_msg = "Failed to run 'cli_inband {cmd}' PAPI command on host " \
- "{host}".format(host=node['host'], cmd=cmd)
-
- with PapiExecutor(node) as papi_exec:
- data = papi_exec.add(cli, **args).get_replies(err_msg). \
- verify_reply(err_msg=err_msg)
-
- if log:
- logger.info("{cmd}:\n{data}".format(cmd=cmd, data=data["reply"]))
-
- return data
-
- 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.
-
- IMPORTANT!
- Do not use this method in L1 keywords. Use:
- - get_replies()
- - get_dump()
- This method will be removed soon.
-
- :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 AssertionError: If PAPI command(s) execution failed.
- """
- # TODO: Migrate callers to get_replies and delete this method.
- return self.get_replies(
- process_reply=process_reply, ignore_errors=ignore_errors,
- err_msg=err_msg, timeout=timeout)
-