Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
raise MessageError("program %s not found" % show_name)
def _get_signal(self, args):
if args[-1].lower() in ('quit', 'hup', 'kill', 'term',):
return args[:-1], getattr(signal, "SIG%s" % args[-1].upper())
raise MessageError("signal %r not supported" % args[-1])
Command = CommandMeta('Command', (Command,), {})
#########################################
# commands
#########################################
class Quit(Command):
"""\
Quit the trainer immediately.
"""
name = "quit"
options = [('', 'terminate', False, "quit immediately")]
def message(self, *args, **opts):
if not opts.get("terminate"):
return "QUIT graceful"
return "QUIT"
def execute(self, trainer, args):
if len(args) > 1:
raise ArgumentError("invalid number of arguments")
return copy.copy(self)
def _get_show(self, trainer, show_name):
""" get show from the trainer if any """
try:
return trainer.get_show(show_name.lower())
except KeyError:
raise MessageError("program %s not found" % show_name)
def _get_signal(self, args):
if args[-1].lower() in ('quit', 'hup', 'kill', 'term',):
return args[:-1], getattr(signal, "SIG%s" % args[-1].upper())
raise MessageError("signal %r not supported" % args[-1])
Command = CommandMeta('Command', (Command,), {})
#########################################
# commands
#########################################
class Quit(Command):
"""\
Quit the trainer immediately.
"""
name = "quit"
options = [('', 'terminate', False, "quit immediately")]
def message(self, *args, **opts):
if not opts.get("terminate"):
return "STATUS %s" % args[0]
else:
return "STATUS"
def execute(self, trainer, args):
if len(args) > 1:
raise MessageError("message invalid")
if len(args) == 1:
show = self._get_show(trainer, args[0])
return show.status()
else:
return trainer.statuses()
class Stats(Command):
"""Get process infos"""
name = "stats"
def message(self, *args, **opts):
if len(args) > 2:
raise ArgumentError("message invalid")
if len(args) == 2:
return "STATS %s %s" % (args[0], args[1])
elif len(args) == 1:
return "STATS %s" % args[0]
else:
return "STATS"
def execute(self, trainer, args):
return MessageError("List of key/values is invalid")
# apply needed changes
action = 0
rest = args
while len(rest) > 0:
kv, rest = rest[:2], rest[2:]
new_action = show.set_opt(kv[0], kv[1])
if new_action == 1:
action = 1
# trigger needed action
show.do_action(action)
class Get(Command):
"""Get the value of a show option"""
name = "get"
def message(self, *args, **opts):
if len(args) < 2:
raise ArgumentError("number of arguments invalid")
return "GET %s %s" % (args[0], args[1])
def execute(self, trainer, args):
if len(args) < 2:
raise MessageError("invalid number of parameters")
show = self._get_show(trainer, args.pop(0))
# get options values. It return an error if one of the asked
def message(self, *args, **opts):
if len(args) < 1:
raise ArgumentError("number of arguments invalid")
return "OPTIONS %s" % args[0]
def execute(self, trainer, args):
if len(args) < 1:
raise ArgumentError("number of arguments invalid")
show = self._get_show(trainer, args[0])
return "\n".join(["%s:%s" % (k, v) for k, v in show.options()])
class Set(Command):
""" Set a show option"""
name = "set"
def message(self, *args, **opts):
if len(args) < 3:
raise ArgumentError("number of arguments invalid")
args = list(args)
show_name = args.pop(0)
if len(args) % 2 != 0:
raise ArgumentError("List of key/values is invalid")
return "SET %s %s" % (show_name, " ".join(args))
def execute(self, trainer, args):
if len(args) == 1:
return "START %s" % args[0]
else:
return "START"
def execute(self, trainer, args):
if len(args) == 1:
show = self._get_show(trainer, args[0])
show.start()
else:
trainer.start_shows()
return "ok"
class Stop(Command):
"""\
Stop a show or all shows gracefully or not
"""
name = "stop"
options = [('', 'terminate', False, "stop immediately")]
def message(self, *args, **opts):
if len(args) > 1:
raise ArgumentError("invalid number of arguments")
if len(args) == 1:
msg = "STOP %s" % args[0]
else:
msg = "STOP"
def message(self, *args, **opts):
if len(args) < 1:
raise ArgumentError("number of arguments invalid")
return "DECR %s" % args[0]
def execute(self, trainer, args):
if len(args) < 1:
raise ArgumentError("number of arguments invalid")
show = self._get_show(trainer, args[0])
return str(show.decr())
class Options(Command):
"""Get show options"""
name = "options"
def message(self, *args, **opts):
if len(args) < 1:
raise ArgumentError("number of arguments invalid")
return "OPTIONS %s" % args[0]
def execute(self, trainer, args):
if len(args) < 1:
raise ArgumentError("number of arguments invalid")
show = self._get_show(trainer, args[0])
raise ArgumentError("number of arguments invalid")
msg = "ADD %s %s" % (args[0], args[1])
if opts.get("start", False):
return [msg, "START %s" % args[0]]
return msg
def execute(self, trainer, args):
if len(args) < 2:
raise MessageError("message invalid")
trainer.add_show(args[0], args[1])
return "ok"
class RmShow(Command):
"""Remove a show"""
name = "rm"
options = [('', 'terminate', False, "stop immediately")]
def message(self, *args, **opts):
if len(args) < 1 or len(args) > 1:
raise ArgumentError("number of arguments invalid")
if not opts.get("terminate", False):
return "RM %s graceful" % args[0]
else:
return "RM %s" % args[0]
def execute(self, trainer, args):
if len(args) < 1 or len(args) > 1:
else:
return "RESTART"
def execute(self, trainer, args):
if len(args) > 1:
raise MessageError("message invalid")
if len(args) == 1:
show = self._get_show(trainer, args[0])
show.restart()
else:
trainer.restart()
return "ok"
class Reload(Command):
"""Reload the trainer or a show """
name = "reload"
options = [('', 'terminate', False, "stop immediately")]
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
show = self._get_show(trainer, args.pop(0))
# get options values. It return an error if one of the asked
# options isn't found
ret = []
for name in args:
if name in show.optnames:
val = show.get_opt(name)
ret.append("%s: %s" % (name, val))
else:
return "error: %r option not found" % name
return "\n".join(ret)
class Signal(Command):
"""Send a signal """
name = "signal"
options = [('', 'children', True, "Only signal children of the fly")]
def message(self, *args, **opts):
largs = len(args)
if largs < 2 or largs > 4:
raise ArgumentError("number of arguments invalid")
msg = "SIGNAL %s" % " ".join(args)
if not opts.get("children", False):
return msg
return "%s children" % msg