import io import os import random import tarfile import unittest import tempfile try: # Python >= 3.10 from collections.abc import Iterator except ImportError: # Python < 3.10 from collections.abc import Iterator import podman.tests.integration.base as base from podman import PodmanClient from podman.domain.containers import Container from podman.domain.images import Image from podman.errors import NotFound # @unittest.skipIf(os.geteuid() != 0, 'Skipping, not running as root') class ContainersIntegrationTest(base.IntegrationTest): """Containers Integration tests.""" def setUp(self): super().setUp() self.client = PodmanClient(base_url=self.socket_uri) self.addCleanup(self.client.close) self.alpine_image = self.client.images.pull("quay.io/libpod/alpine", tag="latest") # TODO should this use podman binary instead? for container in self.client.containers.list(): container.remove(force=True) @unittest.skipUnless(base.HAVE_FUSE, 'This test requires a working fuse setup.') def test_container_crud(self): """Test Container CRUD.""" random_string = f"{random.getrandbits(160):x}" with self.subTest("Create from Alpine Image"): container = self.client.containers.create( self.alpine_image, command=["echo", random_string], ports={'2222/tcp': 3333, 2244: 3344}, ) self.assertIsInstance(container, Container) self.assertGreater(len(container.attrs), 0) self.assertIsNotNone(container.id) self.assertIsNotNone(container.name) self.assertIsInstance(container.image, Image) self.assertTrue(self.client.containers.exists(container.id)) self.assertIn("quay.io/libpod/alpine:latest", container.image.tags) with self.subTest("Inspect Container"): actual = self.client.containers.get(container.id) self.assertIsInstance(actual, Container) self.assertEqual(actual.id, container.id) self.assertIn("2222/tcp", container.attrs["NetworkSettings"]["Ports"]) self.assertEqual( "3333", container.attrs["NetworkSettings"]["Ports"]["2222/tcp"][0]["HostPort"] ) self.assertIn("2244/tcp", container.attrs["NetworkSettings"]["Ports"]) self.assertEqual( "3344", container.attrs["NetworkSettings"]["Ports"]["2244/tcp"][0]["HostPort"] ) file_contents = b"This is an integration test for archive." file_buffer = io.BytesIO(file_contents) tar_buffer = io.BytesIO() with tarfile.open(fileobj=tar_buffer, mode="w") as tar: info = tarfile.TarInfo("/") info.type = tarfile.DIRTYPE tar.addfile(info) info = tarfile.TarInfo("unittest") info.size = len(file_buffer.getbuffer()) tar.addfile(info, file_buffer) tarball = tar_buffer.getvalue() with self.subTest("Archive /root/unittest"): self.assertTrue(container.put_archive("/root", data=tarball)) actual, stats = container.get_archive("/root") with io.BytesIO() as fd: for chunk in actual: fd.write(chunk) fd.seek(0, 0) with tarfile.open(fileobj=fd, mode="r") as tar: contents = tar.extractfile("root/unittest").read() self.assertEqual(contents, file_contents) with self.subTest("List containers --all"): containers = self.client.containers.list(all=True) self.assertGreater(len(containers), 0) ids = [i.id for i in containers] self.assertIn(container.id, ids) with self.subTest("Get container's logs"): container.start() container.wait(condition="exited") logs_iter = container.logs(stream=False) self.assertIsInstance(logs_iter, Iterator) logs = list(logs_iter) self.assertIn((random_string + "\n").encode("utf-8"), logs) with self.subTest("Delete Container"): container.remove() with self.assertRaises(NotFound): self.client.containers.get(container.id) with self.subTest("Run Container"): top_ctnr = self.client.containers.run( self.alpine_image, "/usr/bin/top", name="TestRunPs", detach=True ) self.assertEqual(top_ctnr.status, "running") top_ctnr.pause() top_ctnr.reload() self.assertEqual(top_ctnr.status, "paused") top_ctnr.unpause() top_ctnr.reload() self.assertEqual(top_ctnr.status, "running") report = top_ctnr.top() # See https://github.com/containers/podman/pull/9892 for the service # side fix requires podman >= 3.2 # processes = [i.strip() for i in report["Processes"][0]] self.assertIn("/usr/bin/top", report["Processes"][0][-1]) top_ctnr.stop() top_ctnr.reload() self.assertIn(top_ctnr.status, ("exited", "stopped")) with self.subTest("Create-Init-Start Container"): top_ctnr = self.client.containers.create( self.alpine_image, ["/usr/bin/top"], name="TestInitPs", detach=True ) self.assertEqual(top_ctnr.status, "created") top_ctnr.init() top_ctnr.reload() self.assertEqual(top_ctnr.status, "initialized") top_ctnr.start() top_ctnr.reload() self.assertEqual(top_ctnr.status, "running") top_ctnr.stop() top_ctnr.reload() self.assertIn(top_ctnr.status, ("exited", "stopped")) with self.subTest("Prune Containers"): report = self.client.containers.prune() self.assertIn(top_ctnr.id, report["ContainersDeleted"]) # SpaceReclaimed is the size of the content created during the running of the container # TODO: This should probably check if the podman version is >= 4.6 (guess) self.assertGreater(report["SpaceReclaimed"], 0) with self.assertRaises(NotFound): self.client.containers.get(top_ctnr.id) @unittest.skipUnless(base.HAVE_FUSE, 'This test requires a working fuse setup.') def test_container_commit(self): """Commit new image.""" busybox = self.client.images.pull("quay.io/libpod/busybox", tag="latest") container = self.client.containers.create( busybox, command=["echo", f"{random.getrandbits(160):x}"] ) image = container.commit(repository="busybox.local", tag="unittest") self.assertIn("localhost/busybox.local:unittest", image.attrs["RepoTags"]) busybox.remove(force=True) @unittest.skipUnless(base.HAVE_FUSE, 'This test requires a working fuse setup.') def test_container_rm_anonymous_volume(self): with self.subTest("Check anonymous volume is removed"): container_file = """ FROM alpine VOLUME myvol ENV foo=bar """ tmp_file = tempfile.mktemp() file = open(tmp_file, 'w') file.write(container_file) file.close() self.client.images.build(dockerfile=tmp_file, tag="test-img", path=".") # get existing number of containers and volumes existing_containers = self.client.containers.list(all=True) existing_volumes = self.client.volumes.list() container = self.client.containers.create("test-img") container_list = self.client.containers.list(all=True) self.assertEqual(len(container_list), len(existing_containers) + 1) volume_list = self.client.volumes.list() self.assertEqual(len(volume_list), len(existing_volumes) + 1) # remove the container with v=True container.remove(v=True) container_list = self.client.containers.list(all=True) self.assertEqual(len(container_list), len(existing_containers)) volume_list = self.client.volumes.list() self.assertEqual(len(volume_list), len(existing_volumes)) @unittest.skipIf(os.uname().machine == 'riscv64', 'quay.io/libpod/alpine does not include riscv64.') def test_container_labels(self): labels = {'label1': 'value1', 'label2': 'value2'} labeled_container = self.client.containers.create(self.alpine_image, labels=labels) unlabeled_container = self.client.containers.create( self.alpine_image, ) # inspect and list have 2 different schemas so we need to verify that we can # successfully retrieve the labels on both try: # inspect schema self.assertEqual(labeled_container.labels, labels) self.assertEqual(unlabeled_container.labels, {}) # list schema for container in self.client.containers.list(all=True): if container.id == labeled_container.id: self.assertEqual(container.labels, labels) elif container.id == unlabeled_container.id: self.assertEqual(container.labels, {}) finally: labeled_container.remove(v=True) unlabeled_container.remove(v=True) if __name__ == '__main__': unittest.main()