InteractiveSession¶

This class is intended to be a wrapper around pexpect. The reason for implementing such a wrapper is to avoid having several testcases repeating the same sequences of calls to be pexpect.sendline() / pexpect.spawn() / pexpect.read() e.t.c. Furthermore the intention is to protect users from sequences of calls to these methods that could lead to erroneous results (for example executing two successive commands and not being able to properly distinguish/split the results printed to the terminal from each one).

Major functionality provided by the class is the ability to execute commands in a pseudo-terminals and getting back any bytes sent to it from the command.

This requires sending the command to the terminal, consuming any ‘waste’ printed on it and waiting for the shell prompt in order to be able to execute the next command. Since its type of shell (like bash) provides its own prompt, in order to keep track of prompt changes (for example invoking python shell from bash shell) a stack of objects is used (top object represents the currently active shell).

Each such object in the stack describes allowed actions and way of working of the currently active shell.

Recommended reading: basic knowledge of pexpect would help in understanding and using this library properly.

Bash Shell¶

This example describes how to start a bash shell and execute some commands:

s = InteractiveSession()
s.spawn(BashShell())
print s.current_shell().exec_command("ls -l")
print s.current_shell().exec_command("echo Hello")
print s.current_shell().exec_command("ls -l /some/non/existing/dir")
if s.current_shell().get_status_code() != 0:
    print "Command failed!!"

SSH Connection¶

This example describes how to start an SSH connection and execute command on remote machine:

s = InteractiveSession()
s.spwan(SshShell("1.2.3.4", "username", "password"))
print s.current_shell().exec_command("ls -l")
print s.current_shell().get_status_code()

Stacking of Shells¶

This example describes how stacking of different shells work:

s = InteractiveSession()
s.spawn(BashShell())
print s.current_shell().exec_command("ls -l")
s.push(SshShell("1.2.3.4", "username", "password"))
s.push(PythonShell())
print s.current_shell().exec_command(
    "print 'I am python on the remote host...'")
s.pop()
print s._current_shell().exec_command(
    "echo 'I am bash on the remote host...'")
s.pop()
print s._current_shell().exec_command("echo 'I am back...'")

User Input¶

Executing commands that require user input:

