pytest_mh.conn.container

Classes

ContainerClient(engine, container_name, *, user)

Interactive podman and docker client.

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

Container Process manager.

ContainerInputBuffer(pipe)

Container Input Buffer.

ContainerProcessResult(rc, stdout, stderr, error)

Container Process result.

Exceptions

ContainerConnectionError(engine, ...)

Unable to connect to the container.

ContainerProcessError(rc, id, command, cwd, ...)

Container Process Error.

ContainerProcessTimeoutError(timeout, id, ...)

Container Process Timeout Error.
class pytest_mh.conn.container.ContainerClient(engine: str, container_name: str, *, user: str, sudo: bool = False, sudo_password: str | None = None, shell: Shell, logger: MultihostLogger, timeout: int = 300)

Bases: Connection[ContainerProcess, ContainerProcessResult]

Interactive podman and docker client.

Parameters:

  • container_name (str) – Container name.

  • user (str) – Username that will be used to execute commands.

  • sudo (bool) – Run podman under root, defaults to False.

  • sudo_password (str | None) – SUDO password, defaults to None.

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

  • logger (MultihostLogger) – Multihost logger.

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

property connected: bool
Returns:

True if the connection is established, False otherwise.

Return type:

bool

connect() → None

Connect to the host.

Raises:

ContainerAuthenticationError – 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) → ContainerProcess

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.container.ContainerConnectionError(engine: str, container_name: str, *, user: str, sudo: bool)

Bases: ConnectionError

Unable to connect to the container.

class pytest_mh.conn.container.ContainerProcess(*, 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: ContainerClient)

Bases: Process[ContainerProcessResult, ContainerInputBuffer, ContainerProcessTimeoutError]

Container 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 (ContainerClient) – Container client.

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: ContainerInputBuffer

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.container.ContainerProcessError(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

Container Process Error.

exception pytest_mh.conn.container.ContainerProcessTimeoutError(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

Container Process Timeout Error.

class pytest_mh.conn.container.ContainerInputBuffer(pipe: IO[bytes])

Bases: ProcessInputBuffer

Container Input Buffer.

Allows to write into stdin of opened Container channel.

Parameters:

pipe (BufferedWriter) – Input pipe.

write(data: str | bytes) → None

Write data to stdin.

Parameters:

data (str | bytes) – Data to write.

flush() → None

Flush the input stream.

close() → None

Close the input stream.

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

Bases: ProcessResult[ContainerProcessError]

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