Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
description = "test desc"
options_definitions = [
Option("test-opt", "test-opt desc", aliases=["x"])
]
aliases = ["mango"]
def run(self):
return 42
@register_command
class MyCommand2(Command):
name = "test2"
description = "test desc 2"
options_definitions = [
Option("test-opt", "test-opt desc", aliases=["x"]),
Option("test-opt2", "test-opt2 desc", default="banana"),
]
arguments_definitions = [
Argument("test-arg", "test-arg desc")
]
def run(self):
return 43
def test_command_class():
c = Command()
assert c.arguments is None
c.process_args([])
assert c.arguments is not None
with pytest.raises(AttributeError):
options_definitions = [
Option("test-opt", "test-opt desc", aliases=["x"])
]
aliases = ["mango"]
def run(self):
return 42
@register_command
class MyCommand2(Command):
name = "test2"
description = "test desc 2"
options_definitions = [
Option("test-opt", "test-opt desc", aliases=["x"]),
Option("test-opt2", "test-opt2 desc", default="banana"),
]
arguments_definitions = [
Argument("test-arg", "test-arg desc")
]
def run(self):
return 43
def test_command_class():
c = Command()
assert c.arguments is None
c.process_args([])
assert c.arguments is not None
with pytest.raises(AttributeError):
print(c.arguments.not_there)
except Exception as ex:
logger.info("command %r failed: %r", inp, ex)
ui.notify_message("Error while running command '{}': {}".format(
inp, ex
), level="error")
log_last_traceback()
ui.reload_footer()
@register_command
class PromptCommand(SameThreadCommand):
name = "prompt"
description = "Customize and pre-populate prompt with initial data."
options_definitions = [
Option("prompt-text", "Text forming the actual prompt", default=":"),
Option("initial-text", "Prepopulated text", default="")
]
def run(self):
"""
prompt for text input.
"""
# set up widgets
leftpart = urwid.Text(self.arguments.prompt_text, align='left')
editpart = urwid.Edit(multiline=True, edit_text=self.arguments.initial_text)
# build promptwidget
edit = urwid.Columns([
('fixed', len(self.arguments.prompt_text), leftpart),
('weight', 1, editpart),
])
self.ui.prompt_bar = urwid.AttrMap(edit, "main_list_dg")
@register_command
class QuitCommand(SameThreadCommand):
name = "quit"
# TODO: make this configurable by asking whether to quit or not
description = "Quit sen. No questions asked."
def run(self):
self.ui.quit()
@register_command
class KillBufferCommand(SameThreadCommand):
name = "kill-buffer"
description = "Remove currently displayed buffer."
options_definitions = [
Option("quit-if-no-buffer", "Quit when there's no buffer left", default=False)
]
def run(self):
buffers_left = self.ui.remove_current_buffer(close_if_no_buffer=self.arguments.quit_if_no_buffer)
if buffers_left is None:
self.ui.notify_message("Last buffer will not be removed.")
elif buffers_left == 0:
self.ui.run_command(QuitCommand.name)
@register_command
class SelectBufferCommand(SameThreadCommand):
name = "select-buffer"
description = "Display buffer with selected index."
arguments_definitions = [
Argument("index", "Index of buffer to display", default=1, action=int)
ui.notify_message(str(ex), level="error")
except Exception as ex:
logger.info("command %r failed: %r", inp, ex)
ui.notify_message("Error while running command '{}': {}".format(
inp, ex
), level="error")
log_last_traceback()
ui.reload_footer()
@register_command
class PromptCommand(SameThreadCommand):
name = "prompt"
description = "Customize and pre-populate prompt with initial data."
options_definitions = [
Option("prompt-text", "Text forming the actual prompt", default=":"),
Option("initial-text", "Prepopulated text", default="")
]
def run(self):
"""
prompt for text input.
"""
# set up widgets
leftpart = urwid.Text(self.arguments.prompt_text, align='left')
editpart = urwid.Edit(multiline=True, edit_text=self.arguments.initial_text)
# build promptwidget
edit = urwid.Columns([
('fixed', len(self.arguments.prompt_text), leftpart),
('weight', 1, editpart),
])
class MatchingOperationCommand(OperationCommand):
""" this is just a shortcut """
def run(self):
self.do(self.name)
@register_command
class RemoveCommand(OperationCommand):
name = "rm"
description = "remove provided object, image or container"
options_definitions = [Option("force",
"Force removal of the selected object.",
default=False, aliases=["-f", "f"]),
Option("yes",
"Don't ask before removing.",
default=False, aliases=["-y"])]
def run(self):
logger.debug("remove %s force=%s yes=%s", self.docker_object, self.arguments.force,
self.arguments.yes)
if not self.arguments.yes and not self.ui.yolo:
logger.debug("we need confirmation from user")
cmd = "rm -y" if not self.arguments.force else "rm -y -f"
self.ui.run_command(
"prompt prompt-text=\"You are about to remove %s %s, type enter to continue: \" "
"initial-text=\"%s\"" % (
self.docker_object.pretty_object_type.lower(),
self.docker_object.short_name,
cmd
),
description = "pause a container"
pre_info_message = "Pausing container {container_name}..."
@register_command
class UnpauseContainerCommand(MatchingOperationCommand):
name = "unpause"
description = "unpause a container"
pre_info_message = "Unpausing container {container_name}..."
@register_command
class LogsCommand(BackendCommand):
name = "logs"
description = "display logs of a container"
options_definitions = [Option("follow", "Follow logs.", default=False, aliases=["-f", "f"])]
def run(self):
self.ui.add_and_display_buffer(
LogsBuffer(self.ui, self.docker_object, follow=self.arguments.follow))
@register_command
class InspectCommand(BackendCommand):
name = "inspect"
description = "inspect provided object, a container or an image"
def run(self):
if not self.docker_object:
self.ui.notify_message("No docker object specified.", level="error")
return
self.ui.add_and_display_buffer(InspectBuffer(self.ui, self.docker_object))
self.ui.notify_widget(
get_operation_notify_widget(operation, notif_level=notif_level)
)
class MatchingOperationCommand(OperationCommand):
""" this is just a shortcut """
def run(self):
self.do(self.name)
@register_command
class RemoveCommand(OperationCommand):
name = "rm"
description = "remove provided object, image or container"
options_definitions = [Option("force",
"Force removal of the selected object.",
default=False, aliases=["-f", "f"]),
Option("yes",
"Don't ask before removing.",
default=False, aliases=["-y"])]
def run(self):
logger.debug("remove %s force=%s yes=%s", self.docker_object, self.arguments.force,
self.arguments.yes)
if not self.arguments.yes and not self.ui.yolo:
logger.debug("we need confirmation from user")
cmd = "rm -y" if not self.arguments.force else "rm -y -f"
self.ui.run_command(
"prompt prompt-text=\"You are about to remove %s %s, type enter to continue: \" "
"initial-text=\"%s\"" % (
self.docker_object.pretty_object_type.lower(),