From 7927fe01f165bb4a3f381601d847036a3a130182 Mon Sep 17 00:00:00 2001 From: Jhon Honce Date: Mon, 1 Mar 2021 10:55:20 -0700 Subject: Refactor python tests to run against python3.9 * Introduce sub-package compat to meet packaging and import requirements * Update documenation for running tests * Add requirements.txt to improve IDE support Signed-off-by: Jhon Honce --- test/python/docker/README.md | 38 ----- test/python/docker/__init__.py | 14 +- test/python/docker/common.py | 21 --- test/python/docker/compat/README.md | 38 +++++ test/python/docker/compat/__init__.py | 0 test/python/docker/compat/common.py | 23 +++ test/python/docker/compat/constant.py | 6 + test/python/docker/compat/test_containers.py | 200 +++++++++++++++++++++++++++ test/python/docker/compat/test_images.py | 155 +++++++++++++++++++++ test/python/docker/compat/test_system.py | 68 +++++++++ test/python/docker/constant.py | 6 - test/python/docker/test_containers.py | 197 -------------------------- test/python/docker/test_images.py | 152 -------------------- test/python/docker/test_system.py | 67 --------- 14 files changed, 500 insertions(+), 485 deletions(-) delete mode 100644 test/python/docker/README.md delete mode 100644 test/python/docker/common.py create mode 100644 test/python/docker/compat/README.md create mode 100644 test/python/docker/compat/__init__.py create mode 100644 test/python/docker/compat/common.py create mode 100644 test/python/docker/compat/constant.py create mode 100644 test/python/docker/compat/test_containers.py create mode 100644 test/python/docker/compat/test_images.py create mode 100644 test/python/docker/compat/test_system.py delete mode 100644 test/python/docker/constant.py delete mode 100644 test/python/docker/test_containers.py delete mode 100644 test/python/docker/test_images.py delete mode 100644 test/python/docker/test_system.py (limited to 'test/python/docker') diff --git a/test/python/docker/README.md b/test/python/docker/README.md deleted file mode 100644 index c10fd636d..000000000 --- a/test/python/docker/README.md +++ /dev/null @@ -1,38 +0,0 @@ -# Docker regression test - -Python test suite to validate Podman endpoints using docker library (aka docker-py). -See [Docker SDK for Python](https://docker-py.readthedocs.io/en/stable/index.html). - -## Running Tests - -To run the tests locally in your sandbox (Fedora 32,33): - -```shell -# dnf install python3-docker -``` - -### Run the entire test suite - -```shell -# python3 -m unittest discover test/python/docker -``` - -Passing the -v option to your test script will instruct unittest.main() to enable a higher level of verbosity, and produce detailed output: - -```shell -# python3 -m unittest -v discover test/python/docker -``` - -### Run a specific test class - -```shell -# cd test/python/docker -# python3 -m unittest -v tests.test_images -``` - -### Run a specific test within the test class - -```shell -# cd test/python/docker -# python3 -m unittest tests.test_images.TestImages.test_import_image -``` diff --git a/test/python/docker/__init__.py b/test/python/docker/__init__.py index 351834316..da5630eac 100644 --- a/test/python/docker/__init__.py +++ b/test/python/docker/__init__.py @@ -8,7 +8,7 @@ import tempfile from docker import DockerClient -from test.python.docker import constant +from .compat import constant class Podman(object): @@ -39,7 +39,9 @@ class Podman(object): self.cmd.append("--root=" + os.path.join(self.anchor_directory, "crio")) self.cmd.append("--runroot=" + os.path.join(self.anchor_directory, "crio-run")) - os.environ["REGISTRIES_CONFIG_PATH"] = os.path.join(self.anchor_directory, "registry.conf") + os.environ["REGISTRIES_CONFIG_PATH"] = os.path.join( + self.anchor_directory, "registry.conf" + ) p = configparser.ConfigParser() p.read_dict( { @@ -51,10 +53,14 @@ class Podman(object): with open(os.environ["REGISTRIES_CONFIG_PATH"], "w") as w: p.write(w) - os.environ["CNI_CONFIG_PATH"] = os.path.join(self.anchor_directory, "cni", "net.d") + os.environ["CNI_CONFIG_PATH"] = os.path.join( + self.anchor_directory, "cni", "net.d" + ) os.makedirs(os.environ["CNI_CONFIG_PATH"], exist_ok=True) self.cmd.append("--cni-config-dir=" + os.environ["CNI_CONFIG_PATH"]) - cni_cfg = os.path.join(os.environ["CNI_CONFIG_PATH"], "87-podman-bridge.conflist") + cni_cfg = os.path.join( + os.environ["CNI_CONFIG_PATH"], "87-podman-bridge.conflist" + ) # json decoded and encoded to ensure legal json buf = json.loads( """ diff --git a/test/python/docker/common.py b/test/python/docker/common.py deleted file mode 100644 index 11f512495..000000000 --- a/test/python/docker/common.py +++ /dev/null @@ -1,21 +0,0 @@ -from docker import DockerClient - -from test.python.docker import constant - - -def run_top_container(client: DockerClient): - c = client.containers.create(constant.ALPINE, command="top", detach=True, tty=True, name="top") - c.start() - return c.id - - -def remove_all_containers(client: DockerClient): - for ctnr in client.containers.list(all=True): - ctnr.remove(force=True) - - -def remove_all_images(client: DockerClient): - for img in client.images.list(): - # FIXME should DELETE /images accept the sha256: prefix? - id_ = img.id.removeprefix("sha256:") - client.images.remove(id_, force=True) diff --git a/test/python/docker/compat/README.md b/test/python/docker/compat/README.md new file mode 100644 index 000000000..50796d66b --- /dev/null +++ b/test/python/docker/compat/README.md @@ -0,0 +1,38 @@ +# Docker regression test + +Python test suite to validate Podman endpoints using docker library (aka docker-py). +See [Docker SDK for Python](https://docker-py.readthedocs.io/en/stable/index.html). + +## Running Tests + +To run the tests locally in your sandbox (Fedora 32,33): + +```shell +# dnf install python3-docker +``` + +### Run the entire test suite + +All commands are run from the root of the repository. + +```shell +# python3 -m unittest discover -s test/python/docker +``` + +Passing the -v option to your test script will instruct unittest.main() to enable a higher level of verbosity, and produce detailed output: + +```shell +# python3 -m unittest -v discover -s test/python/docker +``` + +### Run a specific test class + +```shell +# python3 -m unittest -v test.python.docker.compat.test_images.TestImages +``` + +### Run a specific test within the test class + +```shell +# python3 -m unittest test.python.docker.compat.test_images.TestImages.test_tag_valid_image +``` diff --git a/test/python/docker/compat/__init__.py b/test/python/docker/compat/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/test/python/docker/compat/common.py b/test/python/docker/compat/common.py new file mode 100644 index 000000000..bdc67c287 --- /dev/null +++ b/test/python/docker/compat/common.py @@ -0,0 +1,23 @@ +from docker import DockerClient + +from test.python.docker.compat import constant + + +def run_top_container(client: DockerClient): + c = client.containers.create( + constant.ALPINE, command="top", detach=True, tty=True, name="top" + ) + c.start() + return c.id + + +def remove_all_containers(client: DockerClient): + for ctnr in client.containers.list(all=True): + ctnr.remove(force=True) + + +def remove_all_images(client: DockerClient): + for img in client.images.list(): + # FIXME should DELETE /images accept the sha256: prefix? + id_ = img.id.removeprefix("sha256:") + client.images.remove(id_, force=True) diff --git a/test/python/docker/compat/constant.py b/test/python/docker/compat/constant.py new file mode 100644 index 000000000..892293c97 --- /dev/null +++ b/test/python/docker/compat/constant.py @@ -0,0 +1,6 @@ +ALPINE = "quay.io/libpod/alpine:latest" +ALPINE_SHORTNAME = "alpine" +ALPINE_TARBALL = "alpine.tar" +BB = "quay.io/libpod/busybox:latest" +NGINX = "quay.io/libpod/alpine_nginx:latest" +infra = "k8s.gcr.io/pause:3.2" diff --git a/test/python/docker/compat/test_containers.py b/test/python/docker/compat/test_containers.py new file mode 100644 index 000000000..be70efa67 --- /dev/null +++ b/test/python/docker/compat/test_containers.py @@ -0,0 +1,200 @@ +import subprocess +import sys +import time +import unittest + +from docker import DockerClient, errors + +from test.python.docker import Podman +from test.python.docker.compat import common, constant + + +class TestContainers(unittest.TestCase): + podman = None # initialized podman configuration for tests + service = None # podman service instance + topContainerId = "" + + def setUp(self): + super().setUp() + self.client = DockerClient(base_url="tcp://127.0.0.1:8080", timeout=15) + TestContainers.podman.restore_image_from_cache(self.client) + TestContainers.topContainerId = common.run_top_container(self.client) + self.assertIsNotNone(TestContainers.topContainerId) + + def tearDown(self): + common.remove_all_containers(self.client) + common.remove_all_images(self.client) + self.client.close() + return super().tearDown() + + @classmethod + def setUpClass(cls): + super().setUpClass() + TestContainers.podman = Podman() + TestContainers.service = TestContainers.podman.open( + "system", "service", "tcp:127.0.0.1:8080", "--time=0" + ) + # give the service some time to be ready... + time.sleep(2) + + rc = TestContainers.service.poll() + if rc is not None: + raise subprocess.CalledProcessError(rc, "podman system service") + + @classmethod + def tearDownClass(cls): + TestContainers.service.terminate() + stdout, stderr = TestContainers.service.communicate(timeout=0.5) + if stdout: + sys.stdout.write("\nContainers Service Stdout:\n" + stdout.decode("utf-8")) + if stderr: + sys.stderr.write("\nContainers Service Stderr:\n" + stderr.decode("utf-8")) + + TestContainers.podman.tear_down() + return super().tearDownClass() + + def test_create_container(self): + # Run a container with detach mode + self.client.containers.create(image="alpine", detach=True) + self.assertEqual(len(self.client.containers.list(all=True)), 2) + + def test_create_network(self): + net = self.client.networks.create("testNetwork", driver="bridge") + ctnr = self.client.containers.create(image="alpine", detach=True) + + # TODO fix when ready + # This test will not work until all connect|disconnect + # code is fixed. + # net.connect(ctnr) + + # nets = self.client.networks.list(greedy=True) + # self.assertGreaterEqual(len(nets), 1) + + # TODO fix endpoint to include containers + # for n in nets: + # if n.id == "testNetwork": + # self.assertEqual(ctnr.id, n.containers) + # self.assertTrue(False, "testNetwork not found") + + def test_start_container(self): + # Podman docs says it should give a 304 but returns with no response + # # Start a already started container should return 304 + # response = self.client.api.start(container=TestContainers.topContainerId) + # self.assertEqual(error.exception.response.status_code, 304) + + # Create a new container and validate the count + self.client.containers.create(image=constant.ALPINE, name="container2") + containers = self.client.containers.list(all=True) + self.assertEqual(len(containers), 2) + + def test_start_container_with_random_port_bind(self): + container = self.client.containers.create( + image=constant.ALPINE, + name="containerWithRandomBind", + ports={"1234/tcp": None}, + ) + containers = self.client.containers.list(all=True) + self.assertTrue(container in containers) + + def test_stop_container(self): + top = self.client.containers.get(TestContainers.topContainerId) + self.assertEqual(top.status, "running") + + # Stop a running container and validate the state + top.stop() + top.reload() + self.assertIn(top.status, ("stopped", "exited")) + + def test_kill_container(self): + top = self.client.containers.get(TestContainers.topContainerId) + self.assertEqual(top.status, "running") + + # Kill a running container and validate the state + top.kill() + top.reload() + self.assertIn(top.status, ("stopped", "exited")) + + def test_restart_container(self): + # Validate the container state + top = self.client.containers.get(TestContainers.topContainerId) + top.stop() + top.reload() + self.assertIn(top.status, ("stopped", "exited")) + + # restart a running container and validate the state + top.restart() + top.reload() + self.assertEqual(top.status, "running") + + def test_remove_container(self): + # Remove container by ID with force + top = self.client.containers.get(TestContainers.topContainerId) + top.remove(force=True) + self.assertEqual(len(self.client.containers.list()), 0) + + def test_remove_container_without_force(self): + # Validate current container count + self.assertTrue(len(self.client.containers.list()), 1) + + # Remove running container should throw error + top = self.client.containers.get(TestContainers.topContainerId) + with self.assertRaises(errors.APIError) as error: + top.remove() + self.assertEqual(error.exception.response.status_code, 500) + + # Remove container by ID without force + top.stop() + top.remove() + self.assertEqual(len(self.client.containers.list()), 0) + + def test_pause_container(self): + # Validate the container state + top = self.client.containers.get(TestContainers.topContainerId) + self.assertEqual(top.status, "running") + + # Pause a running container and validate the state + top.pause() + top.reload() + self.assertEqual(top.status, "paused") + + def test_pause_stopped_container(self): + # Stop the container + top = self.client.containers.get(TestContainers.topContainerId) + top.stop() + + # Pause exited container should throw error + with self.assertRaises(errors.APIError) as error: + top.pause() + self.assertEqual(error.exception.response.status_code, 500) + + def test_unpause_container(self): + top = self.client.containers.get(TestContainers.topContainerId) + + # Validate the container state + top.pause() + top.reload() + self.assertEqual(top.status, "paused") + + # Pause a running container and validate the state + top.unpause() + top.reload() + self.assertEqual(top.status, "running") + + def test_list_container(self): + # Add container and validate the count + self.client.containers.create(image="alpine", detach=True) + containers = self.client.containers.list(all=True) + self.assertEqual(len(containers), 2) + + def test_filters(self): + self.skipTest("TODO Endpoint does not yet support filters") + + # List container with filter by id + filters = {"id": TestContainers.topContainerId} + ctnrs = self.client.containers.list(all=True, filters=filters) + self.assertEqual(len(ctnrs), 1) + + # List container with filter by name + filters = {"name": "top"} + ctnrs = self.client.containers.list(all=True, filters=filters) + self.assertEqual(len(ctnrs), 1) diff --git a/test/python/docker/compat/test_images.py b/test/python/docker/compat/test_images.py new file mode 100644 index 000000000..842e38f31 --- /dev/null +++ b/test/python/docker/compat/test_images.py @@ -0,0 +1,155 @@ +import collections +import os +import subprocess +import sys +import time +import unittest + +from docker import DockerClient, errors + +from test.python.docker import Podman +from test.python.docker.compat import common, constant + + +class TestImages(unittest.TestCase): + podman = None # initialized podman configuration for tests + service = None # podman service instance + + def setUp(self): + super().setUp() + self.client = DockerClient(base_url="tcp://127.0.0.1:8080", timeout=15) + + TestImages.podman.restore_image_from_cache(self.client) + + def tearDown(self): + common.remove_all_images(self.client) + self.client.close() + return super().tearDown() + + @classmethod + def setUpClass(cls): + super().setUpClass() + TestImages.podman = Podman() + TestImages.service = TestImages.podman.open( + "system", "service", "tcp:127.0.0.1:8080", "--time=0" + ) + # give the service some time to be ready... + time.sleep(2) + + returncode = TestImages.service.poll() + if returncode is not None: + raise subprocess.CalledProcessError(returncode, "podman system service") + + @classmethod + def tearDownClass(cls): + TestImages.service.terminate() + stdout, stderr = TestImages.service.communicate(timeout=0.5) + if stdout: + sys.stdout.write("\nImages Service Stdout:\n" + stdout.decode("utf-8")) + if stderr: + sys.stderr.write("\nImAges Service Stderr:\n" + stderr.decode("utf-8")) + + TestImages.podman.tear_down() + return super().tearDownClass() + + def test_tag_valid_image(self): + """Validates if the image is tagged successfully""" + alpine = self.client.images.get(constant.ALPINE) + self.assertTrue(alpine.tag("demo", constant.ALPINE_SHORTNAME)) + + alpine = self.client.images.get(constant.ALPINE) + for t in alpine.tags: + self.assertIn("alpine", t) + + # @unittest.skip("doesn't work now") + def test_retag_valid_image(self): + """Validates if name updates when the image is retagged""" + alpine = self.client.images.get(constant.ALPINE) + self.assertTrue(alpine.tag("demo", "rename")) + + alpine = self.client.images.get(constant.ALPINE) + self.assertNotIn("demo:test", alpine.tags) + + def test_list_images(self): + """List images""" + self.assertEqual(len(self.client.images.list()), 1) + + # Add more images + self.client.images.pull(constant.BB) + self.assertEqual(len(self.client.images.list()), 2) + + # List images with filter + self.assertEqual( + len(self.client.images.list(filters={"reference": "alpine"})), 1 + ) + + def test_search_image(self): + """Search for image""" + for r in self.client.images.search("alpine"): + self.assertIn("alpine", r["Name"]) + + def test_search_bogus_image(self): + """Search for bogus image should throw exception""" + try: + r = self.client.images.search("bogus/bogus") + except: + return + self.assertTrue(len(r) == 0) + + def test_remove_image(self): + """Remove image""" + # Check for error with wrong image name + with self.assertRaises(errors.NotFound): + self.client.images.remove("dummy") + self.assertEqual(len(self.client.images.list()), 1) + + self.client.images.remove(constant.ALPINE) + self.assertEqual(len(self.client.images.list()), 0) + + def test_image_history(self): + """Image history""" + img = self.client.images.get(constant.ALPINE) + history = img.history() + image_id = img.id[7:] if img.id.startswith("sha256:") else img.id + + found = False + for change in history: + found |= image_id in change.values() + self.assertTrue(found, f"image id {image_id} not found in history") + + def test_get_image_exists_not(self): + """Negative test for get image""" + with self.assertRaises(errors.NotFound): + response = self.client.images.get("image_does_not_exists") + collections.deque(response) + + def test_save_image(self): + """Export Image""" + image = self.client.images.pull(constant.BB) + + file = os.path.join(TestImages.podman.image_cache, "busybox.tar") + with open(file, mode="wb") as tarball: + for frame in image.save(named=True): + tarball.write(frame) + sz = os.path.getsize(file) + self.assertGreater(sz, 0) + + def test_load_image(self): + """Import|Load Image""" + self.assertEqual(len(self.client.images.list()), 1) + + image = self.client.images.pull(constant.BB) + file = os.path.join(TestImages.podman.image_cache, "busybox.tar") + with open(file, mode="wb") as tarball: + for frame in image.save(): + tarball.write(frame) + + with open(file, mode="rb") as saved: + _ = self.client.images.load(saved) + + self.assertEqual(len(self.client.images.list()), 2) + + +if __name__ == "__main__": + # Setup temporary space + unittest.main() diff --git a/test/python/docker/compat/test_system.py b/test/python/docker/compat/test_system.py new file mode 100644 index 000000000..131b18991 --- /dev/null +++ b/test/python/docker/compat/test_system.py @@ -0,0 +1,68 @@ +import subprocess +import sys +import time +import unittest + +from docker import DockerClient + +from test.python.docker import Podman, constant +from test.python.docker.compat import common + + +class TestSystem(unittest.TestCase): + podman = None # initialized podman configuration for tests + service = None # podman service instance + topContainerId = "" + + def setUp(self): + super().setUp() + self.client = DockerClient(base_url="tcp://127.0.0.1:8080", timeout=15) + + TestSystem.podman.restore_image_from_cache(self.client) + TestSystem.topContainerId = common.run_top_container(self.client) + + def tearDown(self): + common.remove_all_containers(self.client) + common.remove_all_images(self.client) + self.client.close() + return super().tearDown() + + @classmethod + def setUpClass(cls): + super().setUpClass() + TestSystem.podman = Podman() + TestSystem.service = TestSystem.podman.open( + "system", "service", "tcp:127.0.0.1:8080", "--time=0" + ) + # give the service some time to be ready... + time.sleep(2) + + returncode = TestSystem.service.poll() + if returncode is not None: + raise subprocess.CalledProcessError(returncode, "podman system service") + + @classmethod + def tearDownClass(cls): + TestSystem.service.terminate() + stdout, stderr = TestSystem.service.communicate(timeout=0.5) + if stdout: + sys.stdout.write("\nImages Service Stdout:\n" + stdout.decode("utf-8")) + if stderr: + sys.stderr.write("\nImAges Service Stderr:\n" + stderr.decode("utf-8")) + + TestSystem.podman.tear_down() + return super().tearDownClass() + + def test_Info(self): + self.assertIsNotNone(self.client.info()) + + def test_info_container_details(self): + info = self.client.info() + self.assertEqual(info["Containers"], 1) + self.client.containers.create(image=constant.ALPINE) + info = self.client.info() + self.assertEqual(info["Containers"], 2) + + def test_version(self): + version = self.client.version() + self.assertIsNotNone(version["Platform"]["Name"]) diff --git a/test/python/docker/constant.py b/test/python/docker/constant.py deleted file mode 100644 index 892293c97..000000000 --- a/test/python/docker/constant.py +++ /dev/null @@ -1,6 +0,0 @@ -ALPINE = "quay.io/libpod/alpine:latest" -ALPINE_SHORTNAME = "alpine" -ALPINE_TARBALL = "alpine.tar" -BB = "quay.io/libpod/busybox:latest" -NGINX = "quay.io/libpod/alpine_nginx:latest" -infra = "k8s.gcr.io/pause:3.2" diff --git a/test/python/docker/test_containers.py b/test/python/docker/test_containers.py deleted file mode 100644 index 337cacd5c..000000000 --- a/test/python/docker/test_containers.py +++ /dev/null @@ -1,197 +0,0 @@ -import subprocess -import sys -import time -import unittest - -from docker import DockerClient, errors - -from test.python.docker import Podman, common, constant - - -class TestContainers(unittest.TestCase): - podman = None # initialized podman configuration for tests - service = None # podman service instance - topContainerId = "" - - def setUp(self): - super().setUp() - self.client = DockerClient(base_url="tcp://127.0.0.1:8080", timeout=15) - TestContainers.podman.restore_image_from_cache(self.client) - TestContainers.topContainerId = common.run_top_container(self.client) - self.assertIsNotNone(TestContainers.topContainerId) - - def tearDown(self): - common.remove_all_containers(self.client) - common.remove_all_images(self.client) - self.client.close() - return super().tearDown() - - @classmethod - def setUpClass(cls): - super().setUpClass() - TestContainers.podman = Podman() - TestContainers.service = TestContainers.podman.open( - "system", "service", "tcp:127.0.0.1:8080", "--time=0" - ) - # give the service some time to be ready... - time.sleep(2) - - rc = TestContainers.service.poll() - if rc is not None: - raise subprocess.CalledProcessError(rc, "podman system service") - - @classmethod - def tearDownClass(cls): - TestContainers.service.terminate() - stdout, stderr = TestContainers.service.communicate(timeout=0.5) - if stdout: - sys.stdout.write("\nContainers Service Stdout:\n" + stdout.decode("utf-8")) - if stderr: - sys.stderr.write("\nContainers Service Stderr:\n" + stderr.decode("utf-8")) - - TestContainers.podman.tear_down() - return super().tearDownClass() - - def test_create_container(self): - # Run a container with detach mode - self.client.containers.create(image="alpine", detach=True) - self.assertEqual(len(self.client.containers.list(all=True)), 2) - - def test_create_network(self): - net = self.client.networks.create("testNetwork", driver="bridge") - ctnr = self.client.containers.create(image="alpine", detach=True) - - # TODO fix when ready - # This test will not work until all connect|disconnect - # code is fixed. - # net.connect(ctnr) - - # nets = self.client.networks.list(greedy=True) - # self.assertGreaterEqual(len(nets), 1) - - # TODO fix endpoint to include containers - # for n in nets: - # if n.id == "testNetwork": - # self.assertEqual(ctnr.id, n.containers) - # self.assertTrue(False, "testNetwork not found") - - def test_start_container(self): - # Podman docs says it should give a 304 but returns with no response - # # Start a already started container should return 304 - # response = self.client.api.start(container=TestContainers.topContainerId) - # self.assertEqual(error.exception.response.status_code, 304) - - # Create a new container and validate the count - self.client.containers.create(image=constant.ALPINE, name="container2") - containers = self.client.containers.list(all=True) - self.assertEqual(len(containers), 2) - - def test_start_container_with_random_port_bind(self): - container = self.client.containers.create(image=constant.ALPINE, - name="containerWithRandomBind", - ports={'1234/tcp': None}) - containers = self.client.containers.list(all=True) - self.assertTrue(container in containers) - - def test_stop_container(self): - top = self.client.containers.get(TestContainers.topContainerId) - self.assertEqual(top.status, "running") - - # Stop a running container and validate the state - top.stop() - top.reload() - self.assertIn(top.status, ("stopped", "exited")) - - def test_kill_container(self): - top = self.client.containers.get(TestContainers.topContainerId) - self.assertEqual(top.status, "running") - - # Kill a running container and validate the state - top.kill() - top.reload() - self.assertIn(top.status, ("stopped", "exited")) - - def test_restart_container(self): - # Validate the container state - top = self.client.containers.get(TestContainers.topContainerId) - top.stop() - top.reload() - self.assertIn(top.status, ("stopped", "exited")) - - # restart a running container and validate the state - top.restart() - top.reload() - self.assertEqual(top.status, "running") - - def test_remove_container(self): - # Remove container by ID with force - top = self.client.containers.get(TestContainers.topContainerId) - top.remove(force=True) - self.assertEqual(len(self.client.containers.list()), 0) - - def test_remove_container_without_force(self): - # Validate current container count - self.assertTrue(len(self.client.containers.list()), 1) - - # Remove running container should throw error - top = self.client.containers.get(TestContainers.topContainerId) - with self.assertRaises(errors.APIError) as error: - top.remove() - self.assertEqual(error.exception.response.status_code, 500) - - # Remove container by ID without force - top.stop() - top.remove() - self.assertEqual(len(self.client.containers.list()), 0) - - def test_pause_container(self): - # Validate the container state - top = self.client.containers.get(TestContainers.topContainerId) - self.assertEqual(top.status, "running") - - # Pause a running container and validate the state - top.pause() - top.reload() - self.assertEqual(top.status, "paused") - - def test_pause_stopped_container(self): - # Stop the container - top = self.client.containers.get(TestContainers.topContainerId) - top.stop() - - # Pause exited container should throw error - with self.assertRaises(errors.APIError) as error: - top.pause() - self.assertEqual(error.exception.response.status_code, 500) - - def test_unpause_container(self): - top = self.client.containers.get(TestContainers.topContainerId) - - # Validate the container state - top.pause() - top.reload() - self.assertEqual(top.status, "paused") - - # Pause a running container and validate the state - top.unpause() - top.reload() - self.assertEqual(top.status, "running") - - def test_list_container(self): - # Add container and validate the count - self.client.containers.create(image="alpine", detach=True) - containers = self.client.containers.list(all=True) - self.assertEqual(len(containers), 2) - - def test_filters(self): - self.skipTest("TODO Endpoint does not yet support filters") - - # List container with filter by id - filters = {"id": TestContainers.topContainerId} - ctnrs = self.client.containers.list(all=True, filters=filters) - self.assertEqual(len(ctnrs), 1) - - # List container with filter by name - filters = {"name": "top"} - ctnrs = self.client.containers.list(all=True, filters=filters) - self.assertEqual(len(ctnrs), 1) diff --git a/test/python/docker/test_images.py b/test/python/docker/test_images.py deleted file mode 100644 index f2b6a5190..000000000 --- a/test/python/docker/test_images.py +++ /dev/null @@ -1,152 +0,0 @@ -import collections -import os -import subprocess -import sys -import time -import unittest - -from docker import DockerClient, errors - -from test.python.docker import Podman, common, constant - - -class TestImages(unittest.TestCase): - podman = None # initialized podman configuration for tests - service = None # podman service instance - - def setUp(self): - super().setUp() - self.client = DockerClient(base_url="tcp://127.0.0.1:8080", timeout=15) - - TestImages.podman.restore_image_from_cache(self.client) - - def tearDown(self): - common.remove_all_images(self.client) - self.client.close() - return super().tearDown() - - @classmethod - def setUpClass(cls): - super().setUpClass() - TestImages.podman = Podman() - TestImages.service = TestImages.podman.open( - "system", "service", "tcp:127.0.0.1:8080", "--time=0" - ) - # give the service some time to be ready... - time.sleep(2) - - returncode = TestImages.service.poll() - if returncode is not None: - raise subprocess.CalledProcessError(returncode, "podman system service") - - @classmethod - def tearDownClass(cls): - TestImages.service.terminate() - stdout, stderr = TestImages.service.communicate(timeout=0.5) - if stdout: - sys.stdout.write("\nImages Service Stdout:\n" + stdout.decode("utf-8")) - if stderr: - sys.stderr.write("\nImAges Service Stderr:\n" + stderr.decode("utf-8")) - - TestImages.podman.tear_down() - return super().tearDownClass() - - def test_tag_valid_image(self): - """Validates if the image is tagged successfully""" - alpine = self.client.images.get(constant.ALPINE) - self.assertTrue(alpine.tag("demo", constant.ALPINE_SHORTNAME)) - - alpine = self.client.images.get(constant.ALPINE) - for t in alpine.tags: - self.assertIn("alpine", t) - - # @unittest.skip("doesn't work now") - def test_retag_valid_image(self): - """Validates if name updates when the image is retagged""" - alpine = self.client.images.get(constant.ALPINE) - self.assertTrue(alpine.tag("demo", "rename")) - - alpine = self.client.images.get(constant.ALPINE) - self.assertNotIn("demo:test", alpine.tags) - - def test_list_images(self): - """List images""" - self.assertEqual(len(self.client.images.list()), 1) - - # Add more images - self.client.images.pull(constant.BB) - self.assertEqual(len(self.client.images.list()), 2) - - # List images with filter - self.assertEqual(len(self.client.images.list(filters={"reference": "alpine"})), 1) - - def test_search_image(self): - """Search for image""" - for r in self.client.images.search("alpine"): - self.assertIn("alpine", r["Name"]) - - def test_search_bogus_image(self): - """Search for bogus image should throw exception""" - try: - r = self.client.images.search("bogus/bogus") - except: - return - self.assertTrue(len(r)==0) - - def test_remove_image(self): - """Remove image""" - # Check for error with wrong image name - with self.assertRaises(errors.NotFound): - self.client.images.remove("dummy") - self.assertEqual(len(self.client.images.list()), 1) - - self.client.images.remove(constant.ALPINE) - self.assertEqual(len(self.client.images.list()), 0) - - def test_image_history(self): - """Image history""" - img = self.client.images.get(constant.ALPINE) - history = img.history() - image_id = img.id[7:] if img.id.startswith("sha256:") else img.id - - found = False - for change in history: - found |= image_id in change.values() - self.assertTrue(found, f"image id {image_id} not found in history") - - def test_get_image_exists_not(self): - """Negative test for get image""" - with self.assertRaises(errors.NotFound): - response = self.client.images.get("image_does_not_exists") - collections.deque(response) - - def test_save_image(self): - """Export Image""" - image = self.client.images.pull(constant.BB) - - file = os.path.join(TestImages.podman.image_cache, "busybox.tar") - with open(file, mode="wb") as tarball: - for frame in image.save(named=True): - tarball.write(frame) - sz = os.path.getsize(file) - self.assertGreater(sz, 0) - - def test_load_image(self): - """Import|Load Image""" - self.assertEqual(len(self.client.images.list()), 1) - - image = self.client.images.pull(constant.BB) - file = os.path.join(TestImages.podman.image_cache, "busybox.tar") - with open(file, mode="wb") as tarball: - for frame in image.save(): - tarball.write(frame) - - with open(file, mode="rb") as saved: - _ = self.client.images.load(saved) - - self.assertEqual(len(self.client.images.list()), 2) - - -if __name__ == "__main__": - # Setup temporary space - unittest.main() diff --git a/test/python/docker/test_system.py b/test/python/docker/test_system.py deleted file mode 100644 index 46b90e5f6..000000000 --- a/test/python/docker/test_system.py +++ /dev/null @@ -1,67 +0,0 @@ -import subprocess -import sys -import time -import unittest - -from docker import DockerClient - -from test.python.docker import Podman, common, constant - - -class TestSystem(unittest.TestCase): - podman = None # initialized podman configuration for tests - service = None # podman service instance - topContainerId = "" - - def setUp(self): - super().setUp() - self.client = DockerClient(base_url="tcp://127.0.0.1:8080", timeout=15) - - TestSystem.podman.restore_image_from_cache(self.client) - TestSystem.topContainerId = common.run_top_container(self.client) - - def tearDown(self): - common.remove_all_containers(self.client) - common.remove_all_images(self.client) - self.client.close() - return super().tearDown() - - @classmethod - def setUpClass(cls): - super().setUpClass() - TestSystem.podman = Podman() - TestSystem.service = TestSystem.podman.open( - "system", "service", "tcp:127.0.0.1:8080", "--time=0" - ) - # give the service some time to be ready... - time.sleep(2) - - returncode = TestSystem.service.poll() - if returncode is not None: - raise subprocess.CalledProcessError(returncode, "podman system service") - - @classmethod - def tearDownClass(cls): - TestSystem.service.terminate() - stdout, stderr = TestSystem.service.communicate(timeout=0.5) - if stdout: - sys.stdout.write("\nImages Service Stdout:\n" + stdout.decode("utf-8")) - if stderr: - sys.stderr.write("\nImAges Service Stderr:\n" + stderr.decode("utf-8")) - - TestSystem.podman.tear_down() - return super().tearDownClass() - - def test_Info(self): - self.assertIsNotNone(self.client.info()) - - def test_info_container_details(self): - info = self.client.info() - self.assertEqual(info["Containers"], 1) - self.client.containers.create(image=constant.ALPINE) - info = self.client.info() - self.assertEqual(info["Containers"], 2) - - def test_version(self): - version = self.client.version() - self.assertIsNotNone(version["Platform"]["Name"]) -- cgit v1.2.3-54-g00ecf