1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
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 test.python.docker import PodmanAPI
from test.python.docker.compat import constant
def run_top_container(client: DockerClient):
"""Run top command in a alpine container."""
ctnr = client.containers.create(
constant.ALPINE,
command="top",
detach=True,
tty=True,
name="top",
)
ctnr.start()
return ctnr.id
def remove_all_containers(client: DockerClient):
"""Delete all containers from the Podman service."""
for ctnr in client.containers.list(all=True):
ctnr.remove(force=True)
def remove_all_images(client: DockerClient):
"""Delete all images from the Podman service."""
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)
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()
|