Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
>>> svstat_parse('unable to chdir: file does not exist')
no such service
>>> svstat_parse('totally unpredictable error message')
totally unpredictable error message
>>> svstat_parse('down (exitcode 0) 0 seconds, normally up, want wat, ready 0 seconds')
Traceback (most recent call last):
...
ValueError: unexpected value for `process`: 'wat'
>>> svstat_parse('down (exitcode 0) 0 seconds, normally up, want up\x00, ready 0 seconds')
down (exitcode 0) 0 seconds, starting
'''
status = svstat_string.strip()
trace('RAW : %s', status)
if status.startswith(('up ', 'down ')):
state, buffer = parse(status, '', ' ')
elif status.startswith('unable to chdir:'):
return SvStat(SvStat.INVALID, None, None, None, None)
elif (
status.startswith('s6-svstat: fatal: unable to read status for ') and status.endswith((
': No such file or directory',
': Broken pipe',
))
):
# the service has never been started before; it's down.
return SvStat(SvStat.UNSUPERVISED, None, None, None, None)
else: # unknown errors
return SvStat(status, None, None, None, None)
pid, buffer = parse(buffer, '(pid ', ') ', int)
def listdir(path):
from os import listdir
try:
return listdir(path)
except EnvironmentError as error:
trace('fuser suppressed: %s', error)
return ()
def svc(args):
"""Wrapper for daemontools svc cmd"""
# svc never writes to stdout.
cmd = ('s6-svc',) + tuple(args)
trace('CMD: %s', cmd)
process = Popen(cmd, stderr=PIPE)
_, error = process.communicate()
if error.startswith(b's6-svc: fatal: unable to control '):
raise Unsupervised(cmd, error)
if process.returncode: # pragma: no cover: there's no known way to hit this.
import sys
sys.stderr.write(error.decode('UTF-8'))
raise CalledProcessError(process.returncode, cmd)
def _svstat_path(self, path):
with path.dirpath().as_cwd():
result = svstat(path.basename)
if not self.notification_fd.exists():
# services without notification need to be considered ready sometimes
if (
# an 'up' service is always ready
(result.state == 'up' and result.process is None) or
# restarting continuously and successfully can/should be considered 'ready'
(result.process == 'starting' and result.exitcode == 0 and result.seconds == 0)
):
result = result._replace(state='ready')
trace('PARSED: %s', result)
return result
def timeout(service, start_time, check_time, curr_time):
limit_time = start_time + service.get_timeout()
next_time = curr_time + (curr_time - check_time)
# assertion can take a long time. we timeout as close to the limit_time as we can.
if abs(curr_time - limit_time) < abs(next_time - limit_time):
return True
else:
trace('service %s still waiting: %.1f seconds.', service.name, limit_time - curr_time)
return False
def stat(path):
from os import stat
try:
return stat(path)
except EnvironmentError as error:
trace('fuser suppressed: %s', error)
return None