pytest_mh.ssh

Classes

`Enum`(value)	Generic enumeration.
`MultihostLogger`(args, *kwargs)	Multihost logger class.
`SSHBashProcess`(args, *kwargs)	SSH Process with Bash.
`SSHClient`(host, *, user, password, port, ...)	Interactive SSH client.
`SSHLog`(value)	SSH command log level.
`SSHPowerShellProcess`(args, *kwargs)	SSH Process with Powershell.
`SSHProcess`(*, command[, cwd, env, input, shell])	SSH Process.
`SSHProcessResult`(rc, stdout, stderr)	SSH Process result.
`auto`()	Instances are replaced with an appropriate value in Enum class suites.

Exceptions

`SSHAuthenticationError`(host, port, user)
`SSHProcessError`(id, command, rc, cwd, env, ...)	SSH Process Error.

class pytest_mh.ssh.SSHLog(value)

Bases: Enum

SSH command log level.

Silent = 1: No log messages are produced.

Short = 2: Command execution and return code is logged. Its output is omitted.

Full = 3: Command execution, its return code and output is logged.

Error = 4: Only log the command and its result on non-zero exit code.

class pytest_mh.ssh.SSHProcess(*, command: str, cwd: str | None = None, env: dict[str, Any] | None = None, input: str | None = None, shell: str | None = None, conn: SSHClient, read_timeout: float, logger: MultihostLogger, log_level: SSHLog, sync_exec: bool)

Bases: object

SSH Process.

Note

You should not create instances of this class yourself. Use method SSHClient.run(), SSHClient.exec(), SSHClient.async_run() and SSHClient.async_exec() from SSHClient to execute a command over SSH.

Parameters:

command (str) – Command to execute.
cwd (str | None, optional) – Working directory, defaults to None
env (dict[str, Any] | None, optional) – Additional environment variables, defaults to None
input (str | None, optional) – Content of standard input, defaults to None
shell (str | None, optional) – Shell used to execute the command, defaults to None (use user’s login shell)
conn (pssh.clients.ssh.SSHClient) – Connected SSH client.
read_timeout (float) – Timeout in seconds, how long should the client wait for output, defaults to 30 seconds
logger (MultihostLogger) – Multihost logger.
log_level (SSHLog) – Log level.
sync_exec (bool) – Is this a blocking execution?

property stdout: Generator[str, None, None]

Standard output, returns generator which yields output line by line.

# Read single line, this will block until there is a line to read or read_timeout is reached
line = next(process.stdout)

# Read all lines, this will block until EOF or read_timeout is reached
lines = list(process.stdout)

# Iterate over all lines
for line in process.stdout:
    pass

Raises:: RuntimeError – If the process is not yet started.
Returns:: Standard output generator.
Return type:: Generator[str, None, None]

property stderr: Generator[str, None, None]

Standard error output, returns generator which yields error output line by line.

# Read single line, this will block until there is a line to read or read_timeout is reached
line = next(process.stderr)

# Read all lines, this will block until EOF or read_timeout is reached
lines = list(process.stderr)

# Iterate over all lines
for line in process.stderr:
    pass

Raises:: RuntimeError – If the process is not yet started.
Returns:: Standard error output generator.
Return type:: Generator[str, None, None]

property stdin: Stdin

File-like object representing command’s standard input.

# Write data
process.stdin.write('Hello World')

# Send EOF to indicate that there will be no more input data.
process.send_eof()

Raises:: RuntimeError – If the process is not yet started.
Returns:: Standard input file.
Return type:: pssh.clients.base.single.Stdin

run() → SSHProcess

Execute the command.

Returns:: Self.
Return type:: SSHProcess

wait(raise_on_error: bool = True) → SSHProcessResult

Wait for the command to finish.

EOF is send to standard input to indicate that there will be no additional input data. Then it waits for the command to finish.

Parameters:: raise_on_error (bool, optional) – If True, SSHProcessError is raised on non-zero return code, defaults to True
Raises:: SSHProcessError – If raise_on_error is True and the command exited with non-zero return code.
Returns:: Command result.
Return type:: SSHProcessResult

send_eof() → None

Send EOF to standard input to indicate that there will be no more input data.

Raises:: RuntimeError – If the process is not yet started.

class pytest_mh.ssh.SSHBashProcess(*args, **kwargs)

Bases: SSHProcess

SSH Process with Bash.

Note

You should not create instances of this class yourself. Use method SSHClient.run(), SSHClient.exec(), SSHClient.async_run() and SSHClient.async_exec() from SSHClient to execute a command over SSH.

Parameters:

