pytest_mh.conn.ssh

Classes

SSHClient(host, *, user[, password, ...])

Interactive SSH client.

SSHProcess(*, command[, cwd, env, input])

SSH Process manager.

SSHInputBuffer(channel)

SSH Input Buffer.

SSHProcessResult(rc, stdout, stderr, error)

SSH Process result.

Exceptions

SSHAuthenticationError(host, port, user, message)

Unable to authenticate over SSH.

SSHProcessError(rc, id, command, cwd, env, ...)

SSH Process Error.

SSHProcessTimeoutError(timeout, id, command, ...)

SSH Process Timeout Error.
class pytest_mh.conn.ssh.SSHClient(host: str, *, user: str, password: str | None = None, private_key_path: str | Path | None = None, private_key_password: str | None = None, port: int = 22, shell: Shell, logger: MultihostLogger, timeout: int = 300)

Bases: Connection[SSHProcess, SSHProcessResult]

Interactive SSH client.

Parameters:

  • host (str) – Host name to connect to.

  • user (str) – Username to authenticate.

  • password (str | None) – Password for authentication, defaults to None.

  • private_key_path (str | Path | None) – Path to the private key for authentication, defaults to None.

  • private_key_password (str | None) – Password to unlock the private key, defaults to None.

  • port (int, optional) – SSH port, defaults to 22

  • shell (str, optional) – User shell used to run commands, defaults to SSHBashProcess

  • logger (MultihostLogger) – Multihost logger.

  • timeout (int) – Timeout in seconds (defaults to 300), value 0 means that timeout is disabled.

property requested_auth_methods: list[str]
Returns:

List of authentication methods requested by the caller.

Return type:

list[str]

property connected: bool
Returns:

True if the connection is established, False otherwise.

Return type:

bool

connect() → None

Connect to the host.

Raises:

SSHAuthenticationError – If user fails to authenticate.

disconnect() → None

Disconnect.

create_process(*, command: str, cwd: str | None = None, env: dict[str, Any] | None = None, input: str | bytes | None = None, log_level: ProcessLogLevel, blocking_call: bool) → SSHProcess

Create a new process.

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 | bytes | None, optional) – Content of standard input, defaults to None

  • log_level (ProcessLogLevel) – Log level.

  • blocking_call (bool) – Is this a blocking execution?

Returns:

Newly created process that is not yet running.

Return type:

ProcessType

classmethod from_confdict(host: MultihostHost, confdict: dict[str, Any]) → Self

Create new instance of this class from configuration dictionary.

Parameters:

  • host (MultihostHost) – Multihost host that will use this connection.

  • confdict (dict[str, Any]) – Configuration dictionary.

Returns:

New instance.

Return type:

Self

exception pytest_mh.conn.ssh.SSHAuthenticationError(host: str, port: int, user: str, message: str)

Bases: ConnectionError

Unable to authenticate over SSH.

class pytest_mh.conn.ssh.SSHProcess(*, command: str, cwd: str | None = None, env: dict[str, Any] | None = None, input: str | bytes | None = None, shell: Shell, logger: MultihostLogger, log_level: ProcessLogLevel, timeout: int, blocking_call: bool, client: SSHClient, conn: Session)

Bases: Process[SSHProcessResult, SSHInputBuffer, SSHProcessTimeoutError]

SSH Process manager.

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 | bytes | 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)

  • logger (MultihostLogger) – Multihost logger.

  • log_level (ProcessLogLevel) – Log level.

  • timeout (int) – Timeout in seconds, value 0 means that timeout is disabled.

  • blocking_call (bool) – Is this a blocking execution?

  • client (SSHClient) – SSH client.

  • conn (LibsshSession) – Connected SSH session.

property in_progress: bool
Returns:

True if the process is already started and running.

Return type:

bool

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
# EOF is reached
line = next(process.stdout)

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

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

RuntimeError – If the process is not running.

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
# EOF is reached
line = next(process.stderr)

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

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

RuntimeError – If the process is not running.

Returns:

Standard error output generator.

Return type:

Generator[str, None, None]

property stdin: SSHInputBuffer

Command’s standard input.

Call send_eof() to close the standard input and notify the process that there will be no more input data.

# 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 running.

Returns:

Standard input file.

Return type:

ProcessInputBufferType

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 running.

send_signal(sig: Signals) → None

Send signal to the running process.

Parameters:

sig (signal.Signals) – A signal constant from signal, e.g. signal.SIGUSR1.

Raises:

RuntimeError – If the process is not running.

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

Bases: ProcessError

SSH Process Error.

exception pytest_mh.conn.ssh.SSHProcessTimeoutError(timeout: int, id: int, command: str, cwd: str | None, env: dict[str, Any], input: str | bytes | None, stdout: list[str], stderr: list[str])

Bases: ProcessTimeoutError

SSH Process Timeout Error.

class pytest_mh.conn.ssh.SSHInputBuffer(channel: Channel)

Bases: ProcessInputBuffer

SSH Input Buffer.

Allows to write into stdin of opened SSH channel.

Parameters:

channel (LibsshChannel) – Opened libssh channel.

write(data: str | bytes) → None

Write data to stdin.

Parameters:

data (str | bytes) – Data to write.

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

Bases: ProcessResult[SSHProcessError]

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.

  • error (ProcessErrorType) – Process error object that can be raised manually with throw().