Skip to content
Back to top
View this page

# pytestshellutils package

## Subpackages

- pytestshellutils.utils package
  - `resolved_pathlib_path()`
  - `format_callback_to_string()`
  - `warn_until()`
  - Submodules
  - pytestshellutils.utils.ports module
    - `get_unused_localhost_port()`
    - `get_connectable_ports()`
  - pytestshellutils.utils.processes module
    - `MatchString`
      - `MatchString.matcher`
    - `convert_string_to_match_string()`
    - `ProcessResult`
      - `ProcessResult.returncode`
      - `ProcessResult.stdout`
      - `ProcessResult.stderr`
      - `ProcessResult.cmdline`
      - `ProcessResult.data_key`
      - `ProcessResult.data`
      - `ProcessResult.exitcode`
      - `ProcessResult.json`
      - `ProcessResult.__str__()`
    - `collect_child_processes()`
    - `terminate_process_list()`
    - `terminate_process()`
  - pytestshellutils.utils.socket module
  - pytestshellutils.utils.time module

## Submodules

## pytestshellutils.customtypes module

Custom Types.

class pytestshellutils.customtypes.EnvironDict

  Bases: `Dict`[`str`, `str`]

  Environ dictionary type.

  __str__() → str

    String representation of the class.

class pytestshellutils.customtypes.GenericCallback(*args, **kwargs)

  Bases: `Protocol`

  Generic callback function.

  __call__(*args: Any, **kwargs: Any) → None

    Call the generic callback.

class pytestshellutils.customtypes.DaemonCallback(*args, **kwargs)

  Bases: `Protocol`

  Daemon callback function.

  __call__(daemon: Daemon) → None

    Call the daemon callback.

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

  Bases: `object`

  Class which “stores” information of a callback.

  func: Callable[[...], Any]
  args: Tuple[Any, ...]
  kwargs: Dict[str, Any]
  __str__() → str

    String representation of the class.

  __call__(*args: Any, **kwargs: Any) → Any

    Call the callback.

## pytestshellutils.exceptions module

Pytest Shell Utilities related exceptions.

exception pytestshellutils.exceptions.ShellUtilsException

  Bases: `Exception`

  Base pytest shell utilities exception.

exception pytestshellutils.exceptions.CallbackException

  Bases: `ShellUtilsException`

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

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

  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

    Return a printable representation of the exception.

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

  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)

  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)

  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)

  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)

  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

  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, environ: EnvironDict = NOTHING)

  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)

  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]

    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]

    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

    Returns true if the sub-process is alive.

    Returns:

      Returns true if the sub-process is alive

  terminate() → ProcessResult

    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]

    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, environ: EnvironDict = NOTHING, slow_stop: bool = True, system_encoding: str = NOTHING, timeout: int | float = NOTHING)

  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

    Post `attrs` class initialization routines.

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

    Method to construct a command line.

  get_display_name() → str

    Returns a human readable name for the factory.

  is_running() → bool

    Returns true if the sub-process is alive.

  terminate() → ProcessResult

    Terminate the started subprocess.

  property pid: int | None

    The pid of the running process. None if not running.

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

  Bases: `Factory`

  Base shell factory class.

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

    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]

    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, environ: EnvironDict = NOTHING, slow_stop: bool = True, system_encoding: str = NOTHING, timeout: int | float = NOTHING, script_name: str, base_script_args: List[str] = NOTHING)

  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

    Returns a human readable name for the factory.

  get_script_path() → str

    Returns the path to the script to run.

  get_base_script_args() → List[str]

    Returns any additional arguments to pass to the CLI script.

  get_script_args() → List[str]

    Returns any additional arguments to pass to the CLI script.

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

    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])

  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, after_start_callbacks: List[Callback] = NOTHING, before_terminate_callbacks: List[Callback] = NOTHING, after_terminate_callbacks: List[Callback] = NOTHING)

  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

    Register a function callback to run before the daemon starts.

    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

    Register a function callback to run after the daemon starts.

    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

    Register a function callback to run before the daemon terminates.

    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

    Register a function callback to run after the daemon terminates.

    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

    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

    Terminate the daemon.

  get_start_arguments() → StartDaemonCallArguments

    Return the arguments and keyword arguments used when starting the daemon.

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

  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

    Post `attrs` class initialization routines.

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

    Register a function callback to run before the daemon starts.

    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

    Register a function callback to run after the daemon starts.

    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

    Register a function callback to run before the daemon terminates.

    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

    Register a function callback to run after the daemon terminates.

    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

    Register a function to run after the daemon starts to confirm readiness for work.

    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]

    Return a list of ports to check against to ensure the daemon is running.

  get_start_check_callbacks() → List[Callback]

    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

    Start the daemon.

  started(*extra_cli_arguments: str, max_start_attempts: int | None = None, start_timeout: int | float | None = None) → Generator[Daemon, None, None]

    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]

    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

    Run checks to confirm that the daemon has started.

  __enter__() → Daemon

    Use class as a context manager.

  __exit__(*_: Any) → None

    Exit the class context manager.

## pytestshellutils.version module