opensource/podman
1
0
Fork 0
mirror of https://github.com/containers/podman.git synced 2026-03-13 08:01:19 +08:00
Code Issues Packages Projects Releases Wiki Activity

Refactor docker-py compatibility tests

Browse Source 
* Add which python client is being used to run tests, see "python
  client" below.
* Remove redundate code from test classes
* Update/Add comments to modules and classes

======================================================= test session starts ========================================================
platform linux -- Python 3.10.0, pytest-6.2.4, py-1.10.0, pluggy-0.13.1
python client -- DockerClient
rootdir: /home/jhonce/Projects/go/src/github.com/containers/podman
plugins: requests-mock-1.8.0
collected 33 items

test/python/docker/compat/test_containers.py ...s..............  [ 54%]
test/python/docker/compat/test_images.py ............  [ 90%]
test/python/docker/compat/test_system.py ...  [100%]

Note: Follow-up PRs will verify the test results and expand the tests.

Signed-off-by: Jhon Honce <jhonce@redhat.com>
This commit is contained in:
Jhon Honce
2022-02-28 16:26:43 -07:00
parent c39dffe83d
commit 7729afe979
9 changed files with 275 additions and 299 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
Makefile
View File

@@ -540,7 +540,7 @@ validate.completions:
.PHONY: run-docker-py-tests .PHONY: run-docker-py-tests
run-docker-py-tests: run-docker-py-tests:
touch test/__init__.py touch test/__init__.py
env CONTAINERS_CONF=$(CURDIR)/test/apiv2/containers.conf pytest test/python/docker/ env CONTAINERS_CONF=$(CURDIR)/test/apiv2/containers.conf pytest --disable-warnings test/python/docker/
-rm test/__init__.py -rm test/__init__.py
.PHONY: localunit .PHONY: localunit

15
pyproject.toml Normal file
View File

@@ -0,0 +1,15 @@
[tool.black]
line-length = 100
target-version = ['py36']
include = '\.pyi?$'
exclude = '''
/(
\.git
| \.tox
| \.venv
| build
| dist
| docs
| hack
)/
'''

30
test/python/docker/__init__.py
View File

@@ -1,4 +1,6 @@
import configparser """
Helpers for integration tests using DockerClient
"""
import json import json
import os import os
import pathlib import pathlib
@@ -11,7 +13,7 @@ from docker import DockerClient
from .compat import constant from .compat import constant
class Podman(object): class PodmanAPI:
""" """
Instances hold the configuration and setup for running podman commands Instances hold the configuration and setup for running podman commands
""" """
@@ -53,17 +55,13 @@ location = "mirror.localhost:5000"
""" """
with open(os.environ["CONTAINERS_REGISTRIES_CONF"], "w") as w: with open(os.environ["CONTAINERS_REGISTRIES_CONF"], "w") as file:
w.write(conf) file.write(conf)
os.environ["CNI_CONFIG_PATH"] = os.path.join( os.environ["CNI_CONFIG_PATH"] = os.path.join(self.anchor_directory, "cni", "net.d")
self.anchor_directory, "cni", "net.d"
)
os.makedirs(os.environ["CNI_CONFIG_PATH"], exist_ok=True) os.makedirs(os.environ["CNI_CONFIG_PATH"], exist_ok=True)
self.cmd.append("--network-config-dir=" + os.environ["CNI_CONFIG_PATH"]) self.cmd.append("--network-config-dir=" + os.environ["CNI_CONFIG_PATH"])
cni_cfg = os.path.join( cni_cfg = os.path.join(os.environ["CNI_CONFIG_PATH"], "87-podman-bridge.conflist")
os.environ["CNI_CONFIG_PATH"], "87-podman-bridge.conflist"
)
# json decoded and encoded to ensure legal json # json decoded and encoded to ensure legal json
buf = json.loads( buf = json.loads(
""" """
@@ -93,8 +91,8 @@ location = "mirror.localhost:5000"
} }
""" """
) )
with open(cni_cfg, "w") as w: with open(cni_cfg, "w") as file:
json.dump(buf, w) json.dump(buf, file)
def open(self, command, *args, **kwargs): def open(self, command, *args, **kwargs):
"""Podman initialized instance to run a given command """Podman initialized instance to run a given command
@@ -111,6 +109,7 @@ location = "mirror.localhost:5000"
shell = kwargs.get("shell", False) shell = kwargs.get("shell", False)
# pylint: disable=consider-using-with
return subprocess.Popen( return subprocess.Popen(
cmd, cmd,
shell=shell, shell=shell,
@@ -144,9 +143,11 @@ location = "mirror.localhost:5000"
) )
def tear_down(self): def tear_down(self):
"""Delete test environment."""
shutil.rmtree(self.anchor_directory, ignore_errors=True) shutil.rmtree(self.anchor_directory, ignore_errors=True)
def restore_image_from_cache(self, client: DockerClient): def restore_image_from_cache(self, client: DockerClient):
"""Populate images from cache."""
path = os.path.join(self.image_cache, constant.ALPINE_TARBALL) path = os.path.join(self.image_cache, constant.ALPINE_TARBALL)
if not os.path.exists(path): if not os.path.exists(path):
img = client.images.pull(constant.ALPINE) img = client.images.pull(constant.ALPINE)
@@ -157,5 +158,6 @@ location = "mirror.localhost:5000"
self.run("load", "-i", path, check=True) self.run("load", "-i", path, check=True)
def flush_image_cache(self): def flush_image_cache(self):
for f in pathlib.Path(self.image_cache).glob("*.tar"): """Delete image cache."""
f.unlink(f) for file in pathlib.Path(self.image_cache).glob("*.tar"):
file.unlink(missing_ok=True)

