Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_short_id():
mock()
b = DockerBackend()
operation = DockerContainer({"Status": "Up", "Id": "voodoo"}, b).top()
top_response = operation.response
pt = ProcessList(top_response)
# 24502
# \ 24542
# \ 23743
# \ 18725
# \ 18733
# \ 18743
# \ 23819
root_process = pt.get_root_process()
assert root_process.pid == "24502"
assert pt.get_parent_process(root_process) is None
p_24542 = pt.get_first_child_process(root_process)
def run(self):
if not isinstance(self.docker_object, DockerContainer):
self.ui.notify_message("No docker container specified.", level="error")
return
if not self.docker_object.running:
self.ui.notify_message("Container is not running - no ports are available.",
level="error")
return
ports = list(self.docker_object.net.ports.keys())
ips = self.docker_object.net.ips
logger.debug("ports = %s, ips = %s", ports, ips)
if not ports:
self.ui.notify_message(
"Container %r doesn't expose any ports." % self.docker_object.short_name,
def _assemble_initial_content(self):
widgets = []
for o in self.d.initial_content():
line = None
if isinstance(o, DockerImage):
image_id = urwid.Text(("image_id", o.image_id[:12]), align="left", wrap="any")
time = urwid.Text(("image_id", o.display_time_created()), align="left", wrap="any")
names = urwid.Text(("image_names", o.names or ""), align="left", wrap="clip")
line = DockerImageColumns(o, [(13, image_id), names, time])
elif isinstance(o, DockerContainer):
container_id = urwid.Text(("image_id", o.container_id[:12]), align="left", wrap="any")
time = urwid.Text(("image_id", o.display_time_created()), align="left", wrap="any")
name = urwid.Text(("image_names", o.name), align="left", wrap="clip")
command = urwid.Text(("image_names", o.command), align="left", wrap="clip")
status = urwid.Text(("image_names", o.status), align="left", wrap="clip")
line = DockerImageColumns(o, [(13, container_id), command, time, status, name])
widgets.append(urwid.AttrMap(line, 'image_id', focus_map='reversed'))
return widgets
def get_row(docker_object):
if isinstance(docker_object, DockerImage):
return get_detailed_image_row(docker_object)
elif isinstance(docker_object, DockerContainer):
return get_detailed_container_row(docker_object)
else:
raise Exception("what ")
def __init__(self, data, docker_backend, object_id=None):
super(DockerContainer, self).__init__(data, docker_backend, object_id)
self.size_root_fs = None
self.size_rw_fs = None
def keypress(self, size, key):
logger.debug("size %r, key %r", size, key)
if key == "i":
docker_object = self.get_focus()[0].original_widget.docker_object
if isinstance(docker_object, DockerImage):
inspect_data = self.d.inspect_image(docker_object.image_id)
elif isinstance(docker_object, DockerContainer):
inspect_data = self.d.inspect_container(docker_object.container_id)
else:
raise RuntimeError("wat")
rendered_json = json.dumps(inspect_data, indent=2)
self.ui.add_and_set_main_widget(ScrollableListBox(rendered_json))
return
if key == "l":
docker_object = self.get_focus()[0].original_widget.docker_object
if isinstance(docker_object, DockerContainer):
logs_data, logs_generator = self.d.logs(docker_object.container_id)
w = AsyncScrollableListBox(logs_data, logs_generator, self.ui)
self.ui.add_and_set_main_widget(w)
return
key = super(MainListBox, self).keypress(size, key)
return key
def get_containers(self, cached=True, stopped=True):
if cached is False or self._containers is None:
logger.debug("doing containers() query")
self._containers = {}
containers_reponse = repeater(self.client.containers, kwargs={"all": stopped}) or []
for c in containers_reponse:
container = DockerContainer(c, self)
self._containers[container.container_id] = container
if not stopped:
return [x for x in list(self._containers.values()) if x.running]
return list(self._containers.values())
def __init__(self, ui, docker_object, follow=False):
"""
:param docker_object: container to display logs
:param ui: ui object so we can refresh
"""
self.display_name += "({})".format(docker_object.short_name)
if isinstance(docker_object, DockerContainer):
try:
pre_message = "Getting logs for container {}...".format(docker_object.short_name)
ui.notify_message(pre_message)
if follow:
# FIXME: this is a bit race-y -- we might lose some logs with this approach
operation = docker_object.logs(follow=follow, lines=0)
static_data = docker_object.logs(follow=False).response
self.widget = AsyncScrollableListBox(operation.response, ui, static_data=static_data)
else:
operation = docker_object.logs(follow=follow)
self.widget = ScrollableListBox(ui, operation.response)
ui.remove_notification_message(pre_message)
ui.notify_widget(get_operation_notify_widget(operation, display_always=False))
except Exception as ex:
# FIXME: let's catch 404 and print that container doesn't exist
# instead of printing ugly HTTP error
def run(self):
# TODO: needs better abstraction, backend object should be tied together with frontend
# object via a new Class
if isinstance(self.docker_object, DockerImage):
buffer_class = ImageInfoBuffer
elif isinstance(self.docker_object, DockerContainer):
buffer_class = ContainerInfoBuffer
else:
self.ui.notify_message("Can't display info for '%s'" % self.docker_object,
level="error")
logger.error("unable to display info buffer for %r", self.docker_object)
return
try:
# FIXME: this try/except block should be in upper frame
self.ui.add_and_display_buffer(buffer_class(self.docker_object, self.ui))
except NotifyError as ex:
self.ui.notify_message(str(ex), level="error")
logger.error(repr(ex))