Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_already_closed():
"""Check that the executor cleans after itself after it exited earlier."""
with SimpleExecutor('python') as executor:
assert executor.running()
os.killpg(executor.process.pid, SIGKILL)
def process_stopped():
"""Return True only only when self.process is not running."""
return executor.running() is False
executor.wait_for(process_stopped)
assert executor.process
assert not executor.process
def __init__( # pylint:disable=too-many-arguments
self,
command: Union[str, List[str], Tuple[str, ...]],
cwd: Optional[str] = None,
shell: bool = False,
timeout: Union[int, float] = 3600,
sleep: float = 0.1,
sig_stop: int = signal.SIGTERM,
sig_kill: int = SIGKILL,
envvars: Optional[Dict[str, str]] = None,
stdin: Union[None, int, IO[Any]] = subprocess.PIPE,
stdout: Union[None, int, IO[Any]] = subprocess.PIPE,
stderr: Union[None, int, IO[Any]] = None
) -> None:
"""
Initialize executor.
:param (str, list) command: command to be run by the subprocess
:param str cwd: current working directory to be set for executor
:param bool shell: same as the `subprocess.Popen` shell definition.
On Windows always set to True.
:param int timeout: number of seconds to wait for the process to start
or stop.
:param float sleep: how often to check for start/stop condition
:param int sig_stop: signal used to stop process run by the executor.
def cleanup_subprocesses() -> None:
"""On python exit: find possibly running subprocesses and kill them."""
# pylint: disable=redefined-outer-name, reimported, import-outside-toplevel
# atexit functions tends to loose global imports sometimes so reimport
# everything what is needed again here:
import os
import errno
from mirakuru.base_env import processes_with_env
from mirakuru.compat import SIGKILL
# pylint: enable=redefined-outer-name, reimported, import-outside-toplevel
pids = processes_with_env(ENV_UUID, str(os.getpid()))
for pid in pids:
try:
os.kill(pid, SIGKILL)
except OSError as err:
if err.errno != errno.ESRCH:
print("Can not kill the", pid, "leaked process", err)