Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
If the process exits with an error while checks are being polled, executor
should raise an exception.
"""
error_code = 12
failing_executor = Executor(
['bash', '-c', f'exit {error_code!s}'],
timeout=5
)
failing_executor.pre_start_check = mock.Mock( # type: ignore
return_value=False)
# After-start check will keep returning False to let the process terminate.
failing_executor.after_start_check = mock.Mock( # type: ignore
return_value=False)
with pytest.raises(ProcessExitedWithError) as exc:
failing_executor.start()
assert exc.value.exit_code == 12
error_msg = f'exited with a non-zero code: {error_code!s}'
assert error_msg in str(exc.value)
# Pre-start check should have been called - after-start check might or
# might not have been called - depending on the timing.
assert failing_executor.pre_start_check.called is True # type: ignore
:rtype: bool
:return: the actual check status or False before starting the process
:raise ProcessExitedWithError: when the main process exits with
an error
"""
if self.process is None: # pragma: no cover
# No process was started.
return False
exit_code = self.process.poll()
if exit_code is not None and exit_code != 0:
# The main process exited with an error. Clean up the children
# if any.
self._kill_all_kids(self._sig_kill)
self._clear_process()
raise ProcessExitedWithError(self, exit_code)
return self.after_start_check()
"stdin": stdin,
"stdout": stdout,
"stderr": stderr,
"universal_newlines": True,
"env": env,
"cwd": self.cwd,
}
if platform.system() != "Windows":
popen_kwargs["preexec_fn"] = os.setsid
self.process = subprocess.Popen(command, **popen_kwargs)
self._set_timeout()
try:
self.wait_for(self.check_subprocess)
except ProcessExitedWithError as e:
if e.exit_code == 127:
raise FileNotFoundError(
f"Can not execute {command!r}, check that the executable exists."
) from e
else:
output_file_names = {io.name for io in (stdout, stderr) if hasattr(io, "name")}
if output_file_names:
log.warning("Process output file(s)", output_files=output_file_names)
raise
return self