from paramiko.ssh_exception import SSHException
from scp import SCPClient
from robot.api import logger
-from robot.utils.asserts import assert_equal
__all__ = ["exec_cmd", "exec_cmd_no_error"]
command = 'sudo -S {c}'.format(c=command)
return self.exec_command(command, timeout)
- def interactive_terminal_open(self, time_out=30):
+ def interactive_terminal_open(self, time_out=45):
"""Open interactive terminal on a new channel on the connected Node.
:param time_out: Timeout in seconds.
chan.set_combine_stderr(True)
buf = ''
- while not buf.endswith((":~$ ", "~]$ ", "~]# ")):
+ while not buf.endswith((":~# ", ":~$ ", "~]$ ", "~]# ")):
try:
chunk = chan.recv(self.__MAX_RECV_BUF)
if not chunk:
"""
chan.close()
- def scp(self, local_path, remote_path, get=False, timeout=30):
+ def scp(self, local_path, remote_path, get=False, timeout=30,
+ wildcard=False):
"""Copy files from local_path to remote_path or vice versa.
connect() method has to be called first!
path to remote file which should be downloaded.
:param get: scp operation to perform. Default is put.
:param timeout: Timeout value in seconds.
+ :param wildcard: If path has wildcard characters. Default is false.
:type local_path: str
:type remote_path: str
:type get: bool
:type timeout: int
+ :type wildcard: bool
"""
if not get:
logger.trace('SCP {0} to {1}:{2}'.format(
self._ssh.get_transport().getpeername(), remote_path,
local_path))
# SCPCLient takes a paramiko transport as its only argument
- scp = SCPClient(self._ssh.get_transport(), socket_timeout=timeout)
+ if not wildcard:
+ scp = SCPClient(self._ssh.get_transport(), socket_timeout=timeout)
+ else:
+ scp = SCPClient(self._ssh.get_transport(), sanitize=lambda x: x,
+ socket_timeout=timeout)
start = time()
if not get:
scp.put(local_path, remote_path)
raise TypeError('Node parameter is None')
if cmd is None:
raise TypeError('Command parameter is None')
- if len(cmd) == 0:
+ if not cmd:
raise ValueError('Empty command parameter')
ssh = SSH()
return ret_code, stdout, stderr
-def exec_cmd_no_error(node, cmd, timeout=600, sudo=False):
+def exec_cmd_no_error(node, cmd, timeout=600, sudo=False, message=None):
"""Convenience function to ssh/exec/return out & err.
Verifies that return code is zero.
- Returns (stdout, stderr).
+ :param node: DUT node.
+ :param cmd: Command to be executed.
+ :param timeout: Timeout value in seconds. Default: 600.
+ :param sudo: Sudo privilege execution flag. Default: False.
+ :param message: Error message in case of failure. Default: None.
+ :type node: dict
+ :type cmd: str
+ :type timeout: int
+ :type sudo: bool
+ :type message: str
+ :returns: Stdout, Stderr.
+ :rtype: tuple(str, str)
+ :raise RuntimeError: If bash return code is not 0.
"""
- (ret_code, stdout, stderr) = exec_cmd(node, cmd, timeout=timeout, sudo=sudo)
- assert_equal(ret_code, 0, 'Command execution failed: "{}"\n{}'.
- format(cmd, stderr))
+ ret_code, stdout, stderr = exec_cmd(node, cmd, timeout=timeout, sudo=sudo)
+ msg = ('Command execution failed: "{cmd}"\n{stderr}'.
+ format(cmd=cmd, stderr=stderr) if message is None else message)
+ if ret_code != 0:
+ raise RuntimeError(msg)
+
return stdout, stderr