pytestshellutils package#

Submodules#

pytestshellutils.customtypes module#

Custom Types.

class pytestshellutils.customtypes.EnvironDict[source]#

Bases: Dict[str, str]

Environ dictionary type.

__str__() → str[source]#: String representation of the class.

class pytestshellutils.customtypes.GenericCallback(*args, **kwargs)[source]#

Bases: Protocol

Generic callback function.

__call__(*args: Any, **kwargs: Any) → None[source]#: Call the generic callback.

class pytestshellutils.customtypes.DaemonCallback(*args, **kwargs)[source]#

Bases: Protocol

Daemon callback function.

__call__(daemon: Daemon) → None[source]#: Call the daemon callback.

class pytestshellutils.customtypes.Callback(*, func: Callable[[...], Any], args: Tuple[Any, ...] | None = None, kwargs: Dict[str, Any] | None = None)[source]#

Bases: object

Class which “stores” information of a callback.

func: Callable[[...], Any]#

args: Tuple[Any, ...]#

kwargs: Dict[str, Any]#

__str__() → str[source]#: String representation of the class.

__call__(*args: Any, **kwargs: Any) → Any[source]#: Call the callback.

pytestshellutils.exceptions module#

Pytest Shell Utilities related exceptions.

exception pytestshellutils.exceptions.ShellUtilsException[source]#

Bases: Exception

Base pytest shell utilities exception.

exception pytestshellutils.exceptions.CallbackException[source]#

Bases: ShellUtilsException

Exception raised during a before/after start/stop daemon callback.

exception pytestshellutils.exceptions.ProcessFailed(message: str, process_result: ProcessResult | None = None)[source]#

Bases: ShellUtilsException

Exception raised when a sub-process fails.

Parameters:: message – The exception message
Keyword Arguments:: process_result – The ProcessResult instance when the exception occurred

__str__() → str[source]#: Return a printable representation of the exception.

exception pytestshellutils.exceptions.FactoryFailure(message: str, process_result: ProcessResult | None = None)[source]#

Bases: ProcessFailed

Exception raised when a sub-process fails on one of the factories.

exception pytestshellutils.exceptions.FactoryNotStarted(message: str, process_result: ProcessResult | None = None)[source]#

Bases: FactoryFailure

Exception raised when a factory failed to start.

Please look at FactoryFailure for the supported keyword arguments documentation.

exception pytestshellutils.exceptions.FactoryNotRunning(message: str, process_result: ProcessResult | None = None)[source]#

Bases: FactoryFailure

Exception raised when trying to use a factory’s .stopped context manager and the factory is not running.

Please look at FactoryFailure for the supported keyword arguments documentation.

exception pytestshellutils.exceptions.ProcessNotStarted(message: str, process_result: ProcessResult | None = None)[source]#

Bases: FactoryFailure

Exception raised when a process failed to start.

Please look at FactoryFailure for the supported keywords. arguments documentation.

exception pytestshellutils.exceptions.FactoryTimeout(message: str, process_result: ProcessResult | None = None)[source]#

Bases: FactoryNotStarted

Exception raised when a process timed-out.

Please look at FactoryFailure for the supported keywords. arguments documentation.

pytestshellutils.plugin module#

Pytest shell utilities plugin.

pytestshellutils.plugin.shell() → Subprocess[source]#

Shell fixture.

Example

def test_assert_good_exitcode(shell):

    ret = shell.run("exit", "0")
    assert ret.returncode == 0


def test_assert_bad_exitcode(shell):

    ret = shell.run("exit", "1")
    assert ret.returncode == 1

pytestshellutils.shell module#

Shelling class implementations.

class pytestshellutils.shell.BaseFactory(*, cwd: str | Path = _Nothing.NOTHING, environ: EnvironDict = _Nothing.NOTHING)[source]#

Bases: object

Base factory class.

Keyword Arguments:

cwd – The path to the desired working directory
environ – A dictionary of key, value pairs to add to the environment.

cwd: Path#

environ: EnvironDict#

class pytestshellutils.shell.SubprocessImpl(*, factory: Factory | Subprocess | ScriptSubprocess)[source]#

Bases: object

Subprocess interaction implementation.

Parameters:: factory – The factory instance, either Subprocess or a sub-class of it.

factory: Factory | Subprocess | ScriptSubprocess#

cmdline(*args: str, **kwargs: Any) → List[str][source]#

Construct a list of arguments to use when starting the subprocess.

