tbot.machine Module

class tbot.machine.Machine[source]

Bases: contextlib.AbstractContextManager, typing.Generic

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

Connect to this machine.

name

Name of this machine.

lh

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

destroy() → None[source]

Destroy and cleanup this machine.

class tbot.machine.InteractiveMachine[source]

Bases: abc.ABC

Machine that can be used interactively.

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

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

fileno() → int[source]

Return the filedescriptor of this channel.

Return type:int
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.