77
test/python/docker/compat/common.py
View File

@@ -1,23 +1,92 @@
"""
Fixtures and Helpers for unittests.
"""
import subprocess
import sys
import time
import unittest
# pylint: disable=no-name-in-module,import-error,wrong-import-order
from docker import DockerClient from docker import DockerClient
from test.python.docker import PodmanAPI
from test.python.docker.compat import constant from test.python.docker.compat import constant
def run_top_container(client: DockerClient): def run_top_container(client: DockerClient):
c = client.containers.create( """Run top command in a alpine container."""
constant.ALPINE, command="top", detach=True, tty=True, name="top" ctnr = client.containers.create(
constant.ALPINE,
command="top",
detach=True,
tty=True,
name="top",
) )
c.start() ctnr.start()
return c.id return ctnr.id
def remove_all_containers(client: DockerClient): def remove_all_containers(client: DockerClient):
"""Delete all containers from the Podman service."""
for ctnr in client.containers.list(all=True): for ctnr in client.containers.list(all=True):
ctnr.remove(force=True) ctnr.remove(force=True)
def remove_all_images(client: DockerClient): def remove_all_images(client: DockerClient):
"""Delete all images from the Podman service."""
for img in client.images.list(): for img in client.images.list():
# FIXME should DELETE /images accept the sha256: prefix? # FIXME should DELETE /images accept the sha256: prefix?
id_ = img.id.removeprefix("sha256:") id_ = img.id.removeprefix("sha256:")
client.images.remove(id_, force=True) client.images.remove(id_, force=True)
class DockerTestCase(unittest.TestCase):
"""Specialized TestCase class for testing against Podman service."""
podman: PodmanAPI = None # initialized podman configuration for tests
service: subprocess.Popen = None # podman service instance
top_container_id: str = None
docker: DockerClient = None
@classmethod
def setUpClass(cls) -> None:
super().setUpClass()
cls.podman = PodmanAPI()
super().addClassCleanup(cls.podman.tear_down)
cls.service = cls.podman.open("system", "service", "tcp:127.0.0.1:8080", "--time=0")
# give the service some time to be ready...
time.sleep(2)
return_code = cls.service.poll()
if return_code is not None:
raise subprocess.CalledProcessError(return_code, "podman system service")
@classmethod
def tearDownClass(cls) -> None:
cls.service.terminate()
stdout, stderr = cls.service.communicate(timeout=0.5)
if stdout:
sys.stdout.write("\ndocker-py -- Service Stdout:\n" + stdout.decode("utf-8"))
if stderr:
sys.stderr.write("\ndocker-py -- Service Stderr:\n" + stderr.decode("utf-8"))
return super().tearDownClass()
def setUp(self) -> None:
super().setUp()
self.docker = DockerClient(base_url="tcp://127.0.0.1:8080", timeout=15)
self.addCleanup(self.docker.close)
self.podman.restore_image_from_cache(self.docker)
self.top_container_id = run_top_container(self.docker)
self.assertIsNotNone(self.top_container_id, "Failed to create 'top' container")
def tearDown(self) -> None:
remove_all_containers(self.docker)
remove_all_images(self.docker)
super().tearDown()

