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_get(self):
actual = self.pclient.containers.get(self.alpine_ctnr.id)
for k in ['id', 'status', 'ports']:
self.assertEqual(actual[k], self.alpine_ctnr[k])
with self.assertRaises(podman.ContainerNotFound):
self.pclient.containers.get("bozo")
def commit(self):
"""Create image from container."""
try:
try:
ctnr = self.client.containers.get(self._args.container[0])
except podman.ContainerNotFound as e:
sys.stdout.flush()
print(
'Container {} not found.'.format(e.name),
file=sys.stderr,
flush=True)
return 1
else:
ident = ctnr.commit(self.opts['image'][0], **self.opts)
print(ident)
except podman.ErrorOccurred as e:
sys.stdout.flush()
print(
'{}'.format(e.reason).capitalize(),
file=sys.stderr,
flush=True)
return 1
def logs(self):
"""Retrieve logs from containers."""
try:
ident = self._args.container[0]
try:
logging.debug('Get container "%s" logs', ident)
ctnr = self.client.containers.get(ident)
except podman.ContainerNotFound as e:
sys.stdout.flush()
print(
'Container "{}" not found'.format(e.name),
file=sys.stderr,
flush=True)
else:
if self._args.tail:
logs = iter(deque(ctnr.logs(), maxlen=self._args.tail))
else:
logs = ctnr.logs()
for line in logs:
sys.stdout.write(line)
except podman.ErrorOccurred as e:
sys.stdout.flush()
print(
def start(self):
"""Start provided containers."""
stdin = sys.stdin if self.opts['interactive'] else None
stdout = sys.stdout if self.opts['attach'] else None
try:
for ident in self._args.containers:
try:
ctnr = self.client.containers.get(ident)
ctnr.attach(
eot=self.opts['detach_keys'],
stdin=stdin,
stdout=stdout)
ctnr.start()
except podman.ContainerNotFound as e:
sys.stdout.flush()
print(
'Container "{}" not found'.format(e.name),
file=sys.stderr,
flush=True)
else:
print(ident)
except podman.ErrorOccurred as e:
sys.stdout.flush()
print(
'{}'.format(e.reason).capitalize(),
file=sys.stderr,
flush=True)
return 1
return 0
def query_model(model, identifiers=None):
"""Retrieve all (default) or given model(s)."""
objs = []
if identifiers is None:
objs.extend(model.list())
else:
try:
for ident in identifiers:
objs.append(model.get(ident))
except (
podman.PodNotFound,
podman.ImageNotFound,
podman.ContainerNotFound,
) as ex:
print(
'"{}" not found'.format(ex.name), file=sys.stderr, flush=True)
return objs
def restart(self):
"""Restart container(s)."""
try:
for ident in self._args.targets:
try:
ctnr = self.client.containers.get(ident)
logging.debug('Restarting Container %s', ctnr.id)
ctnr.restart(timeout=self._args.timeout)
print(ident)
except podman.ContainerNotFound as e:
sys.stdout.flush()
print(
'Container {} not found.'.format(e.name),
file=sys.stderr,
flush=True)
except podman.ErrorOccurred as e:
sys.stdout.flush()
print(
'{}'.format(e.reason).capitalize(),
file=sys.stderr,
flush=True)
def port(self):
"""Retrieve ports from containers."""
try:
ctnrs = []
if self._args.all:
ctnrs = self.client.containers.list()
else:
for ident in self._args.containers:
try:
ctnrs.append(self.client.containers.get(ident))
except podman.ContainerNotFound as e:
sys.stdout.flush()
print(
'Container "{}" not found'.format(e.name),
file=sys.stderr,
flush=True)
for ctnr in ctnrs:
print("{}\n{}".format(ctnr.id, ctnr.ports))
except podman.ErrorOccurred as e:
sys.stdout.flush()
print(
'{}'.format(e.reason).capitalize(),
file=sys.stderr,
flush=True)
return 1
def remove(self):
"""Remove container(s)."""
for ident in self._args.targets:
try:
ctnr = self.client.containers.get(ident)
ctnr.remove(self._args.force)
print(ident)
except podman.ContainerNotFound as e:
sys.stdout.flush()
print(
'Container {} not found.'.format(e.name),
file=sys.stderr,
flush=True)
except podman.ErrorOccurred as e:
sys.stdout.flush()
print(
'{}'.format(e.reason).capitalize(),
file=sys.stderr,
flush=True)
def pause(self):
"""Pause provided containers."""
try:
for ident in self._args.containers:
try:
ctnr = self.client.containers.get(ident)
ctnr.pause()
except podman.ContainerNotFound as e:
sys.stdout.flush()
print(
'Container "{}" not found'.format(e.name),
file=sys.stderr,
flush=True)
else:
print(ident)
except podman.ErrorOccurred as e:
sys.stdout.flush()
print(
'{}'.format(e.reason).capitalize(),
file=sys.stderr,
flush=True)
return 1
def export(self):
"""Create tarball from container filesystem."""
try:
try:
ctnr = self.client.containers.get(self._args.container[0])
except podman.ContainerNotFound as e:
sys.stdout.flush()
print(
'Container {} not found.'.format(e.name),
file=sys.stderr,
flush=True)
return 1
else:
ctnr.export(self._args.output[0])
except podman.ErrorOccurred as e:
sys.stdout.flush()
print(
'{}'.format(e.reason).capitalize(),
file=sys.stderr,
flush=True)
return 1
return 0