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.containerconfig['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.containerconfig['cmd']) self.assertListEqual(['/bin/sh date'], details.containerconfig['entrypoint']) self.assertIn('TEST=test_containers.TestContainers.test_commit', details.containerconfig['env']) self.assertTrue( [e for e in details.containerconfig['env'] if 'PATH=' in e]) self.assertDictEqual({ '80': {}, '8888': {}, }, details.containerconfig['exposedports']) self.assertDictEqual({'unittest': 'test_commit'}, details.labels) self.assertEqual('bozo:circus', details.containerconfig['user']) self.assertEqual({'/data': {}}, details.containerconfig['volumes']) self.assertEqual('/data/application', details.containerconfig['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()