command (str) – Command to execute.
cwd (str | None, optional) – Working directory, defaults to None
env (dict[str, Any] | None, optional) – Additional environment variables, defaults to None
input (str | None, optional) – Content of standard input, defaults to None
shell (str | None, optional) – Shell used to execute the command, defaults to None (use user’s login shell)
conn (pssh.clients.ssh.SSHClient) – Connected SSH client.
read_timeout (float) – Timeout in seconds, how long should the client wait for output, defaults to 30 seconds
logger (MultihostLogger) – Multihost logger.
log_level (SSHLog) – Log level.
sync_exec (bool) – Is this a blocking execution?

class pytest_mh.ssh.SSHPowerShellProcess(*args, **kwargs)

Bases: SSHProcess

SSH Process with Powershell.

Note

You should not create instances of this class yourself. Use method SSHClient.run(), SSHClient.exec(), SSHClient.async_run() and SSHClient.async_exec() from SSHClient to execute a command over SSH.

Parameters:

command (str) – Command to execute.
cwd (str | None, optional) – Working directory, defaults to None
env (dict[str, Any] | None, optional) – Additional environment variables, defaults to None
input (str | None, optional) – Content of standard input, defaults to None
shell (str | None, optional) – Shell used to execute the command, defaults to None (use user’s login shell)
conn (pssh.clients.ssh.SSHClient) – Connected SSH client.
read_timeout (float) – Timeout in seconds, how long should the client wait for output, defaults to 30 seconds
logger (MultihostLogger) – Multihost logger.
log_level (SSHLog) – Log level.
sync_exec (bool) – Is this a blocking execution?

class pytest_mh.ssh.SSHProcessResult(rc: int, stdout: list[str], stderr: list[str])

Bases: object

SSH Process result.

Parameters:

rc (int) – Return code.
stdout (list[str]) – Standard output, line by line.
stderr (list[str]) – Standard error output, line by line.

exception pytest_mh.ssh.SSHProcessError(id: int, command: str, rc: int, cwd: str | None, env: dict[str, Any], input: str | None, stdout: str, stderr: str)

Bases: Exception

SSH Process Error.

exception pytest_mh.ssh.SSHAuthenticationError(host: str, port: int, user: str): Bases: Exception

class pytest_mh.ssh.SSHClient(host: str, *, user: str, password: str, port: int = 22, shell: ~typing.Type[~pytest_mh.ssh.SSHProcess] = <class 'pytest_mh.ssh.SSHProcess'>, logger: ~pytest_mh._private.logging.MultihostLogger)

Bases: object

Interactive SSH client.

Example: Blocking call

# Connect to SSH server, it is automatically disconnected when leaving the with statement
with SSHClient(host, user=username, password=password, logger=logger) as ssh:
    result = ssh.run('echo Hello World')
    print(result.rc)
    print(result.stdout)

    result = ssh.run('cat', input='Hello World')
    print(result.rc)
    print(result.stdout)

Example: Non-blocking call

# Connect to SSH server, it is automatically disconnected when leaving the with statement
with SSHClient(host, user=username, password=password, logger=logger) as ssh:
    # The process is executed, but it does not block. In order to wait for it to finish, run process.wait()
    process = ssh.async_run('echo Hello World')
    result = process.wait()
    print(result.rc)
    print(result.stdout)

    # You can write to stdin directly in asynchronous run
    process = ssh.async_run('cat')
    process.stdin.write('Hello World')
    process.send_eof()
    result = process.wait()
    print(result.rc)
    print(result.stdout)

    # You can also work with inputs and outputs more interactively.
    # The process is automatically waited when leaving the with statement.
    with ssh.async_run('bash') as process:
        process.stdin.write('echo Hello World\n')
        print(next(process.stdout))

        process.stdin.write('echo This works as well\n')
        print(next(process.stdout))

Note

It is possible to set MH_SSH_DEBUG=yes environment variable to log output and exist status to from commands, regardless of what log level is used. This essentially enforces the SSHLog.Full level.

Parameters:

host (BaseRole | str) – Host name to connect to.
user (str) – Username to authenticate.
password (str) – Password.
logger (MultihostLogger) – Multihost logger.
port (int, optional) – SSH port, defaults to 22
shell (str, optional) – User shell used to run commands, defaults to ‘/usr/bin/bash -c’

property connected: bool

Returns:: True if the client is connected, False otherwise.
Return type:: bool

property conn: SSHClient

Low-level connection object.

Returns:: Parallel-ssh connection object.
Return type:: pssh.clients.ssh.SSHClient

connect() → None

Connect to the host.

Raises:: SSHAuthenticationError – If user fails to authenticate.

disconnect() → None: Disconnect client.

async_run(command: str, *, cwd: str | None = None, env: dict[str, Any] | None = None, input: str | None = None, read_timeout: float = 30, log_level: SSHLog = SSHLog.Full) → SSHProcess

