Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
return util.to_bool(val)
elif key == "use_papa":
return util.to_bool(val)
elif key.startswith('stderr_stream.') or key.startswith('stdout_stream.'):
subkey = key.split('.', 1)[-1]
if subkey in ('max_bytes', 'backup_count'):
return int(val)
return val
elif key == 'hooks':
res = {}
for hook in val.split(','):
if hook == '':
continue
hook = hook.split(':')
if len(hook) != 2:
raise ArgumentError(hook)
name, value = hook
if name not in _HOOKS:
raise ArgumentError(name)
res[name] = value
return res
elif key.startswith('hooks.'):
# we can also set a single hook
name = key.split('.', 1)[-1]
if name not in _HOOKS:
raise ArgumentError(name)
return val
elif key.startswith('rlimit_'):
return int(val)
hook = hook.split(':')
if len(hook) != 2:
raise ArgumentError(hook)
name, value = hook
if name not in _HOOKS:
raise ArgumentError(name)
res[name] = value
return res
elif key.startswith('hooks.'):
# we can also set a single hook
name = key.split('.', 1)[-1]
if name not in _HOOKS:
raise ArgumentError(name)
return val
elif key.startswith('rlimit_'):
return int(val)
raise ArgumentError("unknown key %r" % key)
def message(self, *args, **opts):
if len(args) > 1:
raise ArgumentError("Invalid number of arguments")
if len(args) == 1:
return self.make_message(name=args[0], **opts)
return self.make_message(**opts)
def message(self, *args, **opts):
if len(args) > 0:
raise ArgumentError("Invalid number of arguments")
return self.make_message()
def message(self, *args, **opts):
if len(args) < 1:
raise ArgumentError("number of arguments invalid")
return "INCR %s" % args[0]
def validate(self, props):
super(Signal, self).validate(props)
if 'childpid' in props and 'pid' not in props:
raise ArgumentError('cannot specify childpid without pid')
try:
props['signum'] = to_signum(props['signum'])
except ValueError:
raise MessageError('signal invalid')
def message(self, *args, **opts):
if len(args) < 3:
raise ArgumentError("Invalid number of arguments")
args = list(args)
watcher_name = args.pop(0)
if len(args) % 2 != 0:
raise ArgumentError("List of key/values is invalid")
options = {}
while len(args) > 0:
kv, args = args[:2], args[2:]
kvl = kv[0].lower()
options[kvl] = convert_option(kvl, kv[1])
return self.make_message(name=watcher_name, options=options)
def message(self, *args, **opts):
if len(args) < 1 or len(args) > 2:
raise ArgumentError("Invalid number of arguments")
props = {}
props['name'] = args[0]
if len(args) > 1:
props['pid'] = args[1]
if opts['signum'] is not None:
props['signum'] = opts['signum']
if opts['graceful_timeout'] is not None:
props['graceful_timeout'] = opts['graceful_timeout']
return self.make_message(**props)
def message(self, *args, **opts):
if len(args) > 1:
raise ArgumentError("invalid number of arguments")
if len(args) == 1:
msg = "RELOAD %s" % args[0]
else:
msg = "RELOAD"
if not opts.get("terminate", False):
return "%s graceful" % msg
return msg
def message(self, *args, **opts):
if len(args) != 1:
raise ArgumentError("invalid number of arguments")
return self.make_message(name=args[0])