Parameters:: args – Additional arguments to use when starting the subprocess

By default, this method will just call it’s factory’s cmdline() method, but can be overridden.

init_terminal(cmdline: List[str], shell: bool = False, env: EnvironDict | None = None, cwd: str | Path | None = None) → Popen[Any][source]#

Instantiate a terminal with the passed command line(cmdline) and return it.

Additionally, it sets a reference to it in self._terminal and also collects an initial listing of child processes which will be used when terminating the terminal

Parameters:

cmdline – List of strings to pass as args to Popen

Keyword Arguments:

shell – Pass the value of shell to Popen
env – A dictionary of key, value pairs to add to the pytestshellutils.shell.Factory.environ.
cwd – A path for the CWD when running the process.

Returns:

A Popen instance.

is_running() → bool[source]#

Returns true if the sub-process is alive.

Returns:: Returns true if the sub-process is alive

terminate() → ProcessResult[source]#: Terminate the started subprocess.

property pid: int | None#: The pid of the running process. None if not running.

run(*args: str, shell: bool = False, env: EnvironDict | None = None, cwd: str | Path | None = None, **kwargs: Any) → Popen[Any][source]#

Run the given command synchronously.

Parameters:

args – The command to run.

Keyword Arguments:

shell – Pass the value of shell to pytestshellutils.shell.Factory.init_terminal()
env – A dictionary of key, value pairs to add to the pytestshellutils.shell.Factory.environ.
cwd – A path for the CWD when running the process.

Returns:

A Popen instance.

class pytestshellutils.shell.Factory(*, cwd: str | Path = _Nothing.NOTHING, environ: EnvironDict = _Nothing.NOTHING, slow_stop: bool = True, system_encoding: str = _Nothing.NOTHING, timeout: int | float = _Nothing.NOTHING)[source]#

Bases: BaseFactory

Base shell factory class.

Keyword Arguments:

slow_stop – Whether to terminate the processes by sending a SIGTERM signal or by calling terminate() on the sub-process. When code coverage is enabled, one will want slow_stop set to True so that coverage data can be written down to disk.
system_encoding – The system encoding to use when decoding the subprocess output. Defaults to “utf-8”.
timeout – The default maximum amount of seconds that a script should run. This value can be overridden when calling run() through the _timeout keyword argument, and, in that case, the timeout value applied would be that of _timeout instead of self.timeout.

slow_stop: bool#

system_encoding: str#

timeout: int | float#

impl: SubprocessImpl#

__attrs_post_init__() → None[source]#: Post attrs class initialization routines.

cmdline(*args: str) → List[str][source]#: Method to construct a command line.

get_display_name() → str[source]#: Returns a human readable name for the factory.

is_running() → bool[source]#: Returns true if the sub-process is alive.

terminate() → ProcessResult[source]#: Terminate the started subprocess.

property pid: int | None#: The pid of the running process. None if not running.

__ne__(other)#: Method generated by attrs for class Factory.

class pytestshellutils.shell.Subprocess(*, cwd: str | Path = _Nothing.NOTHING, environ: EnvironDict = _Nothing.NOTHING, slow_stop: bool = True, system_encoding: str = _Nothing.NOTHING, timeout: int | float = _Nothing.NOTHING)[source]#

Bases: Factory

Base shell factory class.

run(*args: str, env: EnvironDict | None = None, _timeout: int | float | None = None, **kwargs: Any) → ProcessResult[source]#

Run the given command synchronously.

Keyword Arguments:

args – The list of arguments to pass to cmdline() to construct the command to run
env – Pass a dictionary of environment key, value pairs to inject into the subprocess.
_timeout – The timeout value for this particular run() call. If this value is not None, it will be used instead of timeout, the default timeout.

process_output(stdout: str, stderr: str, cmdline: List[str] | None = None) → Tuple[str, str, Dict[Any, Any] | None][source]#

Process the output. When possible JSON is loaded from the output.

Returns:: Returns a tuple in the form of (stdout, stderr, loaded_json)

class pytestshellutils.shell.ScriptSubprocess(*, cwd: str | Path = _Nothing.NOTHING, environ: EnvironDict = _Nothing.NOTHING, slow_stop: bool = True, system_encoding: str = _Nothing.NOTHING, timeout: int | float = _Nothing.NOTHING, script_name: str, base_script_args: List[str] = _Nothing.NOTHING)[source]#

Bases: Subprocess

Base CLI script/binary class.

