Source code for idem_posix.exec.cmdmod

"""Exec module for executing commands on a posix systems."""
import asyncio
import copy
import json
import os
import sys
import traceback
from typing import Any
from typing import Dict
from typing import List

__virtualname__ = "cmd"


def __virtual__(hub):
    return os.name == "posix", "idem-posix only runs on posix systems"


[docs]async def run( hub, cmd: str or List[str], cwd: str = None, shell: bool = False, stdin: str = None, stdout: int = asyncio.subprocess.PIPE, stderr: int = asyncio.subprocess.PIPE, render_pipe: str = None, env: Dict[str, Any] = None, timeout: int or float = None, is_string_output: bool = False, success_retcodes: List[int] = None, *args, **kwargs, ) -> Dict[str, Any]: """Execute the passed command and return the output as a string Args: cmd(str or list[str]): The command to run. ex: ``ls -lart /home`` cwd(str, Optional): The directory from which to execute the command. Defaults to the home directory of the user specified by ``runas`` (or the user under which Salt is running if ``runas`` is not specified). shell(bool): If ``False``, let python handle the positional arguments. Set to ``True`` to use shell features, such as pipes or redirection. Defaults to False. stdin(str, Optional): A string of standard input can be specified for the command to be run using the ``stdin`` parameter. This can be useful in cases where sensitive information must be read from standard input. stdout(int): A string of standard input can be specified for the command to be run using the ``stdout`` parameter. stderr(int): A string of standard input can be specified for the command to be run using the ``stderr`` parameter. render_pipe(str, Optional): The render pipe to use on the output. Defaults to None. env(dict[str], Optional): Environment variables to be set prior to execution. Defaults to None. .. note:: When passing environment variables on the CLI, they should be passed as the string representation of a dictionary. .. code-block:: bash idem exec cmd.run 'some command' env='{"FOO": "bar"}' timeout(int or float, Optional): A timeout in seconds for the executed process to return. Defaults to None. is_string_output(bool): Give the output in string format irrespective of format the command executed returns. Defaults to False. success_retcodes(list[int], Optional): The result will be True if the command's return code is in this list. Defaults to [0]. args: args that will be forwarded to subprocess. kwargs: kwargs that will be forwarded to subprocess. Returns: Dict[str, Any]: Returns command output * stdout(str): The plaintext output of the command. * stderr(str): The plaintext error/logging output of the command. * retcode(str): The return code from the command. * state(str): "The output as rendered from the render_pipe (if one was given), for use in arg_binding. Example: .. code-block:: bash idem exec cmd.run "shell command --with-flags" cwd=/home render_pipe=json timeout=10 """ ret = dict( result=True, comment="", ret={"retcode": 0, "state": {}, "stdout": b"", "stderr": b""}, ) if success_retcodes is None: success_retcodes = [0] else: success_retcodes = [int(retcode) for retcode in success_retcodes] if getattr(sys, "frozen", False): env = copy.copy(os.environ.copy()) # Remove the LOAD LIBRARY_PATH for running commands # https://pyinstaller.readthedocs.io/en/stable/runtime-information.html#ld-library-path-libpath-considerations env.pop("LD_LIBRARY_PATH", None) # Linux env.pop("LIBPATH", None) # AIX # Run the command try: if shell: proc = await asyncio.create_subprocess_shell( cmd, cwd=cwd, stdout=stdout, stderr=stderr, env=env, **kwargs ) else: proc = await asyncio.create_subprocess_exec( *cmd, *args, cwd=cwd, stdout=stdout, stderr=stderr, env=env, **kwargs ) except Exception as e: ret["result"] = False ret["ret"]["retcode"] = sys.exc_info()[1].errno ret["comment"] = traceback.format_exc() ret["ret"]["stderr"] = f"{e.__class__.__name__}: {e}" return ret # This is where the magic happens out, err = await asyncio.wait_for(proc.communicate(input=stdin), timeout=timeout) std_out = out.decode() if not is_string_output: # check if we can parse the output to json. try: std_out = json.loads(std_out) except Exception: hub.log.info("command output is not json") ret["ret"]["stdout"] = std_out ret["comment"] = ret["ret"]["stderr"] = err.decode() ret["ret"]["retcode"] = await asyncio.wait_for(proc.wait(), timeout=timeout) ret["result"] = ret["ret"]["retcode"] in success_retcodes if render_pipe: block = {"bytes": out} rendered = await hub.rend.init.parse_bytes(block, pipe=render_pipe) ret["ret"]["state"] = rendered return ret