Base Classes#

Factories base classes.

class saltfactories.bases.SaltMixin(*, config, config_dir=_Nothing.NOTHING, python_executable=None, system_service=False)[source]#

Bases: object

Base factory for salt cli’s and daemon’s.

Parameters:

  • config (dict) – The Salt config dictionary

  • python_executable (str) – The path to the python executable to use

  • system_service (bool) – If true, the daemons and CLI’s are run against a system installed salt setup, ie, the default salt system paths apply.

get_display_name()[source]#

Returns a human readable name for the factory.

class saltfactories.bases.SaltCliImpl(*, factory)[source]#

Bases: SubprocessImpl

Salt CLI’s subprocess interaction implementation.

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

Parameters:

factory (Factory | Subprocess | ScriptSubprocess) –

cmdline(*args, minion_tgt=None, **kwargs)[source]#

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

Parameters:

  • args (str) – Additional arguments to use when starting the subprocess

  • minion_tgt (str) – The minion ID to target

  • kwargs – Additional keyword arguments will be converted into key=value pairs to be consumed by the salt CLI’s

init_terminal(cmdline, shell=False, env=None, cwd=None)#

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

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

Parameters:
Return type:

Popen[Any]

is_running()#

Returns true if the sub-process is alive.

Returns:

Returns true if the sub-process is alive

Return type:

bool

property pid: int | None#

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

run(*args, shell=False, env=None, cwd=None, **kwargs)#

Run the given command synchronously.

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

Parameters:
Return type:

Popen[Any]

terminate()#

Terminate the started subprocess.

Return type:

ProcessResult

class saltfactories.bases.SaltCli(*, config, config_dir=_Nothing.NOTHING, python_executable=None, system_service=False, cwd=_Nothing.NOTHING, environ=_Nothing.NOTHING, slow_stop=True, system_encoding=_Nothing.NOTHING, timeout=_Nothing.NOTHING, script_name, base_script_args=_Nothing.NOTHING, hard_crash=False, merge_json_output=True)[source]#

Bases: SaltMixin, ScriptSubprocess

Base factory for salt cli’s.

Parameters:

  • hard_crash (bool) – Pass --hard-crash to Salt’s CLI’s

  • cwd (Path) –

  • environ (EnvironDict) –

  • slow_stop (bool) –

  • system_encoding (str) –

  • timeout (int | float) –

  • script_name (str) –

  • base_script_args (List[str]) –

Please look at Salt and ScriptSubprocess for the additional supported keyword arguments documentation.

get_script_args()[source]#

Returns any additional arguments to pass to the CLI script.

get_minion_tgt(minion_tgt=None)[source]#

Return the minion target ID.

cmdline(*args, minion_tgt=None, merge_json_output=None, **kwargs)[source]#

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

Parameters:

  • args (str) – Additional arguments to use when starting the subprocess

  • minion_tgt (str) – The minion ID to target

  • merge_json_output (bool) – The default behavior of salt outputters is to print one line per minion return, which makes parsing the whole output as JSON impossible when targeting multiple minions. If this value is True, an attempt is made to merge each JSON line into a single dictionary.

  • kwargs – Additional keyword arguments will be converted into key=value pairs to be consumed by the salt CLI’s

process_output(stdout, stderr, cmdline=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)

Return type:

tuple

get_base_script_args()#

Returns any additional arguments to pass to the CLI script.

Return type:

List[str]

get_display_name()#

Returns a human readable name for the factory.

get_script_path()#

Returns the path to the script to run.

Return type:

str

is_running()#

Returns true if the sub-process is alive.

Return type:

bool

property pid: int | None#

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

run(*args, env=None, _timeout=None, **kwargs)#

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.

Parameters:
Return type:

ProcessResult

terminate()#

Terminate the started subprocess.

Return type:

ProcessResult

class saltfactories.bases.SystemdSaltDaemonImpl(*, factory, before_start_callbacks=_Nothing.NOTHING, after_start_callbacks=_Nothing.NOTHING, before_terminate_callbacks=_Nothing.NOTHING, after_terminate_callbacks=_Nothing.NOTHING)[source]#

Bases: DaemonImpl

Daemon systemd interaction implementation.

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

Parameters:
cmdline(*args)[source]#

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

Parameters:

args (str) – Additional arguments to use when starting the subprocess

get_service_name()[source]#

Return the systemd service name.

is_running()[source]#

Returns true if the sub-process is alive.

property pid#

Return the pid of the running process.

start(*extra_cli_arguments, max_start_attempts=None, start_timeout=None)[source]#

Start the daemon.

after_start(callback, *args, **kwargs)#

Register a function callback to run after the daemon starts.

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

Parameters:
Return type:

None

after_terminate(callback, *args, **kwargs)#

Register a function callback to run after the daemon terminates.

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

Parameters:
Return type:

None

before_start(callback, *args, **kwargs)#

