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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
|
import os
import shutil
import unittest
import docker
from .. import helpers
from docker.utils import kwargs_from_env
TEST_IMG = 'alpine:3.10'
TEST_API_VERSION = os.environ.get('DOCKER_TEST_API_VERSION')
class BaseIntegrationTest(unittest.TestCase):
"""
A base class for integration test cases. It cleans up the Docker server
after itself.
"""
def setUp(self):
self.tmp_imgs = []
self.tmp_containers = []
self.tmp_folders = []
self.tmp_volumes = []
self.tmp_networks = []
self.tmp_plugins = []
self.tmp_secrets = []
self.tmp_configs = []
def tearDown(self):
client = docker.from_env(version=TEST_API_VERSION)
try:
for img in self.tmp_imgs:
try:
client.api.remove_image(img)
except docker.errors.APIError:
pass
for container in self.tmp_containers:
try:
client.api.remove_container(container, force=True, v=True)
except docker.errors.APIError:
pass
for network in self.tmp_networks:
try:
client.api.remove_network(network)
except docker.errors.APIError:
pass
for volume in self.tmp_volumes:
try:
client.api.remove_volume(volume)
except docker.errors.APIError:
pass
for secret in self.tmp_secrets:
try:
client.api.remove_secret(secret)
except docker.errors.APIError:
pass
for config in self.tmp_configs:
try:
client.api.remove_config(config)
except docker.errors.APIError:
pass
for folder in self.tmp_folders:
shutil.rmtree(folder)
finally:
client.close()
class BaseAPIIntegrationTest(BaseIntegrationTest):
"""
A test case for `APIClient` integration tests. It sets up an `APIClient`
as `self.client`.
"""
def setUp(self):
super().setUp()
self.client = self.get_client_instance()
def tearDown(self):
super().tearDown()
self.client.close()
@staticmethod
def get_client_instance():
return docker.APIClient(
version=TEST_API_VERSION, timeout=60, **kwargs_from_env()
)
@staticmethod
def _init_swarm(client, **kwargs):
return client.init_swarm(
'127.0.0.1', listen_addr=helpers.swarm_listen_addr(), **kwargs
)
def run_container(self, *args, **kwargs):
container = self.client.create_container(*args, **kwargs)
self.tmp_containers.append(container)
self.client.start(container)
exitcode = self.client.wait(container)['StatusCode']
if exitcode != 0:
output = self.client.logs(container)
raise Exception(
"Container exited with code {}:\n{}"
.format(exitcode, output))
return container
def create_and_start(self, image=TEST_IMG, command='top', **kwargs):
container = self.client.create_container(
image=image, command=command, **kwargs)
self.tmp_containers.append(container)
self.client.start(container)
return container
def execute(self, container, cmd, exit_code=0, **kwargs):
exc = self.client.exec_create(container, cmd, **kwargs)
output = self.client.exec_start(exc)
actual_exit_code = self.client.exec_inspect(exc)['ExitCode']
msg = "Expected `{}` to exit with code {} but returned {}:\n{}".format(
" ".join(cmd), exit_code, actual_exit_code, output)
assert actual_exit_code == exit_code, msg
def init_swarm(self, **kwargs):
return self._init_swarm(self.client, **kwargs)
|
