pytest_mh.conn

Module Attributes

`ProcessErrorType`	Generic process error type.
`ProcessResultType`	Generic process result type.
`ProcessInputBufferType`	Generic process input buffer type.
`ProcessType`	Generic process type.

Classes

`Bash`()	Bash shell abstraction.
`Connection`(*, shell, logger[, timeout])	Abstract connection to the host.
`Powershell`()	Powershell shell abstraction.
`Process`(*, command[, cwd, env, input, ...])	Process manager.
`ProcessInputBuffer`()	Process' input buffer.
`ProcessLogLevel`(value)	Process log level.
`ProcessResult`(rc, stdout, stderr, error)	Process' result.
`Shell`(name, shell_command)	Multihost shell abstraction.

Exceptions

`ConnectionError`	Unable to connect to the host.
`ProcessError`(rc, id, command, cwd, env, ...)	Process Error.
`ProcessTimeoutError`(timeout, id, command, ...)	Process Timeout Error.

class pytest_mh.conn.Bash

Bases: Shell

Bash shell abstraction.

Parameters:

name (str) – Shell name.
shell_command – Shell command that will execute user scripts.
shell_command – str

build_command_line(script: str, *, cwd: str | None, env: dict[str, Any]) → str

Create a complete command that will execute given script in this shell.

User is able to specify a different working directory and additional environment variables. This method adds this information to the user’s command and returns a full script that will switch the working directory, sets environment variables and then executes the given command or script.

Parameters:

script (str) – Script
cwd (str | None) – Working directory, None means no change.
env (dict[str, Any]) – Additional environment variables.

Returns:

Complete command line to run to execute the script in this shell.

Return type:

str

class pytest_mh.conn.Connection(*, shell: Shell, logger: MultihostLogger, timeout: int = 300)

Bases: ABC, Generic[ProcessType, ProcessResultType]

Abstract connection to the host.

This is an abstract class. The following examples utilizes ssh.SSHClient which implements this interface.

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_CONNECTION_DEBUG=yes environment variable to log output and exit status to from commands, regardless of what log level is used. This essentially enforces the ProcessLogLevel.Full level.

Parameters:

shell (str, optional) – Shell used to run commands and scripts.
logger (MultihostLogger) – Multihost logger.
timeout (int) – Timeout in seconds (defaults to 300), value 0 means that timeout is disabled.

shell: Shell: Shell used to run commands and scripts.

logger: MultihostLogger: Multihost logger.

timeout: int: Default timeout for command execution.

abstract property connected: bool

Returns:: True if the connection is established, False otherwise.
Return type:: bool

abstractmethod connect() → None

Connect to the host.

Raises:: ConnectionError – If connection can not be established.

abstractmethod disconnect() → None: Disconnect.

abstractmethod 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) → ProcessType

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

async_run(command: str, *, cwd: str | None = None, env: dict[str, Any] | None = None, input: str | bytes | None = None, log_level: ProcessLogLevel = ProcessLogLevel.Full) → ProcessType

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 | bytes | None, optional) – Content of standard input, defaults to None
log_level (ProcessLogLevel, optional) – Log level, defaults to ProcessLogLevel.Full

Returns:

Instance of Process, the process is already running.

Return type:

ProcessType

run(command: str, *, cwd: str | None = None, env: dict[str, Any] | None = None, input: str | None = None, log_level: ProcessLogLevel = ProcessLogLevel.Full, raise_on_error: bool = True, timeout: int | None = None) → ProcessResultType

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
log_level (ProcessLogLevel, optional) – Log level, defaults to ProcessLogLevel.Full
raise_on_error (bool, optional) – If True, raise ProcessError if command exited with non-zero return code, defaults to True
timeout (int | None) – Timeout in seconds (defaults to None = use default connection timeout)

Raises:

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

Returns:

Command result.

Return type:

ProcessResultType

async_exec(argv: list[Any], *, cwd: str | None = None, env: dict[str, Any] | None = None, input: str | None = None, log_level: ProcessLogLevel = ProcessLogLevel.Full) → ProcessType

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
log_level (ProcessLogLevel, optional) – Log level, defaults to ProcessLogLevel.Full

Returns:

Instance of Process, the process is already running.

Return type:

ProcessType

exec(argv: list[Any], *, cwd: str | None = None, env: dict[str, Any] | None = None, input: str | None = None, log_level: ProcessLogLevel = ProcessLogLevel.Full, raise_on_error: bool = True, timeout: int | None = None) → ProcessResultType

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
log_level (ProcessLogLevel, optional) – Log level, defaults to ProcessLogLevel.Full
raise_on_error (bool, optional) – If True, raise ProcessError if command exited with non-zero return code, defaults to True
timeout (int | None) – Timeout in seconds (defaults to None = use default connection timeout)

Raises:

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

Returns:

Command result.

Return type:

ProcessResultType

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

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 ProcessError if command exited with non-zero return code, defaults to False

Returns:

Expect script result.

Return type:

ProcessResultType

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

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 ProcessError if command exited with non-zero return code, defaults to False

Returns:

Expect return code.

Return type:

ProcessResultType

abstractmethod 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.ConnectionError

Bases: Exception

Unable to connect to the host.

class pytest_mh.conn.Powershell