Register a function callback to run before the daemon starts.

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

Parameters:
Return type:

None

before_terminate(callback, *args, **kwargs)#

Register a function callback to run before the daemon terminates.

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

Parameters:
Return type:

None

get_start_arguments()#

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

Return type:

StartDaemonCallArguments

init_terminal(cmdline, shell=False, env=None, cwd=None)#

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

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

Parameters:
Return type:

Popen[Any]

run(*args, shell=False, env=None, cwd=None, **kwargs)#

Run the given command synchronously.

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

Parameters:
Return type:

Popen[Any]

terminate()#

Terminate the daemon.

Return type:

ProcessResult

class saltfactories.bases.SaltDaemon(*, config, config_dir=_Nothing.NOTHING, python_executable=None, system_service=False, cwd=_Nothing.NOTHING, environ=_Nothing.NOTHING, slow_stop=True, system_encoding=_Nothing.NOTHING, timeout=_Nothing.NOTHING, script_name, base_script_args=_Nothing.NOTHING, check_ports=_Nothing.NOTHING, stats_processes=None, start_timeout, max_start_attempts=3, extra_cli_arguments_after_first_start_failure=_Nothing.NOTHING, start_checks_callbacks=_Nothing.NOTHING, event_listener=None, factories_manager=None, started_at=None)[source]#

Bases: SaltMixin, Daemon

Base factory for salt daemon’s.

Please look at SaltMixin and Daemon for the additional supported keyword arguments documentation.

Parameters:

  • cwd (str | Path) –

  • environ (EnvironDict) –

  • slow_stop (bool) –

  • system_encoding (str) –

  • timeout (int | float) –

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

  • start_checks_callbacks (List[Callback]) –

classmethod configure(factories_manager, daemon_id, root_dir=None, defaults=None, overrides=None, **configure_kwargs)[source]#

Configure the salt daemon.

classmethod verify_config(config)[source]#

Verify the configuration dictionary.

classmethod write_config(config)[source]#

Write the configuration to file.

classmethod load_config(config_file, config)[source]#

Return the loaded configuration.

Should return the configuration as the daemon would have loaded after parsing the CLI

get_check_events()[source]#

Return salt events to check.

Returns list of tuples in the form of (master_id, event_tag) check against to ensure the daemon is running.

cmdline(*args)[source]#

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

Parameters:

args (str) – Additional arguments to use when starting the subprocess

after_start(callback, *args, **kwargs)#

Register a function callback to run after the daemon starts.

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

Parameters:
Return type:

None

after_terminate(callback, *args, **kwargs)#

Register a function callback to run after the daemon terminates.

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

Parameters:
Return type:

None

before_start(callback, *args, **kwargs)#

Register a function callback to run before the daemon starts.

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

Parameters:
Return type:

None

before_terminate(callback, *args, **kwargs)#

Register a function callback to run before the daemon terminates.

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

Parameters:
Return type:

None

get_base_script_args()#

Returns any additional arguments to pass to the CLI script.

Return type:

List[str]

get_check_ports()#

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

Return type:

List[int]

get_display_name()#

Returns a human readable name for the factory.

get_script_args()#

Returns any additional arguments to pass to the CLI script.

Return type:

List[str]

get_script_path()#

Returns the path to the script to run.

Return type:

str

get_start_check_callbacks()#

Return a list of the start check callbacks.

Return type:

List[Callback]

is_running()#

Returns true if the sub-process is alive.

Return type:

bool

property pid: int | None#

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

process_output(stdout, stderr, cmdline=None)#

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

Returns:

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

Parameters:

  • stdout (str) –

  • stderr (str) –

  • cmdline (List[str] | None) –

Return type:

Tuple[str, str, Dict[Any, Any] | None]

run(*args, env=None, _timeout=None, **kwargs)#

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.

Parameters:
Return type:

ProcessResult

run_start_checks(started_at, timeout_at)#

Run checks to confirm that the daemon has started.

Parameters:
Return type:

bool

start(*extra_cli_arguments, max_start_attempts=None, start_timeout=None)#

Start the daemon.

Parameters:

  • extra_cli_arguments (str) –

  • max_start_attempts (int | None) –

  • start_timeout (int | float | None) –

Return type:

bool

start_check(callback, *args, **kwargs)#

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.

Arguments:
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
Parameters:
Return type:

None

started(*extra_cli_arguments, max_start_attempts=None, start_timeout=None)#

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

Parameters:

  • extra_cli_arguments (str) –

  • max_start_attempts (int | None) –

  • start_timeout (int | float | None) –

Return type:

Generator[Daemon, None, None]

stopped(before_stop_callback=None, after_stop_callback=None, before_start_callback=None, after_start_callback=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
Parameters:
Return type:

Generator[Daemon, None, None]

terminate()#

Terminate the started subprocess.

Return type:

ProcessResult