4
test/python/docker/compat/constant.py
View File

@@ -1,6 +1,8 @@
"""
Constants to use in writing unittests.
"""
ALPINE = "quay.io/libpod/alpine:latest" ALPINE = "quay.io/libpod/alpine:latest"
ALPINE_SHORTNAME = "alpine" ALPINE_SHORTNAME = "alpine"
ALPINE_TARBALL = "alpine.tar" ALPINE_TARBALL = "alpine.tar"
BB = "quay.io/libpod/busybox:latest" BB = "quay.io/libpod/busybox:latest"
NGINX = "quay.io/libpod/alpine_nginx:latest" NGINX = "quay.io/libpod/alpine_nginx:latest"
infra = "k8s.gcr.io/pause:3.2"

232
test/python/docker/compat/test_containers.py
View File

@@ -1,111 +1,56 @@
"""
Integration tests for exercising docker-py against Podman Service.
"""
import io import io
import subprocess import tarfile
import sys from typing import IO, List, Optional
import time
import unittest
from typing import IO, Optional, List
from docker import DockerClient, errors from docker import errors
from docker.models.containers import Container from docker.models.containers import Container
from docker.models.images import Image from docker.models.images import Image
from docker.models.volumes import Volume from docker.models.volumes import Volume
from docker.types import Mount from docker.types import Mount
from test.python.docker import Podman # pylint: disable=no-name-in-module,import-error,wrong-import-order
from test.python.docker.compat import common, constant from test.python.docker.compat import common, constant
import tarfile
# pylint: disable=missing-function-docstring
class TestContainers(unittest.TestCase): class TestContainers(common.DockerTestCase):
podman = None # initialized podman configuration for tests """TestCase for exercising containers."""
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): def test_create_container(self):
# Run a container with detach mode """Run a container with detach mode."""
self.client.containers.create(image="alpine", detach=True) self.docker.containers.create(image="alpine", detach=True)
self.assertEqual(len(self.client.containers.list(all=True)), 2) self.assertEqual(len(self.docker.containers.list(all=True)), 2)
def test_create_network(self): def test_create_network(self):
net = self.client.networks.create("testNetwork", driver="bridge") """Add network to a container."""
ctnr = self.client.containers.create(image="alpine", detach=True) self.docker.networks.create("testNetwork", driver="bridge")
self.docker.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): def test_start_container(self):
# Podman docs says it should give a 304 but returns with no response # Podman docs says it should give a 304 but returns with no response
# # Start a already started container should return 304 # # Start a already started container should return 304
# response = self.client.api.start(container=TestContainers.topContainerId) # response = self.docker.api.start(container=self.top_container_id)
# self.assertEqual(error.exception.response.status_code, 304) # self.assertEqual(error.exception.response.status_code, 304)
# Create a new container and validate the count # Create a new container and validate the count
self.client.containers.create(image=constant.ALPINE, name="container2") self.docker.containers.create(image=constant.ALPINE, name="container2")
containers = self.client.containers.list(all=True) containers = self.docker.containers.list(all=True)
self.assertEqual(len(containers), 2) self.assertEqual(len(containers), 2)
def test_start_container_with_random_port_bind(self): def test_start_container_with_random_port_bind(self):
container = self.client.containers.create( container = self.docker.containers.create(
image=constant.ALPINE, image=constant.ALPINE,
name="containerWithRandomBind", name="containerWithRandomBind",
ports={"1234/tcp": None}, ports={"1234/tcp": None},
) )
containers = self.client.containers.list(all=True) containers = self.docker.containers.list(all=True)
self.assertTrue(container in containers) self.assertTrue(container in containers)
def test_stop_container(self): def test_stop_container(self):
top = self.client.containers.get(TestContainers.topContainerId) top = self.docker.containers.get(self.top_container_id)
self.assertEqual(top.status, "running") self.assertEqual(top.status, "running")
# Stop a running container and validate the state # Stop a running container and validate the state
@@ -114,7 +59,7 @@ class TestContainers(unittest.TestCase):
self.assertIn(top.status, ("stopped", "exited")) self.assertIn(top.status, ("stopped", "exited"))
def test_kill_container(self): def test_kill_container(self):
top = self.client.containers.get(TestContainers.topContainerId) top = self.docker.containers.get(self.top_container_id)
self.assertEqual(top.status, "running") self.assertEqual(top.status, "running")
# Kill a running container and validate the state # Kill a running container and validate the state
@@ -124,7 +69,7 @@ class TestContainers(unittest.TestCase):
def test_restart_container(self): def test_restart_container(self):
# Validate the container state # Validate the container state
top = self.client.containers.get(TestContainers.topContainerId) top = self.docker.containers.get(self.top_container_id)
top.stop() top.stop()
top.reload() top.reload()
self.assertIn(top.status, ("stopped", "exited")) self.assertIn(top.status, ("stopped", "exited"))
@@ -136,16 +81,16 @@ class TestContainers(unittest.TestCase):
def test_remove_container(self): def test_remove_container(self):
# Remove container by ID with force # Remove container by ID with force
top = self.client.containers.get(TestContainers.topContainerId) top = self.docker.containers.get(self.top_container_id)
top.remove(force=True) top.remove(force=True)
self.assertEqual(len(self.client.containers.list()), 0) self.assertEqual(len(self.docker.containers.list()), 0)
def test_remove_container_without_force(self): def test_remove_container_without_force(self):
# Validate current container count # Validate current container count
self.assertEqual(len(self.client.containers.list()), 1) self.assertEqual(len(self.docker.containers.list()), 1)
# Remove running container should throw error # Remove running container should throw error
top = self.client.containers.get(TestContainers.topContainerId) top = self.docker.containers.get(self.top_container_id)
with self.assertRaises(errors.APIError) as error: with self.assertRaises(errors.APIError) as error:
top.remove() top.remove()
self.assertEqual(error.exception.response.status_code, 500) self.assertEqual(error.exception.response.status_code, 500)
@@ -153,11 +98,11 @@ class TestContainers(unittest.TestCase):
# Remove container by ID without force # Remove container by ID without force
top.stop() top.stop()
top.remove() top.remove()
self.assertEqual(len(self.client.containers.list()), 0) self.assertEqual(len(self.docker.containers.list()), 0)
def test_pause_container(self): def test_pause_container(self):
# Validate the container state # Validate the container state
top = self.client.containers.get(TestContainers.topContainerId) top = self.docker.containers.get(self.top_container_id)
self.assertEqual(top.status, "running") self.assertEqual(top.status, "running")
# Pause a running container and validate the state # Pause a running container and validate the state
@@ -167,7 +112,7 @@ class TestContainers(unittest.TestCase):
def test_pause_stopped_container(self): def test_pause_stopped_container(self):
# Stop the container # Stop the container
top = self.client.containers.get(TestContainers.topContainerId) top = self.docker.containers.get(self.top_container_id)
top.stop() top.stop()
# Pause exited container should throw error # Pause exited container should throw error
@@ -176,7 +121,7 @@ class TestContainers(unittest.TestCase):
self.assertEqual(error.exception.response.status_code, 500) self.assertEqual(error.exception.response.status_code, 500)
def test_unpause_container(self): def test_unpause_container(self):
top = self.client.containers.get(TestContainers.topContainerId) top = self.docker.containers.get(self.top_container_id)
# Validate the container state # Validate the container state
top.pause() top.pause()
@@ -190,21 +135,21 @@ class TestContainers(unittest.TestCase):
def test_list_container(self): def test_list_container(self):
# Add container and validate the count # Add container and validate the count
self.client.containers.create(image="alpine", detach=True) self.docker.containers.create(image="alpine", detach=True)
containers = self.client.containers.list(all=True) containers = self.docker.containers.list(all=True)
self.assertEqual(len(containers), 2) self.assertEqual(len(containers), 2)
def test_filters(self): def test_filters(self):
self.skipTest("TODO Endpoint does not yet support filters") self.skipTest("TODO Endpoint does not yet support filters")
# List container with filter by id # List container with filter by id
filters = {"id": TestContainers.topContainerId} filters = {"id": self.top_container_id}
ctnrs = self.client.containers.list(all=True, filters=filters) ctnrs = self.docker.containers.list(all=True, filters=filters)
self.assertEqual(len(ctnrs), 1) self.assertEqual(len(ctnrs), 1)
# List container with filter by name # List container with filter by name
filters = {"name": "top"} filters = {"name": "top"}
ctnrs = self.client.containers.list(all=True, filters=filters) ctnrs = self.docker.containers.list(all=True, filters=filters)
self.assertEqual(len(ctnrs), 1) self.assertEqual(len(ctnrs), 1)
def test_copy_to_container(self): def test_copy_to_container(self):
@@ -212,31 +157,33 @@ class TestContainers(unittest.TestCase):
vol: Optional[Volume] = None vol: Optional[Volume] = None
try: try:
test_file_content = b"Hello World!" test_file_content = b"Hello World!"
vol = self.client.volumes.create("test-volume") vol = self.docker.volumes.create("test-volume")
ctr = self.client.containers.create(image="alpine", ctr = self.docker.containers.create(
detach=True, image="alpine",
command="top", detach=True,
volumes=["test-volume:/test-volume-read-only:ro"]) command="top",
volumes=["test-volume:/test-volume-read-only:ro"],
)
ctr.start() ctr.start()
buff: IO[bytes] = io.BytesIO() buff: IO[bytes] = io.BytesIO()
with tarfile.open(fileobj=buff, mode="w:") as tf: with tarfile.open(fileobj=buff, mode="w:") as file:
ti: tarfile.TarInfo = tarfile.TarInfo() info: tarfile.TarInfo = tarfile.TarInfo()
ti.uid = 1042 info.uid = 1042
ti.gid = 1043 info.gid = 1043
ti.name = "a.txt" info.name = "a.txt"
ti.path = "a.txt" info.path = "a.txt"
ti.mode = 0o644 info.mode = 0o644
ti.type = tarfile.REGTYPE info.type = tarfile.REGTYPE
ti.size = len(test_file_content) info.size = len(test_file_content)
tf.addfile(ti, fileobj=io.BytesIO(test_file_content)) file.addfile(info, fileobj=io.BytesIO(test_file_content))
buff.seek(0) buff.seek(0)
ctr.put_archive("/tmp/", buff) ctr.put_archive("/tmp/", buff)
ret, out = ctr.exec_run(["stat", "-c", "%u:%g", "/tmp/a.txt"]) ret, out = ctr.exec_run(["stat", "-c", "%u:%g", "/tmp/a.txt"])
self.assertEqual(ret, 0) self.assertEqual(ret, 0)
self.assertEqual(out.rstrip(), b'1042:1043', "UID/GID of copied file") self.assertEqual(out.rstrip(), b"1042:1043", "UID/GID of copied file")
ret, out = ctr.exec_run(["cat", "/tmp/a.txt"]) ret, out = ctr.exec_run(["cat", "/tmp/a.txt"])
self.assertEqual(ret, 0) self.assertEqual(ret, 0)
@@ -253,48 +200,61 @@ class TestContainers(unittest.TestCase):
vol.remove(force=True) vol.remove(force=True)
def test_mount_preexisting_dir(self): def test_mount_preexisting_dir(self):
dockerfile = (B'FROM quay.io/libpod/alpine:latest\n' dockerfile = (
B'USER root\n' b"FROM quay.io/libpod/alpine:latest\n"
B'RUN mkdir -p /workspace\n' b"USER root\n"
B'RUN chown 1042:1043 /workspace') b"RUN mkdir -p /workspace\n"
b"RUN chown 1042:1043 /workspace"
)
img: Image img: Image
img, out = self.client.images.build(fileobj=io.BytesIO(dockerfile)) img, out = self.docker.images.build(fileobj=io.BytesIO(dockerfile))
ctr: Container = self.client.containers.create(image=img.id, detach=True, command="top", ctr: Container = self.docker.containers.create(
volumes=["test_mount_preexisting_dir_vol:/workspace"]) image=img.id,
detach=True,
command="top",
volumes=["test_mount_preexisting_dir_vol:/workspace"],
)
ctr.start() ctr.start()
ret, out = ctr.exec_run(["stat", "-c", "%u:%g", "/workspace"]) _, out = ctr.exec_run(["stat", "-c", "%u:%g", "/workspace"])
self.assertEqual(out.rstrip(), b'1042:1043', "UID/GID set in dockerfile") self.assertEqual(out.rstrip(), b"1042:1043", "UID/GID set in dockerfile")
def test_non_existant_workdir(self): def test_non_existant_workdir(self):
dockerfile = (B'FROM quay.io/libpod/alpine:latest\n' dockerfile = (
B'USER root\n' b"FROM quay.io/libpod/alpine:latest\n"
B'WORKDIR /workspace/scratch\n' b"USER root\n"
B'RUN touch test') b"WORKDIR /workspace/scratch\n"
b"RUN touch test"
)
img: Image img: Image
img, out = self.client.images.build(fileobj=io.BytesIO(dockerfile)) img, _ = self.docker.images.build(fileobj=io.BytesIO(dockerfile))
ctr: Container = self.client.containers.create(image=img.id, detach=True, command="top", ctr: Container = self.docker.containers.create(
volumes=["test_non_existant_workdir:/workspace"]) image=img.id,
detach=True,
command="top",
volumes=["test_non_existant_workdir:/workspace"],
)
ctr.start() ctr.start()
ret, out = ctr.exec_run(["stat", "/workspace/scratch/test"]) ret, _ = ctr.exec_run(["stat", "/workspace/scratch/test"])
self.assertEqual(ret, 0, "Working directory created if it doesn't exist") self.assertEqual(ret, 0, "Working directory created if it doesn't exist")
def test_mount_rw_by_default(self): def test_mount_rw_by_default(self):
ctr: Optional[Container] = None ctr: Optional[Container] = None
vol: Optional[Volume] = None vol: Optional[Volume] = None
try: try:
vol = self.client.volumes.create("test-volume") vol = self.docker.volumes.create("test-volume")
ctr = self.client.containers.create(image="alpine", ctr = self.docker.containers.create(
detach=True, image="alpine",
command="top", detach=True,
mounts=[Mount(target="/vol-mnt", command="top",
source="test-volume", mounts=[
type='volume', Mount(target="/vol-mnt", source="test-volume", type="volume", read_only=False)
read_only=False)]) ],
ctr_inspect = self.client.api.inspect_container(ctr.id) )
ctr_inspect = self.docker.api.inspect_container(ctr.id)
binds: List[str] = ctr_inspect["HostConfig"]["Binds"] binds: List[str] = ctr_inspect["HostConfig"]["Binds"]
self.assertEqual(len(binds), 1) self.assertEqual(len(binds), 1)
self.assertEqual(binds[0], 'test-volume:/vol-mnt:rw,rprivate,nosuid,nodev,rbind') self.assertEqual(binds[0], "test-volume:/vol-mnt:rw,rprivate,nosuid,nodev,rbind")
finally: finally:
if ctr is not None: if ctr is not None:
ctr.remove() ctr.remove()

