Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def run_cmd(
engine,
cmd,
shell=False,
cwd=None,
check_returncode=True,
callback=None,
plugin_id=None,
):
"""Run command.
From https://github.com/vcs-python/libvcs/.
"""
proc = subprocess.Popen(
cmd,
shell=shell,
stderr=subprocess.PIPE,
stdout=subprocess.PIPE,
creationflags=0,
bufsize=1,
cwd=cwd,
)
if plugin_id is not None:
set_plugin_pid(engine, plugin_id, proc.pid)
all_output = []
code = None
line = None
while code is None:
code = proc.poll()
args = " ".join(args)
logger.info("Task subprocess args: %s", args)
# set system/version dependent "start_new_session" analogs
# https://docs.python.org/2/library/subprocess.html#converting-argument-sequence
kwargs = {}
if sys.platform != "win32":
kwargs.update(preexec_fn=os.setsid)
progress = 100
logging_callback(progress, type="progress")
try:
_env = plugin_env.copy()
_env.update(environment_variables)
logger.debug("Running process with env: %s", _env)
process = subprocess.Popen(
args,
bufsize=1,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
shell=True,
env=_env,
cwd=work_dir,
**kwargs,
)
logging_callback(f"Running subprocess (pid={process.pid}) with {args}")
set_plugin_pid(engine, plugin_id, process.pid)
# Poll process for new output until finished
stdfn = sys.stdout.fileno()
progress = 0
logging_callback(progress, type="progress")