Keyword Arguments:

script_name – This is the string containing the name of the binary to call on the subprocess, either the full path to it, or the basename. In case of the basename, the directory containing the basename must be in your $PATH variable.
base_script_args – An list or tuple iterable of the base arguments to use when building the command line to launch the process

Please look at Factory for the additional supported keyword arguments documentation.

script_name: str#

base_script_args: List[str]#

get_display_name() → str[source]#: Returns a human readable name for the factory.

get_script_path() → str[source]#: Returns the path to the script to run.

get_base_script_args() → List[str][source]#: Returns any additional arguments to pass to the CLI script.

get_script_args() → List[str][source]#: Returns any additional arguments to pass to the CLI script.

cmdline(*args: str) → List[str][source]#

Construct a list of arguments to use when starting the subprocess.

Parameters:: args – Additional arguments to use when starting the subprocess

class pytestshellutils.shell.StartDaemonCallArguments(*, args: Tuple[str, ...], kwargs: Dict[str, Any])[source]#

Bases: object

This class holds the arguments and keyword arguments used to start a daemon.

It’s used when restarting the daemon so that the same call is used.

Keyword Arguments:

args – List of arguments
kwargs – Dictionary of keyword arguments

args: Tuple[str, ...]#

kwargs: Dict[str, Any]#

class pytestshellutils.shell.DaemonImpl(*, factory: Daemon, before_start_callbacks: List[Callback] = _Nothing.NOTHING, after_start_callbacks: List[Callback] = _Nothing.NOTHING, before_terminate_callbacks: List[Callback] = _Nothing.NOTHING, after_terminate_callbacks: List[Callback] = _Nothing.NOTHING)[source]#

Bases: SubprocessImpl

Daemon subprocess interaction implementation.

Please look at SubprocessImpl for the additional supported keyword arguments documentation.

factory: Daemon#

before_start(callback: Callable[[], None], *args: Any, **kwargs: Any) → None[source]#

Parameters:

callback – The function to call back

Keyword Arguments:

args – The arguments to pass to the callback
kwargs – The keyword arguments to pass to the callback

Returns:

Nothing.

after_start(callback: Callable[[], None], *args: Any, **kwargs: Any) → None[source]#

Parameters:

callback – The function to call back

Keyword Arguments:

args – The arguments to pass to the callback
kwargs – The keyword arguments to pass to the callback

Returns:

Nothing.

before_terminate(callback: Callable[[], None], *args: Any, **kwargs: Any) → None[source]#

Parameters:

callback – The function to call back

Keyword Arguments:

args – The arguments to pass to the callback
kwargs – The keyword arguments to pass to the callback

Returns:

Nothing.

after_terminate(callback: Callable[[], None], *args: Any, **kwargs: Any) → None[source]#

Parameters:

callback – The function to call back

Keyword Arguments:

args – The arguments to pass to the callback
kwargs – The keyword arguments to pass to the callback

Returns:

Nothing.

start(*extra_cli_arguments: str, max_start_attempts: int | None = None, start_timeout: int | float | None = None) → bool[source]#

Start the daemon.

Keyword Arguments:

extra_cli_arguments – Extra arguments to pass to the CLI that starts the daemon
max_start_attempts – Maximum number of attempts to try and start the daemon in case of failures
start_timeout – The maximum number of seconds to wait before considering that the daemon did not start

Returns:

A boolean indicating if the start was successful or not.

Return type:

bool

terminate() → ProcessResult[source]#: Terminate the daemon.

get_start_arguments() → StartDaemonCallArguments[source]#: Return the arguments and keyword arguments used when starting the daemon.

class pytestshellutils.shell.Daemon(*, cwd: str | Path = _Nothing.NOTHING, environ: EnvironDict = _Nothing.NOTHING, slow_stop: bool = True, system_encoding: str = _Nothing.NOTHING, timeout: int | float = _Nothing.NOTHING, script_name: str, base_script_args: List[str] = _Nothing.NOTHING, check_ports: List[int] = _Nothing.NOTHING, stats_processes: StatsProcesses = None, start_timeout: int | float, max_start_attempts: int = 3, extra_cli_arguments_after_first_start_failure: List[str] = _Nothing.NOTHING, start_checks_callbacks: List[Callback] = _Nothing.NOTHING)[source]#

Bases: ScriptSubprocess

Base daemon factory.

Keyword Arguments:

