Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def it_stays_locked_for_the_lifetime_of_subprocesses(self, tmpfile):
from pgctl.subprocess import Popen
with flock(tmpfile):
p = Popen(('sleep', '99999'))
assert p.poll() is None
assert_locked(tmpfile)
assert_locked(tmpfile)
p.terminate()
assert p.wait() == -15
with flock(tmpfile):
print('oh hi there!')
if not self.is_logger_running():
Popen(
(
's6-supervise',
self.path.join('.log').strpath,
),
env=self.supervise_env(lock, debug=False),
preexec_fn=functools.partial(
logger_preexec,
log_fifo_path,
),
close_fds=True,
)
Popen(
(
's6-supervise',
self.path.strpath,
),
env=self.supervise_env(lock, debug=False),
preexec_fn=functools.partial(
supervisor_preexec,
log_fifo_path,
),
def background(self):
"""Run supervise(1), while ensuring it is properly symlinked."""
if self.supervised():
return
with self.flock() as lock:
log_fifo_path = self.path.join('log_pipe').strpath
try:
os.mkfifo(log_fifo_path)
except OSError as e:
if e.errno != errno.EEXIST:
raise
if not self.is_logger_running():
Popen(
(
's6-supervise',
self.path.join('.log').strpath,
),
env=self.supervise_env(lock, debug=False),
preexec_fn=functools.partial(
logger_preexec,
log_fifo_path,
),
close_fds=True,
)
Popen(
(
's6-supervise',
self.path.strpath,