pytestshellutils package¶
Subpackages¶
- pytestshellutils.utils package
resolved_pathlib_path()format_callback_to_string()warn_until()- Submodules
- pytestshellutils.utils.ports module
- pytestshellutils.utils.processes module
- pytestshellutils.utils.socket module
- pytestshellutils.utils.time module
Submodules¶
pytestshellutils.customtypes module¶
Custom Types.
- class pytestshellutils.customtypes.GenericCallback(*args, **kwargs)[source]¶
Bases:
ProtocolGeneric callback function.
- class pytestshellutils.customtypes.DaemonCallback(*args, **kwargs)[source]¶
Bases:
ProtocolDaemon callback function.
pytestshellutils.exceptions module¶
Pytest Shell Utilities related exceptions.
- exception pytestshellutils.exceptions.ShellUtilsException[source]¶
Bases:
ExceptionBase pytest shell utilities exception.
- exception pytestshellutils.exceptions.CallbackException[source]¶
Bases:
ShellUtilsExceptionException raised during a before/after start/stop daemon callback.
- exception pytestshellutils.exceptions.ProcessFailed(message: str, process_result: ProcessResult | None = None)[source]¶
Bases:
ShellUtilsExceptionException raised when a sub-process fails.
- Parameters:
message – The exception message
- Keyword Arguments:
process_result – The
ProcessResultinstance when the exception occurred
- exception pytestshellutils.exceptions.FactoryFailure(message: str, process_result: ProcessResult | None = None)[source]¶
Bases:
ProcessFailedException raised when a sub-process fails on one of the factories.
- exception pytestshellutils.exceptions.FactoryNotStarted(message: str, process_result: ProcessResult | None = None)[source]¶
Bases:
FactoryFailureException raised when a factory failed to start.
Please look at
FactoryFailurefor the supported keyword arguments documentation.
- exception pytestshellutils.exceptions.FactoryNotRunning(message: str, process_result: ProcessResult | None = None)[source]¶
Bases:
FactoryFailureException raised when trying to use a factory’s .stopped context manager and the factory is not running.
Please look at
FactoryFailurefor the supported keyword arguments documentation.
- exception pytestshellutils.exceptions.ProcessNotStarted(message: str, process_result: ProcessResult | None = None)[source]¶
Bases:
FactoryFailureException raised when a process failed to start.
Please look at
FactoryFailurefor the supported keywords. arguments documentation.
- exception pytestshellutils.exceptions.FactoryTimeout(message: str, process_result: ProcessResult | None = None)[source]¶
Bases:
FactoryNotStartedException raised when a process timed-out.
Please look at
FactoryFailurefor 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, environ: EnvironDict = NOTHING)[source]¶
Bases:
objectBase factory class.
- Keyword Arguments:
cwd – The path to the desired working directory
environ – A dictionary of
key,valuepairs to add to the environment.
- cwd: Path¶
- environ: EnvironDict¶
- class pytestshellutils.shell.SubprocessImpl(*, factory: Factory | Subprocess | ScriptSubprocess)[source]¶
Bases:
objectSubprocess interaction implementation.
- Parameters:
factory – The factory instance, either
Subprocessor 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._terminaland also collects an initial listing of child processes which will be used when terminating the terminal
- 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.
- 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,valuepairs to add to thepytestshellutils.shell.Factory.environ.cwd – A path for the CWD when running the process.
- Returns:
A
Popeninstance.
- class pytestshellutils.shell.Factory(*, cwd: str | Path = NOTHING, environ: EnvironDict = NOTHING, slow_stop: bool = True, system_encoding: str = NOTHING, timeout: int | float = NOTHING)[source]¶
Bases:
BaseFactoryBase shell factory class.
- Keyword Arguments:
slow_stop – Whether to terminate the processes by sending a
SIGTERMsignal or by callingterminate()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_timeoutkeyword argument, and, in that case, the timeout value applied would be that of_timeoutinstead ofself.timeout.
- impl: SubprocessImpl¶
- terminate() ProcessResult[source]¶
Terminate the started subprocess.
- __ne__(other)¶
Method generated by attrs for class Factory.
- class pytestshellutils.shell.Subprocess(*, cwd: str | Path = NOTHING, environ: EnvironDict = NOTHING, slow_stop: bool = True, system_encoding: str = NOTHING, timeout: int | float = NOTHING)[source]¶
Bases:
FactoryBase 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 runenv – 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 notNone, it will be used instead oftimeout, the default timeout.
- 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)[source]¶
Bases:
SubprocessBase 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
$PATHvariable.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
Factoryfor the additional supported keyword arguments documentation.
- class pytestshellutils.shell.StartDaemonCallArguments(*, args: Tuple[str, ...], kwargs: Dict[str, Any])[source]¶
Bases:
objectThis 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
- 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)[source]¶
Bases:
SubprocessImplDaemon subprocess interaction implementation.
Please look at
SubprocessImplfor the additional supported keyword arguments documentation.- before_start(callback: Callable[[], None], *args: Any, **kwargs: Any) None[source]¶
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[source]¶
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[source]¶
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[source]¶
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[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:
- 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, 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)[source]¶
Bases:
ScriptSubprocessBase 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
Subprocessfor the additional supported keyword arguments documentation.- impl: DaemonImpl¶
- stats_processes: StatsProcesses¶
- before_start(callback: Callable[[], None], *args: Any, **kwargs: Any) None[source]¶
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[source]¶
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[source]¶
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[source]¶
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[source]¶
Register a function to run after the daemon starts to confirm readiness for work.
The callback must accept as the first argument
timeout_atwhich is a float. The callback must stop trying to confirm running behavior oncetime.time() > timeout_at. The callback should returnTrueto 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.
- 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