check_ports – List of ports to try and connect to while confirming that the daemon is up and running
extra_cli_arguments_after_first_start_failure – Extra arguments to pass to the CLI that starts the daemon after the first failure
max_start_attempts – Maximum number of attempts to try and start the daemon in case of failures
start_timeout – The maximum number of seconds to wait before considering that the daemon did not start

Please look at Subprocess for the additional supported keyword arguments documentation.

impl: DaemonImpl#

script_name: str#

base_script_args: List[str]#

check_ports: List[int]#

stats_processes: StatsProcesses#

start_timeout: int | float#

max_start_attempts: int#

extra_cli_arguments_after_first_start_failure: List[str]#

listen_ports: List[int]#

__attrs_post_init__() → None[source]#: Post attrs class initialization routines.

before_start(callback: Callable[[], None], *args: Any, **kwargs: Any) → None[source]#

Parameters:

callback – The function to call back

Keyword Arguments:

args – The arguments to pass to the callback
kwargs – The keyword arguments to pass to the callback

Returns:

Nothing.

after_start(callback: Callable[[], None], *args: Any, **kwargs: Any) → None[source]#

Parameters:

callback – The function to call back

Keyword Arguments:

args – The arguments to pass to the callback
kwargs – The keyword arguments to pass to the callback

Returns:

Nothing.

before_terminate(callback: Callable[[], None], *args: Any, **kwargs: Any) → None[source]#

Parameters:

callback – The function to call back

Keyword Arguments:

args – The arguments to pass to the callback
kwargs – The keyword arguments to pass to the callback

Returns:

Nothing.

after_terminate(callback: Callable[[], None], *args: Any, **kwargs: Any) → None[source]#

Parameters:

callback – The function to call back

Keyword Arguments:

args – The arguments to pass to the callback
kwargs – The keyword arguments to pass to the callback

Returns:

Nothing.

start_check(callback: Callable[[...], bool], *args: Any, **kwargs: Any) → None[source]#

The callback must accept as the first argument timeout_at which is a float. The callback must stop trying to confirm running behavior once time.time() > timeout_at. The callback should return True to confirm that the daemon is ready for work.

Parameters:

callback – The function to call back

Keyword Arguments:

args – The arguments to pass to the callback
kwargs – The keyword arguments to pass to the callback

Returns:

Nothing.

Example

def check_running_state(timeout_at: float) -> bool:
    while time.time() <= timeout_at:
        # run some checks
        ...
        # if all is good
        break
    else:
        return False
    return True

get_check_ports() → List[int][source]#: Return a list of ports to check against to ensure the daemon is running.

get_start_check_callbacks() → List[Callback][source]#: Return a list of the start check callbacks.

start(*extra_cli_arguments: str, max_start_attempts: int | None = None, start_timeout: int | float | None = None) → bool[source]#: Start the daemon.

started(*extra_cli_arguments: str, max_start_attempts: int | None = None, start_timeout: int | float | None = None) → Generator[Daemon, None, None][source]#: Start the daemon and return it’s instance so it can be used as a context manager.

stopped(before_stop_callback: Callable[[Daemon], None] | None = None, after_stop_callback: Callable[[Daemon], None] | None = None, before_start_callback: Callable[[Daemon], None] | None = None, after_start_callback: Callable[[Daemon], None] | None = None) → Generator[Daemon, None, None][source]#

Stop the daemon and return it’s instance so it can be used as a context manager.

Keyword Arguments:

before_stop_callback – A callable to run before stopping the daemon. The callback must accept one argument, the daemon instance.
after_stop_callback – A callable to run after stopping the daemon. The callback must accept one argument, the daemon instance.
before_start_callback – A callable to run before starting the daemon. The callback must accept one argument, the daemon instance.
after_start_callback – A callable to run after starting the daemon. The callback must accept one argument, the daemon instance.

This context manager will stop the factory while the context is in place, it re-starts it once out of context.

Example

assert factory.is_running() is True

with factory.stopped():
    assert factory.is_running() is False

assert factory.is_running() is True

run_start_checks(started_at: float, timeout_at: float) → bool[source]#: Run checks to confirm that the daemon has started.

__enter__() → Daemon[source]#: Use class as a context manager.

__exit__(*_: Any) → None[source]#: Exit the class context manager.

pytestshellutils package#

Subpackages#

Submodules#

pytestshellutils.customtypes module#

pytestshellutils.exceptions module#

pytestshellutils.plugin module#

pytestshellutils.shell module#

pytestshellutils.version module#