tbot.machine Module

Warning

This is the documentation for an old version of tbot. Please head over to

for the latest docs!

class tbot.machine.Machine[source]

Bases: contextlib.AbstractContextManager, typing.Generic

Connect to a machine (host, board, etc.).

Connect to this machine.

abstract property name

Name of this machine.

abstract property lh

Return the LabHost that was used to establish the connection for this machine.

abstract destroy() → None[source]

Destroy and cleanup this machine.

class tbot.machine.InteractiveMachine[source]

Bases: abc.ABC

Machine that can be used interactively.

abstract interactive() → None[source]

Drop into an interactive shell on this machine.

Raises

RuntimeError – If tbot was not able to reacquire the shell after the session finished.

exception tbot.machine.CommandFailedException(host: tbot.machine.machine.Machine, command: str, stdout: Optional[str], *args: Any)[source]

Bases: Exception

Command that was issued did not finish successfully.

Create a new CommandFailedException.

Parameters

  • host (machine.Machine) – The machine the command was issued on

  • command (str) – The command that failed

  • stdout (str) – Optional output of the command

exception tbot.machine.WrongHostException(host: tbot.machine.machine.Machine, arg: Any)[source]

Bases: Exception

The host associated with an object did not match.

Create a new WrongHostException.

Parameters

  • host (machine.Machine) – The wrong host

  • arg – The object that is not associated with host

tbot.machine.channel

class tbot.machine.channel.Channel[source]

Bases: abc.ABC

Generic channel.

Create a new channel.

abstract send(data: Union[bytes, str]) → None[source]

Send some data to this channel.

Parameters

str data (bytes,) – Data to be sent. It data is a str it will be encoded using utf-8.

Raises

ChannelClosedException – If the cannel is no longer open.

abstract recv(timeout: Optional[float] = None, max: Optional[int] = None) → bytes[source]

Receive some data from this channel.

recv will block until at least one byte is available or the timeout is reached.

Warning

recv does in no way attempt to ensure that bytes can be decoded as unicode. It is very well possible that recv returns after half a unicode sequence. Code using recv needs to be robust in regards to this.

Because of this, recv should only be used if no other method of Channel suits your needs.

Parameters

  • timeout (float) – Optional timeout after which recv should return if no data is avalable.

  • max (int) – Optional maximum number of bytes to read.

Raises
abstract close() → None[source]

Close this channel.

An Implementation of close must call Channel.cleanup before closing the channel if it is still open.

Calls to send/recv must fail after calling close.

abstract fileno() → int[source]

Return the filedescriptor of this channel.

Return type

int

abstract isopen() → bool[source]

Return whether this channel is still open.

Warning

The result of this call should only be taken as a hint, as the remote end might unexpectedly close the channel shortly after the call.

Return type

bool

register_cleanup(clean: Callable[[tbot.machine.channel.channel.Channel], None]) → None[source]

Register a cleanup function for this channel.

attach_interactive(end_magic: Union[str, bytes, None] = None) → None[source]

Connect tbot’s terminal to this channel.

Allows the user to interact directly with whatever this channel is connected to.

Parameters

bytes end_magic (str,) – String that when read should end the interactive session.

initialize(*, sh: Type[tbot.machine.linux.shell.shell.Shell] = <class 'tbot.machine.linux.shell.bash.Bash'>) → None[source]

Initialize this channel so it is ready to receive commands.

Internally runs commands to set the prompt to a known value and disable history + line-editing.

Parameters

sh (tbot.machine.linux.shell.Shell) – Type of the Shell this channel is connected to.

recv_n(n: int, timeout: Optional[float] = None) → bytes[source]

Receive exactly N bytes.

Return exactly N bytes or raise an exception if the channel closed/the timeout was reached.

Parameters

  • n (int) – Number of bytes

  • timeout (float) – Optional timeout

read_until_prompt(prompt: str, *, regex: bool = False, stream: Optional[TextIO] = None, timeout: Optional[float] = None, must_end: bool = True) → str[source]

Read until receiving prompt.

Parameters

  • prompt (str) – The prompt to wait for. If regex is True, this will be interpreted as a regular expression.

  • regex (bool) – Whether the prompt should be interpreted as is or as a regular expression. In the latter case, a '$' will be added to the end of the expression.

  • stream (io.TextIOBase) – Optional stream where read_until_prompt should write everything received up until the prompt is detected.

  • timeout (float) – Optional timeout.

  • must_end (bool) – Whether the prompt has to appear at the end of the stream.

Raises

TimeoutError – If a timeout is set and this timeout is reached before the prompt is detected.

Return type

str

Returns

Everything read up until the prompt.

raw_command(command: str, *, prompt: str = 'TBOT-VEJPVC1QUk9NUFQK$ ', stream: Optional[TextIO] = None, timeout: Optional[float] = None) → str[source]

Send a command to this channel and wait until it finishes.

Parameters

  • command (str) – The command without a trailing newline

  • prompt (str) – The prompt of the shell on this channel. This is needed to detect when the command is done.

  • stream (io.TextIOBase) – Optional stream where the commands output should be written.

  • timeout (float) – Optional timeout.

Raises

TimeoutError – If the timeout was reached before the command finished.

Return type

str

Returns

The ouput of the command. Will contain a trailing newline unless the command did not send one (eg. printf)

raw_command_with_retval(command: str, *, prompt: str = 'TBOT-VEJPVC1QUk9NUFQK$ ', retval_check_cmd: str = 'echo $?', stream: Optional[TextIO] = None, timeout: Optional[float] = None) → Tuple[int, str][source]

Send a command to this channel, wait until it finishes, and check its retcode.

Parameters

  • command (str) – The command without a trailing newline

  • prompt (str) – The prompt of the shell on this channel. This is needed to detect when the command is done.

  • retval_check_cmd (str) – Command used to check the exit code.

  • stream (io.TextIOBase) – Optional stream where the commands output should be written.

  • timeout (float) – Optional timeout.

Raises

TimeoutError – If the timeout was reached before the command finished.

Return type

tuple[int, str]

Returns

The retcode and ouput of the command. Will contain a trailing newline unless the command did not send one (eg. printf)

Implementations

class tbot.machine.channel.ParamikoChannel(ch: paramiko.channel.Channel)[source]

Bases: tbot.machine.channel.channel.Channel

Paramiko based channel.

Create a new tbot channel based on a Paramiko channel.

Parameters

ch (paramiko.Channel) – Paramiko Channel

class tbot.machine.channel.SubprocessChannel[source]

Bases: tbot.machine.channel.channel.Channel

Subprocess based channel.

Create a new subprocess based channel.

Helpers

exception tbot.machine.channel.ChannelClosedException[source]

Bases: Exception

Exception when attempting to interact with a closed channel.

class tbot.machine.channel.channel.SkipStream(stream: TextIO, n: int)[source]

Bases: _io.StringIO

Stream wrapper that skips a few character at the start.

Create a new SkipStream that skips n chars.

Parameters

  • stream (io.TextIOBase) – The underlying stream. The first n chars written to this SkipStream will not be written to stream.

  • n (int) – Number of characters to skip.

write(s: str) → int[source]

Write some string to this stream.