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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
|
import functools
import os
import os.path
import random
import re
import socket
import tarfile
import tempfile
import time
import docker
import paramiko
import pytest
import six
def make_tree(dirs, files):
base = tempfile.mkdtemp()
for path in dirs:
os.makedirs(os.path.join(base, path))
for path in files:
with open(os.path.join(base, path), 'w') as f:
f.write("content")
return base
def simple_tar(path):
f = tempfile.NamedTemporaryFile()
t = tarfile.open(mode='w', fileobj=f)
abs_path = os.path.abspath(path)
t.add(abs_path, arcname=os.path.basename(path), recursive=False)
t.close()
f.seek(0)
return f
def untar_file(tardata, filename):
with tarfile.open(mode='r', fileobj=tardata) as t:
f = t.extractfile(filename)
result = f.read()
f.close()
return result
def requires_api_version(version):
test_version = os.environ.get(
'DOCKER_TEST_API_VERSION', docker.constants.DEFAULT_DOCKER_API_VERSION
)
return pytest.mark.skipif(
docker.utils.version_lt(test_version, version),
reason="API version is too low (< {0})".format(version)
)
def requires_experimental(until=None):
test_version = os.environ.get(
'DOCKER_TEST_API_VERSION', docker.constants.DEFAULT_DOCKER_API_VERSION
)
def req_exp(f):
@functools.wraps(f)
def wrapped(self, *args, **kwargs):
if not self.client.info()['ExperimentalBuild']:
pytest.skip('Feature requires Docker Engine experimental mode')
return f(self, *args, **kwargs)
if until and docker.utils.version_gte(test_version, until):
return f
return wrapped
return req_exp
def wait_on_condition(condition, delay=0.1, timeout=40):
start_time = time.time()
while not condition():
if time.time() - start_time > timeout:
raise AssertionError("Timeout: %s" % condition)
time.sleep(delay)
def random_name():
return u'dockerpytest_{0:x}'.format(random.getrandbits(64))
def force_leave_swarm(client):
"""Actually force leave a Swarm. There seems to be a bug in Swarm that
occasionally throws "context deadline exceeded" errors when leaving."""
while True:
try:
if isinstance(client, docker.DockerClient):
return client.swarm.leave(force=True)
return client.leave_swarm(force=True) # elif APIClient
except docker.errors.APIError as e:
if e.explanation == "context deadline exceeded":
continue
else:
return
def swarm_listen_addr():
return '0.0.0.0:{0}'.format(random.randrange(10000, 25000))
def assert_cat_socket_detached_with_keys(sock, inputs):
if six.PY3 and hasattr(sock, '_sock'):
sock = sock._sock
for i in inputs:
sock.sendall(i)
time.sleep(0.5)
# If we're using a Unix socket, the sock.send call will fail with a
# BrokenPipeError ; INET sockets will just stop receiving / sending data
# but will not raise an error
if isinstance(sock, paramiko.Channel):
with pytest.raises(OSError):
sock.sendall(b'make sure the socket is closed\n')
else:
if getattr(sock, 'family', -9) == getattr(socket, 'AF_UNIX', -1):
# We do not want to use pytest.raises here because future versions
# of the daemon no longer cause this to raise an error.
try:
sock.sendall(b'make sure the socket is closed\n')
except socket.error:
return
sock.sendall(b"make sure the socket is closed\n")
data = sock.recv(128)
# New in 18.06: error message is broadcast over the socket when reading
# after detach
assert data == b'' or data.startswith(
b'exec attach failed: error on attach stdin: read escape sequence'
)
def ctrl_with(char):
if re.match('[a-z]', char):
return chr(ord(char) - ord('a') + 1).encode('ascii')
else:
raise(Exception('char must be [a-z]'))
|