131
test/python/docker/compat/test_images.py
View File

@@ -1,116 +1,73 @@
import collections """
Integration tests for exercising docker-py against Podman Service.
"""
import io import io
import os import os
import subprocess
import sys
import time
import unittest import unittest
from docker import DockerClient, errors from docker import errors
from docker.errors import APIError
from test.python.docker import Podman # pylint: disable=no-name-in-module,import-error,wrong-import-order
from test.python.docker.compat import common, constant from test.python.docker.compat import common, constant
class TestImages(unittest.TestCase): class TestImages(common.DockerTestCase):
podman = None # initialized podman configuration for tests """TestCase for exercising images."""
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): def test_tag_valid_image(self):
"""Validates if the image is tagged successfully""" """Validates if the image is tagged successfully"""
alpine = self.client.images.get(constant.ALPINE) alpine = self.docker.images.get(constant.ALPINE)
self.assertTrue(alpine.tag("demo", constant.ALPINE_SHORTNAME)) self.assertTrue(alpine.tag("demo", constant.ALPINE_SHORTNAME))
alpine = self.client.images.get(constant.ALPINE) alpine = self.docker.images.get(constant.ALPINE)
for t in alpine.tags: for tag in alpine.tags:
self.assertIn("alpine", t) self.assertIn("alpine", tag)
# @unittest.skip("doesn't work now")
def test_retag_valid_image(self): def test_retag_valid_image(self):
"""Validates if name updates when the image is retagged""" """Validates if name updates when the image is re-tagged."""
alpine = self.client.images.get(constant.ALPINE) alpine = self.docker.images.get(constant.ALPINE)
self.assertTrue(alpine.tag("demo", "rename")) self.assertTrue(alpine.tag("demo", "rename"))
alpine = self.client.images.get(constant.ALPINE) alpine = self.docker.images.get(constant.ALPINE)
self.assertNotIn("demo:test", alpine.tags) self.assertNotIn("demo:test", alpine.tags)
def test_list_images(self): def test_list_images(self):
"""List images""" """List images"""
self.assertEqual(len(self.client.images.list()), 1) self.assertEqual(len(self.docker.images.list()), 1)
# Add more images # Add more images
self.client.images.pull(constant.BB) self.docker.images.pull(constant.BB)
self.assertEqual(len(self.client.images.list()), 2) self.assertEqual(len(self.docker.images.list()), 2)
self.assertEqual(len(self.client.images.list(all=True)), 2) self.assertEqual(len(self.docker.images.list(all=True)), 2)
# List images with filter # List images with filter
self.assertEqual(len(self.client.images.list(filters={"reference": "alpine"})), 1) self.assertEqual(len(self.docker.images.list(filters={"reference": "alpine"})), 1)
def test_search_image(self): def test_search_image(self):
"""Search for image""" """Search for image"""
for r in self.client.images.search("alpine"): for registry in self.docker.images.search("alpine"):
# registry matches if string is in either one # registry matches if string is in either one
self.assertIn("alpine", r["Name"]+" "+r["Description"].lower()) self.assertIn("alpine", registry["Name"] + " " + registry["Description"].lower())
def test_search_bogus_image(self): def test_search_bogus_image(self):
"""Search for bogus image should throw exception""" """Search for bogus image should throw exception"""
try: with self.assertRaises(errors.APIError):
r = self.client.images.search("bogus/bogus") self.docker.images.search("bogus/bogus")
except:
return
self.assertTrue(len(r) == 0)
def test_remove_image(self): def test_remove_image(self):
"""Remove image""" """Remove image"""
# Check for error with wrong image name # Check for error with wrong image name
with self.assertRaises(errors.NotFound): with self.assertRaises(errors.NotFound):
self.client.images.remove("dummy") self.docker.images.remove("dummy")
self.assertEqual(len(self.client.images.list()), 1)
self.client.images.remove(constant.ALPINE) common.remove_all_containers(self.docker)
self.assertEqual(len(self.client.images.list()), 0) self.assertEqual(len(self.docker.images.list()), 1)
self.docker.images.remove(constant.ALPINE)
self.assertEqual(len(self.docker.images.list()), 0)
def test_image_history(self): def test_image_history(self):
"""Image history""" """Image history"""
img = self.client.images.get(constant.ALPINE) img = self.docker.images.get(constant.ALPINE)
history = img.history() history = img.history()
image_id = img.id[7:] if img.id.startswith("sha256:") else img.id image_id = img.id[7:] if img.id.startswith("sha256:") else img.id
@@ -122,47 +79,49 @@ class TestImages(unittest.TestCase):
def test_get_image_exists_not(self): def test_get_image_exists_not(self):
"""Negative test for get image""" """Negative test for get image"""
with self.assertRaises(errors.NotFound): with self.assertRaises(errors.NotFound):
response = self.client.images.get("image_does_not_exists") self.docker.images.get("image_does_not_exists")
collections.deque(response)
def test_save_image(self): def test_save_image(self):
"""Export Image""" """Export Image"""
image = self.client.images.pull(constant.BB) image = self.docker.images.pull(constant.BB)
file = os.path.join(TestImages.podman.image_cache, "busybox.tar") file = os.path.join(TestImages.podman.image_cache, "busybox.tar")
with open(file, mode="wb") as tarball: with open(file, mode="wb") as tarball:
for frame in image.save(named=True): for frame in image.save(named=True):
tarball.write(frame) tarball.write(frame)
sz = os.path.getsize(file) self.assertGreater(os.path.getsize(file), 0)
self.assertGreater(sz, 0)
def test_load_image(self): def test_load_image(self):
"""Import|Load Image""" """Import|Load Image"""
self.assertEqual(len(self.client.images.list()), 1) self.assertEqual(len(self.docker.images.list()), 1)
image = self.client.images.pull(constant.BB) image = self.docker.images.pull(constant.BB)
file = os.path.join(TestImages.podman.image_cache, "busybox.tar") file = os.path.join(TestImages.podman.image_cache, "busybox.tar")
with open(file, mode="wb") as tarball: with open(file, mode="wb") as tarball:
for frame in image.save(): for frame in image.save():
tarball.write(frame) tarball.write(frame)
with open(file, mode="rb") as saved: with open(file, mode="rb") as saved:
_ = self.client.images.load(saved) self.docker.images.load(saved)
self.assertEqual(len(self.client.images.list()), 2) self.assertEqual(len(self.docker.images.list()), 2)
def test_load_corrupt_image(self): def test_load_corrupt_image(self):
"""Import|Load Image failure""" """Import|Load Image failure"""
tarball = io.BytesIO("This is a corrupt tarball".encode("utf-8")) tarball = io.BytesIO("This is a corrupt tarball".encode("utf-8"))
with self.assertRaises(APIError): with self.assertRaises(errors.APIError):
self.client.images.load(tarball) self.docker.images.load(tarball)
def test_build_image(self): def test_build_image(self):
"""Build Image with custom labels."""
labels = {"apple": "red", "grape": "green"} labels = {"apple": "red", "grape": "green"}
_ = self.client.images.build( self.docker.images.build(
path="test/python/docker/build_labels", labels=labels, tag="labels", isolation="default" path="test/python/docker/build_labels",
labels=labels,
tag="labels",
isolation="default",
) )
image = self.client.images.get("labels") image = self.docker.images.get("labels")
self.assertEqual(image.labels["apple"], labels["apple"]) self.assertEqual(image.labels["apple"], labels["apple"])
self.assertEqual(image.labels["grape"], labels["grape"]) self.assertEqual(image.labels["grape"], labels["grape"])