Non-blocking command call.

The command is run under shell specified in the constructor and it is executed immediately, however it does not wait for the command to finish.

Parameters:

command (str) – Command to run.
cwd (str | None, optional) – Working directory, defaults to None (= do not change)
env (dict[str, Any] | None, optional) – Additional environment variables, defaults to None
input (str | None, optional) – Content of standard input, defaults to None
read_timeout (float, optional) – Timeout in seconds, how long should the client wait for output, defaults to 30 seconds
log_level (SSHLog, optional) – Log level, defaults to SSHLog.Full

Returns:

Instance of SSHProcess, the process is already running.

Return type:

SSHProcess

run(command: str, *, cwd: str | None = None, env: dict[str, Any] | None = None, input: str | None = None, read_timeout: float = 2, log_level: SSHLog = SSHLog.Full, raise_on_error: bool = True) → SSHProcessResult

Blocking command call.

The command is run under shell specified in the constructor and it is executed immediately. It waits for the command to finish and returns its result.

Parameters:

command (str) – Command to run.
cwd (str | None, optional) – Working directory, defaults to None (= do not change)
env (dict[str, Any] | None, optional) – Additional environment variables, defaults to None
input (str | None, optional) – Content of standard input, defaults to None
read_timeout (float, optional) – Timeout in seconds, how long should the client wait for output, defaults to 30 seconds
log_level (SSHLog, optional) – Log level, defaults to SSHLog.Full
raise_on_error (bool, optional) – If True, raise SSHProcessError if command exited with non-zero return code, defaults to True

Raises:

SSHProcessError – If raise_on_error is True and the command exited with non-zero return code.

Returns:

Command result.

Return type:

SSHProcessResult

async_exec(argv: list[Any], *, cwd: str | None = None, env: dict[str, Any] | None = None, input: str | None = None, read_timeout: float = 2, log_level: SSHLog = SSHLog.Full) → SSHProcess

Non-blocking command call.

The command is run under shell specified in the constructor and it is executed immediately, however it does not wait for the command to finish.

The command is provided as argv list.

Parameters:

argv (list[Any]) – Command to run.
cwd (str | None, optional) – Working directory, defaults to None (= do not change)
env (dict[str, Any] | None, optional) – Additional environment variables, defaults to None
input (str | None, optional) – Content of standard input, defaults to None
read_timeout (float, optional) – Timeout in seconds, how long should the client wait for output, defaults to 30 seconds
log_level (SSHLog, optional) – Log level, defaults to SSHLog.Full

Returns:

Instance of SSHProcess, the process is already running.

Return type:

SSHProcess

exec(argv: list[Any], *, cwd: str | None = None, env: dict[str, Any] | None = None, input: str | None = None, read_timeout: float = 2, log_level: SSHLog = SSHLog.Full, raise_on_error: bool = True) → SSHProcessResult

Blocking command call.

The command is run under shell specified in the constructor and it is executed immediately. It waits for the command to finish and returns its result.

The command is provided as argv list.

Parameters:

argv (list[Any]) – Command to run.
cwd (str | None, optional) – Working directory, defaults to None (= do not change)
env (dict[str, Any] | None, optional) – Additional environment variables, defaults to None
input (str | None, optional) – Content of standard input, defaults to None
read_timeout (float, optional) – Timeout in seconds, how long should the client wait for output, defaults to 30 seconds
log_level (SSHLog, optional) – Log level, defaults to SSHLog.Full
raise_on_error (bool, optional) – If True, raise SSHProcessError if command exited with non-zero return code, defaults to True

Raises:

SSHProcessError – If raise_on_error is True and the command exited with non-zero return code.

Returns:

Command result.

Return type:

SSHProcessResult

expect(expect_script: str, *, verbose: bool = True, raise_on_error: bool = False) → SSHProcessResult

Run expect script.

Parameters:

expect_script (str) – Expect script.
verbose (bool, optional) – Enable expect debug output (-d), default to True.
raise_on_error (bool, optional) – If True, raise SSHProcessError if command exited with non-zero return code, defaults to False

Returns:

Expect script result.

Return type:

SSHProcessResult

expect_nobody(expect_script: str, *, verbose: bool = True, raise_on_error: bool = False) → SSHProcessResult

Run expect script as user nobody.

The main use case is to avoid running the command as root if the client is connected to the root user SSH session.

Parameters:

expect_script (str) – Expect script.
verbose (bool, optional) – Enable expect debug output (-d), default to True.
raise_on_error (bool, optional) – If True, raise SSHProcessError if command exited with non-zero return code, defaults to False

Returns:

Expect return code.

Return type:

SSHProcessResult