summaryrefslogtreecommitdiff
path: root/contrib/python/podman/test/test_containers.py
diff options
context:
space:
mode:
Diffstat (limited to 'contrib/python/podman/test/test_containers.py')
-rw-r--r--contrib/python/podman/test/test_containers.py244
1 files changed, 0 insertions, 244 deletions
diff --git a/contrib/python/podman/test/test_containers.py b/contrib/python/podman/test/test_containers.py
deleted file mode 100644
index 5201956e8..000000000
--- a/contrib/python/podman/test/test_containers.py
+++ /dev/null
@@ -1,244 +0,0 @@
-import os
-import signal
-import unittest
-from pathlib import Path
-from test.podman_testcase import PodmanTestCase
-from test.retry_decorator import retry
-
-import podman
-
-
-class TestContainers(PodmanTestCase):
- @classmethod
- def setUpClass(cls):
- super().setUpClass()
-
- @classmethod
- def tearDownClass(cls):
- super().tearDownClass()
-
- def setUp(self):
- self.tmpdir = os.environ['TMPDIR']
- self.host = os.environ['PODMAN_HOST']
-
- self.pclient = podman.Client(self.host)
- self.loadCache()
-
- def tearDown(self):
- pass
-
- def loadCache(self):
- self.containers = list(self.pclient.containers.list())
-
- self.alpine_ctnr = next(
- iter([c for c in self.containers if 'alpine' in c['image']] or []),
- None)
-
- if self.alpine_ctnr and self.alpine_ctnr.status != 'running':
- self.alpine_ctnr.start()
-
- def test_list(self):
- self.assertGreaterEqual(len(self.containers), 2)
- self.assertIsNotNone(self.alpine_ctnr)
- self.assertIn('alpine', self.alpine_ctnr.image)
-
- def test_delete_stopped(self):
- before = len(self.containers)
-
- self.alpine_ctnr.stop()
- target = self.alpine_ctnr.id
- actual = self.pclient.containers.delete_stopped()
- self.assertIn(target, actual)
-
- self.loadCache()
- after = len(self.containers)
-
- self.assertLess(after, before)
- TestContainers.setUpClass()
-
- 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 test_attach(self):
- # StringIO does not support fileno() so we had to go old school
- input = os.path.join(self.tmpdir, 'test_attach.stdin')
- output = os.path.join(self.tmpdir, 'test_attach.stdout')
-
- with open(input, 'w+') as mock_in, open(output, 'w+') as mock_out:
- # double quote is indeed in the expected place
- mock_in.write('echo H"ello, World"; exit\n')
- mock_in.seek(0, 0)
-
- ctnr = self.pclient.images.get(self.alpine_ctnr.image).container(
- detach=True, tty=True)
- ctnr.attach(stdin=mock_in.fileno(), stdout=mock_out.fileno())
- ctnr.start()
-
- mock_out.flush()
- mock_out.seek(0, 0)
- output = mock_out.read()
- self.assertIn('Hello', output)
-
- ctnr.remove(force=True)
-
- def test_processes(self):
- actual = list(self.alpine_ctnr.processes())
- self.assertGreaterEqual(len(actual), 2)
-
- def test_start_stop_wait(self):
- ctnr = self.alpine_ctnr.stop()
- self.assertFalse(ctnr['running'])
-
- ctnr.start()
- self.assertTrue(ctnr.running)
-
- ctnr.stop()
- self.assertFalse(ctnr['containerrunning'])
-
- actual = ctnr.wait()
- self.assertGreaterEqual(actual, 0)
-
- def test_changes(self):
- actual = self.alpine_ctnr.changes()
-
- self.assertListEqual(
- sorted(['changed', 'added', 'deleted']), sorted(
- list(actual.keys())))
-
- # TODO: brittle, depends on knowing history of ctnr
- self.assertGreaterEqual(len(actual['changed']), 0)
- self.assertGreaterEqual(len(actual['added']), 0)
- self.assertEqual(len(actual['deleted']), 0)
-
- def test_kill(self):
- self.assertTrue(self.alpine_ctnr.running)
- ctnr = self.alpine_ctnr.kill(signal.SIGKILL)
- self.assertFalse(ctnr.running)
-
- def test_inspect(self):
- actual = self.alpine_ctnr.inspect()
- self.assertEqual(actual.id, self.alpine_ctnr.id)
- # TODO: Datetime values from inspect missing offset in CI instance
- # self.assertEqual(
- # datetime_parse(actual.created),
- # datetime_parse(self.alpine_ctnr.createdat))
-
- def test_export(self):
- target = os.path.join(self.tmpdir, 'alpine_export_ctnr.tar')
-
- actual = self.alpine_ctnr.export(target)
- self.assertEqual(actual, target)
- self.assertTrue(os.path.isfile(target))
- self.assertGreater(os.path.getsize(target), 0)
-
- def test_commit(self):
- # TODO: Test for STOPSIGNAL when supported by OCI
- # TODO: Test for message when supported by OCI
- details = self.pclient.images.get(self.alpine_ctnr.image).inspect()
- changes = ['ENV=' + i for i in details.config['env']]
- changes.append('CMD=/usr/bin/zsh')
- changes.append('ENTRYPOINT=/bin/sh date')
- changes.append('ENV=TEST=test_containers.TestContainers.test_commit')
- changes.append('EXPOSE=80')
- changes.append('EXPOSE=8888')
- changes.append('LABEL=unittest=test_commit')
- changes.append('USER=bozo:circus')
- changes.append('VOLUME=/data')
- changes.append('WORKDIR=/data/application')
-
- id = self.alpine_ctnr.commit(
- 'alpine3', author='Bozo the clown', change=changes, pause=True)
- img = self.pclient.images.get(id)
- self.assertIsNotNone(img)
-
- details = img.inspect()
- self.assertEqual(details.author, 'Bozo the clown')
- self.assertListEqual(['/usr/bin/zsh'], details.config['cmd'])
- self.assertListEqual(['/bin/sh date'],
- details.config['entrypoint'])
- self.assertIn('TEST=test_containers.TestContainers.test_commit',
- details.config['env'])
- self.assertTrue(
- [e for e in details.config['env'] if 'PATH=' in e])
- self.assertDictEqual({
- '80': {},
- '8888': {},
- }, details.config['exposedports'])
- self.assertDictEqual({'unittest': 'test_commit'}, details.labels)
- self.assertEqual('bozo:circus', details.config['user'])
- self.assertEqual({'/data': {}}, details.config['volumes'])
- self.assertEqual('/data/application',
- details.config['workingdir'])
-
- def test_remove(self):
- before = len(self.containers)
-
- with self.assertRaises(podman.ErrorOccurred):
- self.alpine_ctnr.remove()
-
- self.assertEqual(
- self.alpine_ctnr.id, self.alpine_ctnr.remove(force=True))
- self.loadCache()
- after = len(self.containers)
-
- self.assertLess(after, before)
- TestContainers.setUpClass()
-
- def test_restart(self):
- self.assertTrue(self.alpine_ctnr.running)
- before = self.alpine_ctnr.runningfor
-
- ctnr = self.alpine_ctnr.restart()
- self.assertTrue(ctnr.running)
-
- after = self.alpine_ctnr.runningfor
-
- # TODO: restore check when restart zeros counter
- # self.assertLess(after, before)
-
- def test_rename(self):
- with self.assertRaisesNotImplemented():
- self.alpine_ctnr.rename('new_alpine')
-
- def test_resize_tty(self):
- with self.assertRaisesNotImplemented():
- self.alpine_ctnr.resize_tty(132, 43)
-
- def test_pause_unpause(self):
- self.assertTrue(self.alpine_ctnr.running)
-
- ctnr = self.alpine_ctnr.pause()
- self.assertEqual(ctnr.status, 'paused')
-
- ctnr = self.alpine_ctnr.unpause()
- self.assertTrue(ctnr.running)
- self.assertTrue(ctnr.status, 'running')
-
- # creating cgoups can be flakey
- @retry(podman.libs.errors.ErrorOccurred, tries=4, delay=2, print_=print)
- def test_stats(self):
- try:
- self.assertTrue(self.alpine_ctnr.running)
-
- actual = self.alpine_ctnr.stats()
- self.assertEqual(self.alpine_ctnr.id, actual.id)
- self.assertEqual(self.alpine_ctnr.names, actual.name)
- except Exception:
- info = Path('/proc/self/mountinfo')
- with info.open() as fd:
- print('{} {}'.format(self.alpine_ctnr.id, info))
- print(fd.read())
-
- def test_logs(self):
- self.assertTrue(self.alpine_ctnr.running)
- actual = list(self.alpine_ctnr.logs())
- self.assertIsNotNone(actual)
-
-
-if __name__ == '__main__':
- unittest.main()