s = InteractiveSession()
s.spawn(BashShell())
s.current_shell().exec_command("rm my_important_file",
                               confirmation_msg="Are you sure (Y/n):",
                               confirmation_rsp="y')
s.current_shell().exec_command("rm my_other_file1 my_other_file2",
                               confirmation_msg=[
                               "Are you sure (Y/n):",
                               "Are you sure (Y/n):"], ["y", "y"])

class crl.interactivesessions.InteractiveSession.InteractiveSession(dump_received=None, dump_outgoing=None)¶

Bases: object

InteractiveSession is a wrapper class of pexpect.

Two arguments for storing the sent and the received data to files are provided. By default, nothing is stored. If dump_received is given, then all the bytes received from the pseudo-terminal are stored to the file with full path dump_received. Similarly dump_outgoing is a path to a file to where all the sent data is stored.

__init__(dump_received=None, dump_outgoing=None)¶: x.__init__(…) initializes x; see help(type(x)) for signature

current_shell()¶: Return the currently active shell or raise exception if there is no shell

get_parent_shell()¶: Returns the shell below currently active shell

pop()¶

Should be called in order to terminate currently active shell and return to the previous one.

Note

The originally spawned shell is not part of the shell stack.

Example:

s = InteractiveSession()
s.spawn(BashShell())
s.push(SshShell("1.2.3.4", "username", "password"))
s.current_shell().exec_command("remote host command here")
s.pop()
s.current_shell().exec_command("local host command here")

pop_until(shell)¶

Pop shells from the stack, until the received prompt matches the one of ‘shell’.

Should be used to ensure the shell stack is in a known state (e.g. at the beginning of each test case reusing the same InteractiveSession.)

push(shell, timeout=30)¶

Should be called for any action that would change the command prompt. Failing to keep this rule will result in unpredictable results from Shell.exec_command set the last shell’s prompt back in case of ShellStartError (hostname is wrong in the started shell)

Args:: shell: any class implementing the Shell interface

Example:

s = InteractiveSession()
s.spawn(BashShell()
s.push(SshShell("1.2.3.4", "username", "password"))

spawn(shell)¶

Thin wrapper around pexpect.spawn. This method is the first one that should be called in order to create a new terminal.

Args:: shell: any class implementing the Shell interface

Example:

s = InteractiveSession()
s.spawn(BashShell())

exception crl.interactivesessions.InteractiveSession.UnexpectedPrompt(received, expected)¶

Bases: crl.interactivesessions.interactivesessionexceptions.InteractiveSessionError

Prompt received from terminal does not match with expected prompt.

__init__(received, expected)¶: x.__init__(…) initializes x; see help(type(x)) for signature

exception crl.interactivesessions.InteractiveSession.UnknownShellState¶

Bases: crl.interactivesessions.interactivesessionexceptions.InteractiveSessionError

Raised when pop_until() results in an unexpected shell stack state.

InteractiveSession Shells¶

Shells serve as front-ends for the pseudo terminal by providing initialization and finalization methods of a new shell in the pseudo terminal as well as the pseudo terminal shell specific functionality.

In this section available shells as well as the base classes for them are documented.

Shell Abstract Bases¶

AutoCompletableShell¶

class crl.interactivesessions.shells.autocompletableshell.AutoCompletableShell(tty_echo=False)¶

Bases: crl.interactivesessions.shells.shell.Shell

Several shells support tab autocompletion. Any custom shell implementation can inherit from this class in order to provide such functionality.

Args:

cmd: part of a command.

autocompletion_trigger: characters that the user would press to trigger autocompletion (for example '\t' for bash)

timeout: how much time to wait (in seconds) for the shell prompt to reappear after the tab completion

Autocompletion works in different ways for different shells. Furthermore in some cases the autocompleted content just appears in the same line that the user was typing the command while in other cases it is printed below this line followed by a new prompt. The default implementation that can be reused by Shell derivatives works by expecting the new prompt for a finite amount of time. If the prompt does not appear then the implementation assumes that autocompletion occured on the same line and returns as a result any bytes printed to the terminal until the timeout occured.

Shell¶

class crl.interactivesessions.shells.shell.Shell(tty_echo=False)¶

This is the basic interface that any new shell should implement.

exec_command(cmd, timeout=-1)¶

Execute a command and return and text printed to the terminal.

Args:

cmd: string containing the command to be executed

timeout: how many seconds to wait for command completion

Returns:

The terminal output of the command execution.

exec_command_expecting_more(cmd, more_prompt='--More--')¶: This method is a variation of exec_command() that can be used when the command generates paged output. The method returns a tuple (more_counter, output), where more_counter is an integer representing the number of occurences of the more_prompt and output contains any text directed to the terminal (including more_prompt).

exec_prompting_command(cmd, responses, timeout=-1)¶

Execute a command that prompts for additional input.

responses should be a list of tuples, each tuple specifying the prompt(s) expected, and the response to be sent.

Args:

cmd(str): Command to be executed.

responses(list of tuples): Expected prompts.

timeout(number): Timeout for receiving each prompt.

Returns:

Terminal output.

exit()¶: Close this shell.

get_prompt()¶: Get the expected prompt for this shell.

get_prompt_from_terminal(empty_command='', timeout=-1)¶: Send terminal an empty command which should cause terminal to send back the prompt. First, expected prompt is waited for timeout. If succesful, the expected prompt is returned but if timeout occurs, then output read so far is returned.

get_start_cmd()¶

Return the command to be run in order to start this shell.

This method will be called by InteractiveSession.spawn() and InteractiveSession.push().

get_status_code(timeout=5)¶

Abstract method. Implementation must return the status of the last command executed.

Returned value should be an integer indicating success/failure of the last command executed.

The default timeout DEFAULT_STATUS_TIMEOUT is globally adjustable:

>>> from crl.interactivesessions.shells.shell import (
...     DefaultStatusTimeout,
...     DEFAULT_STATUS_TIMEOUT,
...     Shell)
>>>
>>> class ExampleShell(Shell):
...    def get_start_cmd(self):
...        pass
...    def start(self):
...        pass
...    def exit(self):
...        pass
...    def get_status_code(self, timeout=DEFAULT_STATUS_TIMEOUT):
...        return float(timeout)
>>>
>>> DefaultStatusTimeout.set(20)
>>> ExampleShell().get_status_code()
20.0

is_terminal_prompt_matching(terminal_prompt)¶

Return True if and only if prompt is matching with the terminal output.

Args:

terminal_prompt(str): expected terminal prompt

start()¶

Initialize the new shell.

This method will be called after running the command provided by get_start_cmd() and provides ability for the new shell to initialize itself.

exception crl.interactivesessions.shells.shell.TimeoutError(output)¶: Raised when remote command execution times out.

Built-In Shells¶

BashShell¶

exception crl.interactivesessions.shells.bashshell.BashError¶

Bases: crl.interactivesessions.interactivesessionexceptions.InteractiveSessionError

Raised by BashShell errors

class crl.interactivesessions.shells.bashshell.BashShell(cmd='bash', confirmation_msg=None, confirmation_rsp='Y', tty_echo=False, set_rand_promt=True, init_env=None, workdir=None, **kwargs)¶

Bases: crl.interactivesessions.shells.autocompletableshell.AutoCompletableShell

InteractiveSession Shell interface for bash.

Args:

cmd (str): command to start bash shell, default: bash

confirmation_msg (str): string to expect for confirmation to start bash shell

confirmation_rsp (str): expected response to confirmation

tty_echo (bool): terminal echo value to be set when started in spawn

init_env (str): path to the file to be sourced in init or: bash command to be executed if starts with ‘content:’. example: ‘content: alias python=python3’ default: None

workdir (bool): change to this directory in start

banner_timeout (float): timeout in seconds between lines in the start banner, default 0.1

exit()¶: Close this shell.

get_pid()¶: Returns the pid of the last crated process using ‘echo $!’

get_start_cmd()¶

Return the command to be run in order to start this shell.

This method will be called by InteractiveSession.spawn() and InteractiveSession.push().

get_status_code(timeout=5)¶

Returns the status code of the last executed command using ‘echo $?’

The default status timeout is globally adjustable. The setting

>>> from crl.interactivesessions.shells.shell import DefaultStatusTimeout
>>> DefaultStatusTimeout.set(20)

sets default status timeout to 20 seconds.

scp_copy_file(source_file, dest_ip, dest_user, dest_pass, dest_file)¶

Copy a file from localhost to a destination. This method is just a convenience wrapper around scp command line tool. Same results could be achieved using exec_command.

Args:

source_file: name of the local file to be transfered

dest_ip: ip of the host that should receive the file

dest_user: username for host login

dest_pass: password for host login

dest_file: where to put the file on the destination host

scp_download_file(source_file, source_ip, source_user, source_pass, dest_file)¶

Download a file from a remote machine to local. This method is just a convenience wrapper around scp command line tool. Same results could be achieved using exec_command.

Args:

source_file: name of the remote file to be transfered

source_ip: ip of the host that contains the file

source_user: username for host login

source_pass: password for host login

dest_file: where to put the file on the local machine

start()¶

Initialize the new shell.

This method will be called after running the command provided by get_start_cmd() and provides ability for the new shell to initialize itself.

terminate_process_forced(pid)¶: This method send signal -9 to a process forcing its termination.

Note

Consider using exec_command(“kill -9 my_pid”)

After executing kill -9 … in a bash shell, the shell itself prints output notifying the user that some process was killed. This output is not guaranteed to be printed before the prompt that appears once the user hits Enter and can confuse exec_command.

exception crl.interactivesessions.shells.bashshell.BashShellTypeError¶: Bases: exceptions.TypeError

exception crl.interactivesessions.shells.bashshell.CdToWorkdirFailed¶: Bases: crl.interactivesessions.interactivesessionexceptions.InteractiveSessionError

exception crl.interactivesessions.shells.bashshell.FailedToStartApplication¶: Bases: crl.interactivesessions.interactivesessionexceptions.InteractiveSessionError

exception crl.interactivesessions.shells.bashshell.FailedToTerminateProcess¶: Bases: crl.interactivesessions.interactivesessionexceptions.InteractiveSessionError

exception crl.interactivesessions.shells.bashshell.ScpError¶: Bases: crl.interactivesessions.interactivesessionexceptions.InteractiveSessionError

exception crl.interactivesessions.shells.bashshell.SourceInitFileFailed¶: Bases: crl.interactivesessions.interactivesessionexceptions.InteractiveSessionError

NamespaceShell¶

class crl.interactivesessions.shells.namespaceshell.NamespaceShell(namespace, *args, **kwargs)¶

Bases: crl.interactivesessions.shells.bashshell.BashShell

This class can be used in order to start a bash shell inside a given namespace

Example:

s = InteractiveSession()
s.spawn(SshShell("open_stack_controller", "username", "password"))
s.push(NamespaceShell("qdhcp-1234...")
s.push(SshShell("virtual_machine", "username", "password"))
s.current_shell().exec_command("ls -l")

get_start_cmd()¶

Return the command to be run in order to start this shell.

This method will be called by InteractiveSession.spawn() and InteractiveSession.push().

start()¶

Initialize the new shell.

This method will be called after running the command provided by get_start_cmd() and provides ability for the new shell to initialize itself.

PythonShell¶

exception crl.interactivesessions.shells.pythonshell.PythonRunNotStarted¶: Bases: crl.interactivesessions.interactivesessionexceptions.InteractiveSessionError

class crl.interactivesessions.shells.pythonshell.PythonShell(start_cmd='python', tty_echo=False, readline_init=None)¶

Bases: crl.interactivesessions.shells.pythonshellbase.PythonShellBase

The path to Python executable can be altered via start_cmd.

Args:

start_cmd: command to start Python shell (Default: python)

tty_echo: if True then commands are echoed in shell. (Default: False)

readline_init: If not None, then readline initialization commands in: readline_init string are executed in terminal in start. Please see syntax from readline man pages (Default: None).

start()¶

Initialize the new shell.

This method will be called after running the command provided by get_start_cmd() and provides ability for the new shell to initialize itself.

transfer_text_file(source_path, destination_dir='')¶: Transfer the text file source_path to current host of the shell

SshShell¶

exception crl.interactivesessions.shells.sshshell.SshError¶

Bases: crl.interactivesessions.interactivesessionexceptions.InteractiveSessionError

Raised when SshShell start fails

class crl.interactivesessions.shells.sshshell.SshShell(ip, username=None, password=None, tty_echo=False, second_password=None, port=None, init_env=None)¶

Bases: crl.interactivesessions.shells.bashshell.BashShell

This class can be used in order to start a remote bash shell. See also BashShell.

Args:

ip: IP address of the host

username: login username

password: login passowrd. If not given, passwordless login is expected.

tty_echo: If True, then terminal echo is set on.

second_password: if not None, this password is send to terminal after: the first password succesfully aplied.
port: If port is not None, alternate port is used in the connection: instead of the detfault 22.
init_env: Path to initialization file which is sourced after the all: other initialization is done.

For setting timeout for reading login banner, i.e. message-of-day, please use msgreader.MsgReader.set_timeout().

check_start_success()¶: This method is called right after the shell is pushed and the prompt is not set yet. To be implemented in derivative classes if needed Should raise ShellStartError if not successful.

exit()¶: Close this shell.

get_start_cmd()¶

Return the command to be run in order to start this shell.

This method will be called by InteractiveSession.spawn() and InteractiveSession.push().

SftpShell¶

exception crl.interactivesessions.shells.sftpshell.SftpConnectionError¶: Bases: crl.interactivesessions.interactivesessionexceptions.InteractiveSessionError

class crl.interactivesessions.shells.sftpshell.SftpShell(ip, username, password, cmd='sftp')¶

Bases: crl.interactivesessions.shells.shell.Shell

SFTP Shell

exit()¶: Close this shell.

get_prompt()¶: Get the expected prompt for this shell.

get_start_cmd()¶

Return the command to be run in order to start this shell.

This method will be called by InteractiveSession.spawn() and InteractiveSession.push().

get_status_code(timeout=5)¶

Abstract method. Implementation must return the status of the last command executed.

Returned value should be an integer indicating success/failure of the last command executed.

The default timeout DEFAULT_STATUS_TIMEOUT is globally adjustable:

>>> from crl.interactivesessions.shells.shell import (
...     DefaultStatusTimeout,
...     DEFAULT_STATUS_TIMEOUT,
...     Shell)
>>>
>>> class ExampleShell(Shell):
...    def get_start_cmd(self):
...        pass
...    def start(self):
...        pass
...    def exit(self):
...        pass
...    def get_status_code(self, timeout=DEFAULT_STATUS_TIMEOUT):
...        return float(timeout)
>>>
>>> DefaultStatusTimeout.set(20)
>>> ExampleShell().get_status_code()
20.0

start()¶

Initialize the new shell.

This method will be called after running the command provided by get_start_cmd() and provides ability for the new shell to initialize itself.

KeyAuthenticatedSshShell¶

class crl.interactivesessions.shells.keyauthenticatedsshshell.KeyAuthenticatedSshShell(host, initial_prompt, tty_echo=False)¶

Bases: crl.interactivesessions.shells.sshshell.SshShell

This class can be used in order to start a remote Ssh shell with keyauthentication instead of password authentication. It requires that ssh keys are set before pushing the shell. See also SshShell.

Args:

host: IP address or name of the host

initial_prompt: check_start_success method requires prompt,: status check command is sent when initial prompt is found, example:’$ ‘

tty_echo: If True, then terminal echo is set on.

check_start_success()¶

check status after pushing KeyAuthenticatedSshShell raise ShellStartError if status is not 0

Sending something after receiving status output is needed because sshshell _common_start keyword expects not empty buffer after check_start_success.

get_start_cmd()¶

Return the command to be run in order to start this shell.

This method will be called by InteractiveSession.spawn() and InteractiveSession.push().

wait_for_initial_prompt()¶: read until prompt is coming raise ShellStartError if prompt is not found

exception crl.interactivesessions.shells.keyauthenticatedsshshell.ShellStartError¶: Bases: crl.interactivesessions.interactivesessionexceptions.InteractiveSessionError

SudoShell¶

exception crl.interactivesessions.shells.sudoshell.SudoError¶

Bases: crl.interactivesessions.interactivesessionexceptions.InteractiveSessionError

Raised when SudoShell start fails

class crl.interactivesessions.shells.sudoshell.SudoShell(cmd='sudo bash', password=None)¶

Bases: crl.interactivesessions.shells.bashshell.BashShell

Running commands under sudo.

For setting timeout for reading login banner, i.e. lecture, please use msgreader.MsgReader.set_timeout().

get_start_cmd()¶

Return the command to be run in order to start this shell.

This method will be called by InteractiveSession.spawn() and InteractiveSession.push().

start()¶

Initialize the new shell.

This method will be called after running the command provided by get_start_cmd() and provides ability for the new shell to initialize itself.

Shell Tools¶

The shells based on Shell can be registered via Registering Shells. The registered shells can be used in the shell stack container described in Shell Stack Container.

Registering Shells¶

class crl.interactivesessions.shells.registershell.RegisterShell¶

Registration decorator for the shell.Shell based shells.

Usage example:

@RegisterShell()
class AliasBashShell(BashShell):
    pass

class crl.interactivesessions.shells.registershell.RegisteredShells¶

Dictionary for shell.Shell based shell classes registered via RegisterShell.

create_shell(shellname, **kwargs)¶: Create shell instance with class name shellname and with keyword arguments kwargs.

get_shellcls(shellclsname)¶: Return shell class for the name shellcls. Raises UnregisteredShell if not found.

exception crl.interactivesessions.shells.registershell.ShellAlreadyRegistered¶: Raised when the shell is already registerd with RegisterShell.

exception crl.interactivesessions.shells.registershell.UnregisteredShell¶: Raised when the shell name is not registered with RegisterShell.

Shell Stack Container¶

class crl.interactivesessions.shells.shellstack.DefaultSshShell(**kwargs)¶

Alias for InteractiveSession.SshShell which converts the __init__ arguments in the following fashion:

`DefaultSshShell` Argument	`sshshell.SshShell` Argument
host	ip
username or user	username

Other keyword arguments are passsed without any conversion to sshshell.SshShell.

class crl.interactivesessions.shells.shellstack.ShellStack¶

This class is a stack of the shell.Shell based shells.

initialize(shelldicts)¶: Initialize shells with list of dictionaries, shelldicts, where each dictionary is mapped to registered shell.Shell based shells in the order of the list. The dictionary can contain the class name (shellname) of the registered shell. The rest of the name-value pairs are passed as keyword arguments to the associated shell class. If no shellname is given, then DefaultSshShell is created from the dictionary.

shells¶: Shell stack created from shelldicts.

MsgReader¶

class crl.interactivesessions.shells.msgreader.MsgReader(read_until_end)¶

MsgReader is for reading login banner e.g. ‘lecture’ or ‘message-of-day’ messages from the terminal in shells.

Args:: read_until_end: shell.Shell._read_until_end()

classmethod get_timeout()¶: Get timeout.

read_until_end()¶: Reads message and returns it. Timeout _timeout is used for reading timeout.

classmethod reset_timeout()¶: Reset timeout to original default value.

classmethod set_timeout(timeout)¶

Set timeout.

Args:: timeout: timeout in seconds

InteractiveSession¶

Bash Shell¶

SSH Connection¶

Stacking of Shells¶

User Input¶

InteractiveSession Shells¶

Shell Abstract Bases¶

AutoCompletableShell¶

Shell¶

Built-In Shells¶

BashShell¶

NamespaceShell¶

PythonShell¶

SshShell¶

SftpShell¶

KeyAuthenticatedSshShell¶

SudoShell¶

Shell Tools¶

Registering Shells¶

Shell Stack Container¶

MsgReader¶

crl.interactivesessions

Navigation

Related Topics