Bases: Shell

Powershell shell abstraction.

Parameters:

name (str) – Shell name.
shell_command – Shell command that will execute user scripts.
shell_command – str

build_command_line(script: str, *, cwd: str | None, env: dict[str, Any]) → str

Create a complete command that will execute given script in this shell.

User is able to specify a different working directory and additional environment variables. This method adds this information to the user’s command and returns a full script that will switch the working directory, sets environment variables and then executes the given command or script.

Parameters:

script (str) – Script
cwd (str | None) – Working directory, None means no change.
env (dict[str, Any]) – Additional environment variables.

Returns:

Complete command line to run to execute the script in this shell.

Return type:

str

class pytest_mh.conn.Process(*, 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, additional_log_data: dict[str, Any] | None = None)

Bases: ABC, Generic[ProcessResultType, ProcessInputBufferType, ProcessTimeoutErrorType]

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 (Shell) – Shell that will execute the command
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?
additional_log_data (dict[str, Any] | None, optional) – Additional data that will be added to the log messages, defaults to None

id: int: Autogenerated command ID.

command: str: Executed command.

cwd: str | None: Working directory.

env: dict[str, Any]: Additional environment variables.

input: str | bytes | None: Input data.

shell: Shell: Shell that executes the command.

logger: MultihostLogger: Multihost logger.

timeout: int: Timeout.

blocking_call: bool: True if this is a blocking call.

additional_log_data: dict[str, Any]: Additional data that will be added to the log messages.

full_command_line: str: Full command line that will be executed.

log_level: ProcessLogLevel: Log level.

abstract property in_progress: bool

Returns:: True if the process is already started and running.
Return type:: bool

abstract 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]

abstract 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]

abstract property stdin: ProcessInputBufferType

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

run() → Self

Execute the command.

Returns:: Self.
Return type:: Self

wait(*, raise_on_error: bool = True, timeout: int | None = None) → ProcessResultType

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, ProcessError is raised on non-zero return code, defaults to True
timeout (int | None) – Timeout in seconds (default to None = use default value, 0 means no timeout)

Raises:

ProcessError – If raise_on_error is True and the command exited with non-zero return code.
ProcessTimeoutErrorType – If timeout occurred.

Returns:

Command result.

Return type:

ProcessResultType

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

abstractmethod 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.ProcessError(rc: int, id: int, command: str, cwd: str | None, env: dict[str, Any], input: str | bytes | None, stdout: list[str], stderr: list[str])

Bases: ProcessErrorBase

Process Error.

rc: int: Return code.

class pytest_mh.conn.ProcessErrorType

Generic process error type. Must be a subclass of ProcessError.

alias of TypeVar(‘ProcessErrorType’, bound=ProcessError)

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

Bases: ProcessErrorBase

Process Timeout Error.

timeout: int: Number of seconds required for time out.

class pytest_mh.conn.ProcessInputBuffer

Bases: ABC

Process’ input buffer.

Allows to write into stdin of the process.

abstractmethod write(data: str | bytes) → None

Write data to stdin.

Parameters:: data (str | bytes) – Data to write.

class pytest_mh.conn.ProcessInputBufferType

Generic process input buffer type. Must be a subclass of ProcessInputBuffer.

alias of TypeVar(‘ProcessInputBufferType’, bound=ProcessInputBuffer)

class pytest_mh.conn.ProcessLogLevel(value)

Bases: Enum

Process 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.conn.ProcessResult(rc: int, stdout: list[str], stderr: list[str], error: ProcessErrorType)

Bases: Generic[ProcessErrorType]

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().

rc: int: Return code.

stdout: str: Standard output.

stderr: str: Standard error output.

stdout_lines: list[str]: Standard output, line by line.

stderr_lines: list[str]: Standard error output, line by line.

error: ProcessErrorType: Process error object that can be raised manually with throw().

throw() → NoReturn

Raise ProcessErrorType for this command manually.

The error is available to raise even if return code is 0. Therefore the caller may choose to raise the error on any condition, not just the return code.

Raises:

ValueError – If no error is set.
ProcessErrorType – Error generated for this process result.

class pytest_mh.conn.ProcessResultType

Generic process result type. Must be a subclass of ProcessResult.

alias of TypeVar(‘ProcessResultType’, bound=ProcessResult)

class pytest_mh.conn.ProcessType

Generic process type. Must be a subclass of Process.

alias of TypeVar(‘ProcessType’, bound=Process)

class pytest_mh.conn.Shell(name: str, shell_command: str)

Bases: ABC

Multihost shell abstraction.

Parameters:

name (str) – Shell name.
shell_command – Shell command that will execute user scripts.
shell_command – str

name: str: Shell name.

shell_command: str: Shell command that will execute user scripts.

abstractmethod build_command_line(script: str, *, cwd: str | None, env: dict[str, Any]) → str

Create a complete command that will execute given script in this shell.

User is able to specify a different working directory and additional environment variables. This method adds this information to the user’s command and returns a full script that will switch the working directory, sets environment variables and then executes the given command or script.

Parameters:

script (str) – Script
cwd (str | None) – Working directory, None means no change.
env (dict[str, Any]) – Additional environment variables.

Returns:

Complete command line to run to execute the script in this shell.

Return type:

str