75
test/python/docker/compat/test_system.py
View File

@@ -1,71 +1,32 @@
import subprocess """
import sys Integration tests for exercising docker-py against Podman Service.
import time """
import unittest
from docker import DockerClient # pylint: disable=no-name-in-module,import-error,wrong-import-order
from test.python.docker import constant
from test.python.docker import Podman, constant
from test.python.docker.compat import common from test.python.docker.compat import common
class TestSystem(unittest.TestCase): # pylint: disable=missing-function-docstring
podman = None # initialized podman configuration for tests class TestSystem(common.DockerTestCase):
service = None # podman service instance """TestCase for exercising Podman system services."""
topContainerId = ""
def setUp(self): def test_info(self):
super().setUp() info = self.docker.info()
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):
info = self.client.info()
self.assertIsNotNone(info) self.assertIsNotNone(info)
self.assertEqual(info["RegistryConfig"]["IndexConfigs"]["localhost:5000"]["Secure"], False) self.assertEqual(info["RegistryConfig"]["IndexConfigs"]["localhost:5000"]["Secure"], False)
self.assertEqual(info["RegistryConfig"]["IndexConfigs"]["localhost:5000"]["Mirrors"], ["mirror.localhost:5000"]) self.assertEqual(
info["RegistryConfig"]["IndexConfigs"]["localhost:5000"]["Mirrors"],
["mirror.localhost:5000"],
)
def test_info_container_details(self): def test_info_container_details(self):
info = self.client.info() info = self.docker.info()
self.assertEqual(info["Containers"], 1) self.assertEqual(info["Containers"], 1)
self.client.containers.create(image=constant.ALPINE) self.docker.containers.create(image=constant.ALPINE)
info = self.client.info() info = self.docker.info()
self.assertEqual(info["Containers"], 2) self.assertEqual(info["Containers"], 2)
def test_version(self): def test_version(self):
version = self.client.version() version = self.docker.version()
self.assertIsNotNone(version["Platform"]["Name"]) self.assertIsNotNone(version["Platform"]["Name"])

8
test/python/docker/conftest.py Normal file
View File

@@ -0,0 +1,8 @@
"""
Configure pytest
"""
def pytest_report_header(config):
"""Add header to report."""
return "python client -- DockerClient"