summaryrefslogtreecommitdiff
path: root/test/python/docker
diff options
context:
space:
mode:
Diffstat (limited to 'test/python/docker')
-rw-r--r--test/python/docker/__init__.py17
-rw-r--r--test/python/docker/common.py24
-rw-r--r--test/python/docker/test_containers.py157
-rw-r--r--test/python/docker/test_images.py111
-rw-r--r--test/python/docker/test_system.py9
5 files changed, 135 insertions, 183 deletions
diff --git a/test/python/docker/__init__.py b/test/python/docker/__init__.py
index 0e10676b9..316b102f4 100644
--- a/test/python/docker/__init__.py
+++ b/test/python/docker/__init__.py
@@ -6,6 +6,8 @@ import shutil
import subprocess
import tempfile
+from docker import DockerClient
+
from test.python.docker import constant
@@ -141,16 +143,15 @@ class Podman(object):
def tear_down(self):
shutil.rmtree(self.anchor_directory, ignore_errors=True)
- def restore_image_from_cache(self, client):
- img = os.path.join(self.image_cache, constant.ALPINE_TARBALL)
- if not os.path.exists(img):
- client.pull(constant.ALPINE)
- image = client.get_image(constant.ALPINE)
- with open(img, mode="wb") as tarball:
- for frame in image:
+ def restore_image_from_cache(self, client: DockerClient):
+ path = os.path.join(self.image_cache, constant.ALPINE_TARBALL)
+ if not os.path.exists(path):
+ img = client.images.pull(constant.ALPINE)
+ with open(path, mode="wb") as tarball:
+ for frame in img.save(named=True):
tarball.write(frame)
else:
- self.run("load", "-i", img, check=True)
+ self.run("load", "-i", path, check=True)
def flush_image_cache(self):
for f in pathlib.Path(self.image_cache).glob("*.tar"):
diff --git a/test/python/docker/common.py b/test/python/docker/common.py
index 2828d2d20..e79d64a9b 100644
--- a/test/python/docker/common.py
+++ b/test/python/docker/common.py
@@ -1,21 +1,23 @@
-from docker import APIClient
+from docker import DockerClient
from test.python.docker import constant
-def run_top_container(client: APIClient):
- c = client.create_container(
+def run_top_container(client: DockerClient):
+ c = client.containers.create(
constant.ALPINE, command="top", detach=True, tty=True, name="top"
)
- client.start(c.get("Id"))
- return c.get("Id")
+ c.start()
+ return c.id
-def remove_all_containers(client: APIClient):
- for ctnr in client.containers(quiet=True):
- client.remove_container(ctnr, force=True)
+def remove_all_containers(client: DockerClient):
+ for ctnr in client.containers.list(all=True):
+ ctnr.remove(force=True)
-def remove_all_images(client: APIClient):
- for image in client.images(quiet=True):
- client.remove_image(image, 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/test_containers.py b/test/python/docker/test_containers.py
index 1c4c9ab53..5fb340fd4 100644
--- a/test/python/docker/test_containers.py
+++ b/test/python/docker/test_containers.py
@@ -3,7 +3,7 @@ import sys
import time
import unittest
-from docker import APIClient, errors
+from docker import DockerClient, errors
from test.python.docker import Podman, common, constant
@@ -15,7 +15,7 @@ class TestContainers(unittest.TestCase):
def setUp(self):
super().setUp()
- self.client = APIClient(base_url="tcp://127.0.0.1:8080", timeout=15)
+ 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)
@@ -52,146 +52,115 @@ class TestContainers(unittest.TestCase):
TestContainers.podman.tear_down()
return super().tearDownClass()
- def test_inspect_container(self):
- # Inspect bogus container
- with self.assertRaises(errors.NotFound) as error:
- self.client.inspect_container("dummy")
- self.assertEqual(error.exception.response.status_code, 404)
+ 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)
- # Inspect valid container by Id
- container = self.client.inspect_container(TestContainers.topContainerId)
- self.assertIn("top", container["Name"])
+ def test_create_network(self):
+ net = self.client.networks.create("testNetwork", driver="bridge")
+ ctnr = self.client.containers.create(image="alpine", detach=True)
+ net.connect(ctnr)
- # Inspect valid container by name
- container = self.client.inspect_container("top")
- self.assertIn(TestContainers.topContainerId, container["Id"])
+ nets = self.client.networks.list(greedy=True)
+ self.assertGreaterEqual(len(nets), 1)
- def test_create_container(self):
- # Run a container with detach mode
- container = self.client.create_container(image="alpine", detach=True)
- self.assertEqual(len(container), 2)
+ # 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):
- # Start bogus container
- with self.assertRaises(errors.NotFound) as error:
- self.client.start("dummy")
- self.assertEqual(error.exception.response.status_code, 404)
-
# Podman docs says it should give a 304 but returns with no response
# # Start a already started container should return 304
- # response = self.client.start(container=TestContainers.topContainerId)
+ # 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.create_container(image=constant.ALPINE, name="container2")
- containers = self.client.containers(quiet=True, all=True)
+ self.client.containers.create(image=constant.ALPINE, name="container2")
+ containers = self.client.containers.list(all=True)
self.assertEqual(len(containers), 2)
def test_stop_container(self):
- # Stop bogus container
- with self.assertRaises(errors.NotFound) as error:
- self.client.stop("dummy")
- self.assertEqual(error.exception.response.status_code, 404)
-
- # Validate the container state
- container = self.client.inspect_container("top")
- self.assertEqual(container["State"]["Status"], "running")
+ top = self.client.containers.get("top")
+ self.assertEqual(top.status, "running")
# Stop a running container and validate the state
- self.client.stop(TestContainers.topContainerId)
- container = self.client.inspect_container("top")
- self.assertIn(
- container["State"]["Status"],
- "stopped exited",
- )
+ top.stop()
+ top.reload()
+ self.assertIn(top.status, ("stopped", "exited"))
def test_restart_container(self):
- # Restart bogus container
- with self.assertRaises(errors.NotFound) as error:
- self.client.restart("dummy")
- self.assertEqual(error.exception.response.status_code, 404)
-
# Validate the container state
- self.client.stop(TestContainers.topContainerId)
- container = self.client.inspect_container("top")
- self.assertEqual(container["State"]["Status"], "stopped")
+ 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
- self.client.restart(TestContainers.topContainerId)
- container = self.client.inspect_container("top")
- self.assertEqual(container["State"]["Status"], "running")
+ top.restart()
+ top.reload()
+ self.assertEqual(top.status, "running")
def test_remove_container(self):
- # Remove bogus container
- with self.assertRaises(errors.NotFound) as error:
- self.client.remove_container("dummy")
- self.assertEqual(error.exception.response.status_code, 404)
-
# Remove container by ID with force
- self.client.remove_container(TestContainers.topContainerId, force=True)
- containers = self.client.containers()
- self.assertEqual(len(containers), 0)
+ 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
- containers = self.client.containers()
- self.assertTrue(len(containers), 1)
+ 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:
- self.client.remove_container(TestContainers.topContainerId)
+ top.remove()
self.assertEqual(error.exception.response.status_code, 500)
- # Remove container by ID with force
- self.client.stop(TestContainers.topContainerId)
- self.client.remove_container(TestContainers.topContainerId)
- containers = self.client.containers()
- self.assertEqual(len(containers), 0)
+ # Remove container by ID without force
+ top.stop()
+ top.remove()
+ self.assertEqual(len(self.client.containers.list()), 0)
def test_pause_container(self):
- # Pause bogus container
- with self.assertRaises(errors.NotFound) as error:
- self.client.pause("dummy")
- self.assertEqual(error.exception.response.status_code, 404)
-
# Validate the container state
- container = self.client.inspect_container("top")
- self.assertEqual(container["State"]["Status"], "running")
+ top = self.client.containers.get(TestContainers.topContainerId)
+ self.assertEqual(top.status, "running")
# Pause a running container and validate the state
- self.client.pause(container["Id"])
- container = self.client.inspect_container("top")
- self.assertEqual(container["State"]["Status"], "paused")
+ top.pause()
+ top.reload()
+ self.assertEqual(top.status, "paused")
def test_pause_stopped_container(self):
# Stop the container
- self.client.stop(TestContainers.topContainerId)
+ top = self.client.containers.get(TestContainers.topContainerId)
+ top.stop()
# Pause exited container should trow error
with self.assertRaises(errors.APIError) as error:
- self.client.pause(TestContainers.topContainerId)
+ top.pause()
self.assertEqual(error.exception.response.status_code, 500)
def test_unpause_container(self):
- # Unpause bogus container
- with self.assertRaises(errors.NotFound) as error:
- self.client.unpause("dummy")
- self.assertEqual(error.exception.response.status_code, 404)
+ top = self.client.containers.get(TestContainers.topContainerId)
# Validate the container state
- self.client.pause(TestContainers.topContainerId)
- container = self.client.inspect_container("top")
- self.assertEqual(container["State"]["Status"], "paused")
+ top.pause()
+ top.reload()
+ self.assertEqual(top.status, "paused")
# Pause a running container and validate the state
- self.client.unpause(TestContainers.topContainerId)
- container = self.client.inspect_container("top")
- self.assertEqual(container["State"]["Status"], "running")
+ top.unpause()
+ top.reload()
+ self.assertEqual(top.status, "running")
def test_list_container(self):
# Add container and validate the count
- self.client.create_container(image="alpine", detach=True)
- containers = self.client.containers(all=True)
+ self.client.containers.create(image="alpine", detach=True)
+ containers = self.client.containers.list(all=True)
self.assertEqual(len(containers), 2)
def test_filters(self):
@@ -199,16 +168,18 @@ class TestContainers(unittest.TestCase):
# List container with filter by id
filters = {"id": TestContainers.topContainerId}
- ctnrs = self.client.containers(all=True, filters=filters)
+ 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(all=True, filters=filters)
+ ctnrs = self.client.containers.list(all=True, filters=filters)
self.assertEqual(len(ctnrs), 1)
def test_rename_container(self):
+ top = self.client.containers.get(TestContainers.topContainerId)
+
# rename bogus container
with self.assertRaises(errors.APIError) as error:
- self.client.rename(container="dummy", name="newname")
+ top.rename(name="newname")
self.assertEqual(error.exception.response.status_code, 404)
diff --git a/test/python/docker/test_images.py b/test/python/docker/test_images.py
index f049da96f..7ef3d708b 100644
--- a/test/python/docker/test_images.py
+++ b/test/python/docker/test_images.py
@@ -5,7 +5,7 @@ import sys
import time
import unittest
-from docker import APIClient, errors
+from docker import DockerClient, errors
from test.python.docker import Podman, common, constant
@@ -16,7 +16,7 @@ class TestImages(unittest.TestCase):
def setUp(self):
super().setUp()
- self.client = APIClient(base_url="tcp://127.0.0.1:8080", timeout=15)
+ self.client = DockerClient(base_url="tcp://127.0.0.1:8080", timeout=15)
TestImages.podman.restore_image_from_cache(self.client)
@@ -51,83 +51,57 @@ class TestImages(unittest.TestCase):
TestImages.podman.tear_down()
return super().tearDownClass()
- def test_inspect_image(self):
- """Inspect Image"""
- # Check for error with wrong image name
- with self.assertRaises(errors.NotFound):
- self.client.inspect_image("dummy")
- alpine_image = self.client.inspect_image(constant.ALPINE)
- self.assertIn(constant.ALPINE, alpine_image["RepoTags"])
-
- def test_tag_invalid_image(self):
- """Tag Image
-
- Validates if invalid image name is given a bad response is encountered
- """
- with self.assertRaises(errors.NotFound):
- self.client.tag("dummy", "demo")
-
def test_tag_valid_image(self):
"""Validates if the image is tagged successfully"""
- self.client.tag(constant.ALPINE, "demo", constant.ALPINE_SHORTNAME)
- alpine_image = self.client.inspect_image(constant.ALPINE)
- for x in alpine_image["RepoTags"]:
- self.assertIn("alpine", x)
+ 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"""
- self.client.tag(constant.ALPINE_SHORTNAME, "demo", "rename")
- alpine_image = self.client.inspect_image(constant.ALPINE)
- self.assertNotIn("demo:test", alpine_image["RepoTags"])
+ 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"""
- all_images = self.client.images()
- self.assertEqual(len(all_images), 1)
+ self.assertEqual(len(self.client.images.list()), 1)
+
# Add more images
- self.client.pull(constant.BB)
- all_images = self.client.images()
- self.assertEqual(len(all_images), 2)
+ self.client.images.pull(constant.BB)
+ self.assertEqual(len(self.client.images.list()), 2)
# List images with filter
- filters = {"reference": "alpine"}
- all_images = self.client.images(filters=filters)
- self.assertEqual(len(all_images), 1)
+ self.assertEqual(
+ len(self.client.images.list(filters={"reference": "alpine"})), 1
+ )
def test_search_image(self):
"""Search for image"""
- response = self.client.search("libpod/alpine")
- for i in response:
- self.assertIn("quay.io/libpod/alpine", i["Name"])
+ for r in self.client.images.search("libpod/alpine"):
+ self.assertIn("quay.io/libpod/alpine", r["Name"])
def test_remove_image(self):
"""Remove image"""
# Check for error with wrong image name
with self.assertRaises(errors.NotFound):
- self.client.remove_image("dummy")
- all_images = self.client.images()
- self.assertEqual(len(all_images), 1)
+ self.client.images.remove("dummy")
+ self.assertEqual(len(self.client.images.list()), 1)
- alpine_image = self.client.inspect_image(constant.ALPINE)
- self.client.remove_image(alpine_image["Id"])
- all_images = self.client.images()
- self.assertEqual(len(all_images), 0)
+ self.client.images.remove(constant.ALPINE)
+ self.assertEqual(len(self.client.images.list()), 0)
def test_image_history(self):
"""Image history"""
- # Check for error with wrong image name
- with self.assertRaises(errors.NotFound):
- self.client.history("dummy")
-
- # NOTE: history() has incorrect return type hint
- history = self.client.history(constant.ALPINE)
- alpine_image = self.client.inspect_image(constant.ALPINE)
- image_id = (
- alpine_image["Id"][7:]
- if alpine_image["Id"].startswith("sha256:")
- else alpine_image["Id"]
- )
+ 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:
@@ -137,31 +111,34 @@ class TestImages(unittest.TestCase):
def test_get_image_exists_not(self):
"""Negative test for get image"""
with self.assertRaises(errors.NotFound):
- response = self.client.get_image("image_does_not_exists")
+ response = self.client.images.get("image_does_not_exists")
collections.deque(response)
- def test_export_image(self):
+ def test_save_image(self):
"""Export Image"""
- self.client.pull(constant.BB)
- image = self.client.get_image(constant.BB)
+ 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:
+ for frame in image.save(named=True):
tarball.write(frame)
sz = os.path.getsize(file)
self.assertGreater(sz, 0)
- def test_import_image(self):
+ def test_load_image(self):
"""Import|Load Image"""
- all_images = self.client.images()
- self.assertEqual(len(all_images), 1)
+ 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)
- file = os.path.join(TestImages.podman.image_cache, constant.ALPINE_TARBALL)
- self.client.import_image_from_file(filename=file)
+ with open(file, mode="rb") as saved:
+ _ = self.client.images.load(saved)
- all_images = self.client.images()
- self.assertEqual(len(all_images), 2)
+ self.assertEqual(len(self.client.images.list()), 2)
if __name__ == "__main__":
diff --git a/test/python/docker/test_system.py b/test/python/docker/test_system.py
index f911baee4..46b90e5f6 100644
--- a/test/python/docker/test_system.py
+++ b/test/python/docker/test_system.py
@@ -3,7 +3,7 @@ import sys
import time
import unittest
-from docker import APIClient
+from docker import DockerClient
from test.python.docker import Podman, common, constant
@@ -15,7 +15,7 @@ class TestSystem(unittest.TestCase):
def setUp(self):
super().setUp()
- self.client = APIClient(base_url="tcp://127.0.0.1:8080", timeout=15)
+ 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)
@@ -58,9 +58,10 @@ class TestSystem(unittest.TestCase):
def test_info_container_details(self):
info = self.client.info()
self.assertEqual(info["Containers"], 1)
- self.client.create_container(image=constant.ALPINE)
+ self.client.containers.create(image=constant.ALPINE)
info = self.client.info()
self.assertEqual(info["Containers"], 2)
def test_version(self):
- self.assertIsNotNone(self.client.version())
+ version = self.client.version()
+ self.assertIsNotNone(version["Platform"]["Name"])
generated by cgit v1.2.3-54-g00ecf (git 2.39.0) at 2024-11-19 01:11:06 +0000