summaryrefslogtreecommitdiff
path: root/tests/unit
diff options
context:
space:
mode:
authorJoffrey F <joffrey@docker.com>2018-01-29 19:10:12 -0800
committerJoffrey F <joffrey@docker.com>2018-01-30 14:26:24 -0800
commit342221130918f4525f01e31d3697cfc077df090e (patch)
treeecec590176339d863a51d78837c231bb4fa735fd /tests/unit
parent4ff296247b4ed9c4afbb0b5293bd4deecb4fe708 (diff)
downloaddocker-py-pytest-asserts.tar.gz
Use pytest assertspytest-asserts
Signed-off-by: Joffrey F <joffrey@docker.com>
Diffstat (limited to 'tests/unit')
-rw-r--r--tests/unit/api_build_test.py13
-rw-r--r--tests/unit/api_container_test.py846
-rw-r--r--tests/unit/api_exec_test.py90
-rw-r--r--tests/unit/api_image_test.py38
-rw-r--r--tests/unit/api_network_test.py97
-rw-r--r--tests/unit/api_test.py79
-rw-r--r--tests/unit/api_volume_test.py68
-rw-r--r--tests/unit/auth_test.py284
-rw-r--r--tests/unit/client_test.py23
-rw-r--r--tests/unit/dockertypes_test.py194
-rw-r--r--tests/unit/models_containers_test.py15
-rw-r--r--tests/unit/ssladapter_test.py11
-rw-r--r--tests/unit/swarm_test.py12
-rw-r--r--tests/unit/utils_test.py303
14 files changed, 878 insertions, 1195 deletions
diff --git a/tests/unit/api_build_test.py b/tests/unit/api_build_test.py
index 927aa97..e366bce 100644
--- a/tests/unit/api_build_test.py
+++ b/tests/unit/api_build_test.py
@@ -5,6 +5,7 @@ import docker
from docker import auth
from .api_test import BaseAPIClientTest, fake_request, url_prefix
+import pytest
class BuildTest(BaseAPIClientTest):
@@ -110,12 +111,10 @@ class BuildTest(BaseAPIClientTest):
})
def test_build_container_invalid_container_limits(self):
- self.assertRaises(
- docker.errors.DockerException,
- lambda: self.client.build('.', container_limits={
+ with pytest.raises(docker.errors.DockerException):
+ self.client.build('.', container_limits={
'foo': 'bar'
})
- )
def test_set_auth_headers_with_empty_dict_and_auth_configs(self):
self.client._auth_configs = {
@@ -130,7 +129,7 @@ class BuildTest(BaseAPIClientTest):
expected_headers = {
'X-Registry-Config': auth.encode_header(self.client._auth_configs)}
self.client._set_auth_headers(headers)
- self.assertEqual(headers, expected_headers)
+ assert headers == expected_headers
def test_set_auth_headers_with_dict_and_auth_configs(self):
self.client._auth_configs = {
@@ -147,7 +146,7 @@ class BuildTest(BaseAPIClientTest):
'X-Registry-Config': auth.encode_header(self.client._auth_configs)}
self.client._set_auth_headers(headers)
- self.assertEqual(headers, expected_headers)
+ assert headers == expected_headers
def test_set_auth_headers_with_dict_and_no_auth_configs(self):
headers = {'foo': 'bar'}
@@ -156,4 +155,4 @@ class BuildTest(BaseAPIClientTest):
}
self.client._set_auth_headers(headers)
- self.assertEqual(headers, expected_headers)
+ assert headers == expected_headers
diff --git a/tests/unit/api_container_test.py b/tests/unit/api_container_test.py
index 8a897cc..3cb718a 100644
--- a/tests/unit/api_container_test.py
+++ b/tests/unit/api_container_test.py
@@ -30,31 +30,20 @@ class StartContainerTest(BaseAPIClientTest):
self.client.start(fake_api.FAKE_CONTAINER_ID)
args = fake_request.call_args
- self.assertEqual(
- args[0][1],
- url_prefix + 'containers/3cc2351ab11b/start'
- )
+ assert args[0][1] == url_prefix + 'containers/3cc2351ab11b/start'
assert 'data' not in args[1]
- self.assertEqual(
- args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS
- )
+ assert args[1]['timeout'] == DEFAULT_TIMEOUT_SECONDS
def test_start_container_none(self):
with pytest.raises(ValueError) as excinfo:
self.client.start(container=None)
- self.assertEqual(
- str(excinfo.value),
- 'Resource ID was not provided',
- )
+ assert str(excinfo.value) == 'Resource ID was not provided'
with pytest.raises(ValueError) as excinfo:
self.client.start(None)
- self.assertEqual(
- str(excinfo.value),
- 'Resource ID was not provided',
- )
+ assert str(excinfo.value) == 'Resource ID was not provided'
def test_start_container_regression_573(self):
self.client.start(**{'container': fake_api.FAKE_CONTAINER_ID})
@@ -134,14 +123,9 @@ class StartContainerTest(BaseAPIClientTest):
self.client.start({'Id': fake_api.FAKE_CONTAINER_ID})
args = fake_request.call_args
- self.assertEqual(
- args[0][1],
- url_prefix + 'containers/3cc2351ab11b/start'
- )
+ assert args[0][1] == url_prefix + 'containers/3cc2351ab11b/start'
assert 'data' not in args[1]
- self.assertEqual(
- args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS
- )
+ assert args[1]['timeout'] == DEFAULT_TIMEOUT_SECONDS
class CreateContainerTest(BaseAPIClientTest):
@@ -149,17 +133,15 @@ class CreateContainerTest(BaseAPIClientTest):
self.client.create_container('busybox', 'true')
args = fake_request.call_args
- self.assertEqual(args[0][1],
- url_prefix + 'containers/create')
- self.assertEqual(json.loads(args[1]['data']),
- json.loads('''
- {"Tty": false, "Image": "busybox", "Cmd": ["true"],
- "AttachStdin": false,
- "AttachStderr": true, "AttachStdout": true,
- "StdinOnce": false,
- "OpenStdin": false, "NetworkDisabled": false}'''))
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
+ assert args[0][1] == url_prefix + 'containers/create'
+ assert json.loads(args[1]['data']) == json.loads('''
+ {"Tty": false, "Image": "busybox", "Cmd": ["true"],
+ "AttachStdin": false,
+ "AttachStderr": true, "AttachStdout": true,
+ "StdinOnce": false,
+ "OpenStdin": false, "NetworkDisabled": false}
+ ''')
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
def test_create_container_with_binds(self):
mount_dest = '/mnt'
@@ -168,19 +150,17 @@ class CreateContainerTest(BaseAPIClientTest):
volumes=[mount_dest])
args = fake_request.call_args
- self.assertEqual(args[0][1],
- url_prefix + 'containers/create')
- self.assertEqual(json.loads(args[1]['data']),
- json.loads('''
- {"Tty": false, "Image": "busybox",
- "Cmd": ["ls", "/mnt"], "AttachStdin": false,
- "Volumes": {"/mnt": {}},
- "AttachStderr": true,
- "AttachStdout": true, "OpenStdin": false,
- "StdinOnce": false,
- "NetworkDisabled": false}'''))
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
+ assert args[0][1] == url_prefix + 'containers/create'
+ assert json.loads(args[1]['data']) == json.loads('''
+ {"Tty": false, "Image": "busybox",
+ "Cmd": ["ls", "/mnt"], "AttachStdin": false,
+ "Volumes": {"/mnt": {}},
+ "AttachStderr": true,
+ "AttachStdout": true, "OpenStdin": false,
+ "StdinOnce": false,
+ "NetworkDisabled": false}
+ ''')
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
def test_create_container_with_volume_string(self):
mount_dest = '/mnt'
@@ -189,80 +169,72 @@ class CreateContainerTest(BaseAPIClientTest):
volumes=mount_dest)
args = fake_request.call_args
- self.assertEqual(args[0][1],
- url_prefix + 'containers/create')
- self.assertEqual(json.loads(args[1]['data']),
- json.loads('''
- {"Tty": false, "Image": "busybox",
- "Cmd": ["ls", "/mnt"], "AttachStdin": false,
- "Volumes": {"/mnt": {}},
- "AttachStderr": true,
- "AttachStdout": true, "OpenStdin": false,
- "StdinOnce": false,
- "NetworkDisabled": false}'''))
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
+ assert args[0][1] == url_prefix + 'containers/create'
+ assert json.loads(args[1]['data']) == json.loads('''
+ {"Tty": false, "Image": "busybox",
+ "Cmd": ["ls", "/mnt"], "AttachStdin": false,
+ "Volumes": {"/mnt": {}},
+ "AttachStderr": true,
+ "AttachStdout": true, "OpenStdin": false,
+ "StdinOnce": false,
+ "NetworkDisabled": false}
+ ''')
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
def test_create_container_with_ports(self):
self.client.create_container('busybox', 'ls',
ports=[1111, (2222, 'udp'), (3333,)])
args = fake_request.call_args
- self.assertEqual(args[0][1],
- url_prefix + 'containers/create')
- self.assertEqual(json.loads(args[1]['data']),
- json.loads('''
- {"Tty": false, "Image": "busybox",
- "Cmd": ["ls"], "AttachStdin": false,
- "ExposedPorts": {
- "1111/tcp": {},
- "2222/udp": {},
- "3333/tcp": {}
- },
- "AttachStderr": true,
- "AttachStdout": true, "OpenStdin": false,
- "StdinOnce": false,
- "NetworkDisabled": false}'''))
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
+ assert args[0][1] == url_prefix + 'containers/create'
+ assert json.loads(args[1]['data']) == json.loads('''
+ {"Tty": false, "Image": "busybox",
+ "Cmd": ["ls"], "AttachStdin": false,
+ "ExposedPorts": {
+ "1111/tcp": {},
+ "2222/udp": {},
+ "3333/tcp": {}
+ },
+ "AttachStderr": true,
+ "AttachStdout": true, "OpenStdin": false,
+ "StdinOnce": false,
+ "NetworkDisabled": false}
+ ''')
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
def test_create_container_with_entrypoint(self):
self.client.create_container('busybox', 'hello',
entrypoint='cowsay entry')
args = fake_request.call_args
- self.assertEqual(args[0][1],
- url_prefix + 'containers/create')
- self.assertEqual(json.loads(args[1]['data']),
- json.loads('''
- {"Tty": false, "Image": "busybox",
- "Cmd": ["hello"], "AttachStdin": false,
- "AttachStderr": true,
- "AttachStdout": true, "OpenStdin": false,
- "StdinOnce": false,
- "NetworkDisabled": false,
- "Entrypoint": ["cowsay", "entry"]}'''))
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
+ assert args[0][1] == url_prefix + 'containers/create'
+ assert json.loads(args[1]['data']) == json.loads('''
+ {"Tty": false, "Image": "busybox",
+ "Cmd": ["hello"], "AttachStdin": false,
+ "AttachStderr": true,
+ "AttachStdout": true, "OpenStdin": false,
+ "StdinOnce": false,
+ "NetworkDisabled": false,
+ "Entrypoint": ["cowsay", "entry"]}
+ ''')
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
def test_create_container_with_cpu_shares(self):
with pytest.deprecated_call():
self.client.create_container('busybox', 'ls', cpu_shares=5)
args = fake_request.call_args
- self.assertEqual(args[0][1],
- url_prefix + 'containers/create')
- self.assertEqual(json.loads(args[1]['data']),
- json.loads('''
- {"Tty": false, "Image": "busybox",
- "Cmd": ["ls"], "AttachStdin": false,
- "AttachStderr": true,
- "AttachStdout": true, "OpenStdin": false,
- "StdinOnce": false,
- "NetworkDisabled": false,
- "CpuShares": 5}'''))
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
+ assert args[0][1] == url_prefix + 'containers/create'
+ assert json.loads(args[1]['data']) == json.loads('''
+ {"Tty": false, "Image": "busybox",
+ "Cmd": ["ls"], "AttachStdin": false,
+ "AttachStderr": true,
+ "AttachStdout": true, "OpenStdin": false,
+ "StdinOnce": false,
+ "NetworkDisabled": false,
+ "CpuShares": 5}
+ ''')
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
@requires_api_version('1.18')
def test_create_container_with_host_config_cpu_shares(self):
@@ -273,43 +245,39 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1],
- url_prefix + 'containers/create')
-
- self.assertEqual(json.loads(args[1]['data']),
- json.loads('''
- {"Tty": false, "Image": "busybox",
- "Cmd": ["ls"], "AttachStdin": false,
- "AttachStderr": true,
- "AttachStdout": true, "OpenStdin": false,
- "StdinOnce": false,
- "NetworkDisabled": false,
- "HostConfig": {
- "CpuShares": 512,
- "NetworkMode": "default"
- }}'''))
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
+ assert args[0][1] == url_prefix + 'containers/create'
+
+ assert json.loads(args[1]['data']) == json.loads('''
+ {"Tty": false, "Image": "busybox",
+ "Cmd": ["ls"], "AttachStdin": false,
+ "AttachStderr": true,
+ "AttachStdout": true, "OpenStdin": false,
+ "StdinOnce": false,
+ "NetworkDisabled": false,
+ "HostConfig": {
+ "CpuShares": 512,
+ "NetworkMode": "default"
+ }}
+ ''')
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
def test_create_container_with_cpuset(self):
with pytest.deprecated_call():
self.client.create_container('busybox', 'ls', cpuset='0,1')
args = fake_request.call_args
- self.assertEqual(args[0][1],
- url_prefix + 'containers/create')
- self.assertEqual(json.loads(args[1]['data']),
- json.loads('''
- {"Tty": false, "Image": "busybox",
- "Cmd": ["ls"], "AttachStdin": false,
- "AttachStderr": true,
- "AttachStdout": true, "OpenStdin": false,
- "StdinOnce": false,
- "NetworkDisabled": false,
- "Cpuset": "0,1",
- "CpusetCpus": "0,1"}'''))
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
+ assert args[0][1] == url_prefix + 'containers/create'
+ assert json.loads(args[1]['data']) == json.loads('''
+ {"Tty": false, "Image": "busybox",
+ "Cmd": ["ls"], "AttachStdin": false,
+ "AttachStderr": true,
+ "AttachStdout": true, "OpenStdin": false,
+ "StdinOnce": false,
+ "NetworkDisabled": false,
+ "Cpuset": "0,1",
+ "CpusetCpus": "0,1"}
+ ''')
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
@requires_api_version('1.18')
def test_create_container_with_host_config_cpuset(self):
@@ -320,23 +288,21 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1],
- url_prefix + 'containers/create')
-
- self.assertEqual(json.loads(args[1]['data']),
- json.loads('''
- {"Tty": false, "Image": "busybox",
- "Cmd": ["ls"], "AttachStdin": false,
- "AttachStderr": true,
- "AttachStdout": true, "OpenStdin": false,
- "StdinOnce": false,
- "NetworkDisabled": false,
- "HostConfig": {
- "CpusetCpus": "0,1",
- "NetworkMode": "default"
- }}'''))
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
+ assert args[0][1] == url_prefix + 'containers/create'
+
+ assert json.loads(args[1]['data']) == json.loads('''
+ {"Tty": false, "Image": "busybox",
+ "Cmd": ["ls"], "AttachStdin": false,
+ "AttachStderr": true,
+ "AttachStdout": true, "OpenStdin": false,
+ "StdinOnce": false,
+ "NetworkDisabled": false,
+ "HostConfig": {
+ "CpusetCpus": "0,1",
+ "NetworkMode": "default"
+ }}
+ ''')
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
@requires_api_version('1.19')
def test_create_container_with_host_config_cpuset_mems(self):
@@ -347,23 +313,21 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1],
- url_prefix + 'containers/create')
-
- self.assertEqual(json.loads(args[1]['data']),
- json.loads('''
- {"Tty": false, "Image": "busybox",
- "Cmd": ["ls"], "AttachStdin": false,
- "AttachStderr": true,
- "AttachStdout": true, "OpenStdin": false,
- "StdinOnce": false,
- "NetworkDisabled": false,
- "HostConfig": {
- "CpusetMems": "0",
- "NetworkMode": "default"
- }}'''))
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
+ assert args[0][1] == url_prefix + 'containers/create'
+
+ assert json.loads(args[1]['data']) == json.loads('''
+ {"Tty": false, "Image": "busybox",
+ "Cmd": ["ls"], "AttachStdin": false,
+ "AttachStderr": true,
+ "AttachStdout": true, "OpenStdin": false,
+ "StdinOnce": false,
+ "NetworkDisabled": false,
+ "HostConfig": {
+ "CpusetMems": "0",
+ "NetworkMode": "default"
+ }}
+ ''')
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
def test_create_container_with_cgroup_parent(self):
self.client.create_container(
@@ -373,47 +337,42 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1],
- url_prefix + 'containers/create')
+ assert args[0][1] == url_prefix + 'containers/create'
data = json.loads(args[1]['data'])
- self.assertIn('HostConfig', data)
- self.assertIn('CgroupParent', data['HostConfig'])
- self.assertEqual(data['HostConfig']['CgroupParent'], 'test')
+ assert 'HostConfig' in data
+ assert 'CgroupParent' in data['HostConfig']
+ assert data['HostConfig']['CgroupParent'] == 'test'
def test_create_container_with_working_dir(self):
self.client.create_container('busybox', 'ls',
working_dir='/root')
args = fake_request.call_args
- self.assertEqual(args[0][1],
- url_prefix + 'containers/create')
- self.assertEqual(json.loads(args[1]['data']),
- json.loads('''
- {"Tty": false, "Image": "busybox",
- "Cmd": ["ls"], "AttachStdin": false,
- "AttachStderr": true,
- "AttachStdout": true, "OpenStdin": false,
- "StdinOnce": false,
- "NetworkDisabled": false,
- "WorkingDir": "/root"}'''))
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
+ assert args[0][1] == url_prefix + 'containers/create'
+ assert json.loads(args[1]['data']) == json.loads('''
+ {"Tty": false, "Image": "busybox",
+ "Cmd": ["ls"], "AttachStdin": false,
+ "AttachStderr": true,
+ "AttachStdout": true, "OpenStdin": false,
+ "StdinOnce": false,
+ "NetworkDisabled": false,
+ "WorkingDir": "/root"}
+ ''')
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
def test_create_container_with_stdin_open(self):
self.client.create_container('busybox', 'true', stdin_open=True)
args = fake_request.call_args
- self.assertEqual(args[0][1],
- url_prefix + 'containers/create')
- self.assertEqual(json.loads(args[1]['data']),
- json.loads('''
- {"Tty": false, "Image": "busybox", "Cmd": ["true"],
- "AttachStdin": true,
- "AttachStderr": true, "AttachStdout": true,
- "StdinOnce": true,
- "OpenStdin": true, "NetworkDisabled": false}'''))
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
+ assert args[0][1] == url_prefix + 'containers/create'
+ assert json.loads(args[1]['data']) == json.loads('''
+ {"Tty": false, "Image": "busybox", "Cmd": ["true"],
+ "AttachStdin": true,
+ "AttachStderr": true, "AttachStdout": true,
+ "StdinOnce": true,
+ "OpenStdin": true, "NetworkDisabled": false}
+ ''')
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
def test_create_container_with_volumes_from(self):
vol_names = ['foo', 'bar']
@@ -421,17 +380,17 @@ class CreateContainerTest(BaseAPIClientTest):
self.client.create_container('busybox', 'true',
volumes_from=vol_names)
except docker.errors.DockerException:
- self.assertTrue(
- docker.utils.compare_version('1.10', self.client._version) >= 0
- )
+ assert docker.utils.compare_version(
+ '1.10', self.client._version
+ ) >= 0
return
args = fake_request.call_args
- self.assertEqual(args[0][1], url_prefix + 'containers/create')
- self.assertEqual(json.loads(args[1]['data'])['VolumesFrom'],
- ','.join(vol_names))
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
+ assert args[0][1] == url_prefix + 'containers/create'
+ assert json.loads(args[1]['data'])['VolumesFrom'] == ','.join(
+ vol_names
+ )
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
def test_create_container_empty_volumes_from(self):
with pytest.raises(docker.errors.InvalidVersion):
@@ -442,18 +401,16 @@ class CreateContainerTest(BaseAPIClientTest):
name='marisa-kirisame')
args = fake_request.call_args
- self.assertEqual(args[0][1],
- url_prefix + 'containers/create')
- self.assertEqual(json.loads(args[1]['data']),
- json.loads('''
- {"Tty": false, "Image": "busybox", "Cmd": ["true"],
- "AttachStdin": false,
- "AttachStderr": true, "AttachStdout": true,
- "StdinOnce": false,
- "OpenStdin": false, "NetworkDisabled": false}'''))
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
- self.assertEqual(args[1]['params'], {'name': 'marisa-kirisame'})
+ assert args[0][1] == url_prefix + 'containers/create'
+ assert json.loads(args[1]['data']) == json.loads('''
+ {"Tty": false, "Image": "busybox", "Cmd": ["true"],
+ "AttachStdin": false,
+ "AttachStderr": true, "AttachStdout": true,
+ "StdinOnce": false,
+ "OpenStdin": false, "NetworkDisabled": false}
+ ''')
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
+ assert args[1]['params'] == {'name': 'marisa-kirisame'}
def test_create_container_with_mem_limit_as_int(self):
self.client.create_container(
@@ -464,7 +421,7 @@ class CreateContainerTest(BaseAPIClientTest):
args = fake_request.call_args
data = json.loads(args[1]['data'])
- self.assertEqual(data['HostConfig']['Memory'], 128.0)
+ assert data['HostConfig']['Memory'] == 128.0
def test_create_container_with_mem_limit_as_string(self):
self.client.create_container(
@@ -475,7 +432,7 @@ class CreateContainerTest(BaseAPIClientTest):
args = fake_request.call_args
data = json.loads(args[1]['data'])
- self.assertEqual(data['HostConfig']['Memory'], 128.0)
+ assert data['HostConfig']['Memory'] == 128.0
def test_create_container_with_mem_limit_as_string_with_k_unit(self):
self.client.create_container(
@@ -486,7 +443,7 @@ class CreateContainerTest(BaseAPIClientTest):
args = fake_request.call_args
data = json.loads(args[1]['data'])
- self.assertEqual(data['HostConfig']['Memory'], 128.0 * 1024)
+ assert data['HostConfig']['Memory'] == 128.0 * 1024
def test_create_container_with_mem_limit_as_string_with_m_unit(self):
self.client.create_container(
@@ -497,7 +454,7 @@ class CreateContainerTest(BaseAPIClientTest):
args = fake_request.call_args
data = json.loads(args[1]['data'])
- self.assertEqual(data['HostConfig']['Memory'], 128.0 * 1024 * 1024)
+ assert data['HostConfig']['Memory'] == 128.0 * 1024 * 1024
def test_create_container_with_mem_limit_as_string_with_g_unit(self):
self.client.create_container(
@@ -508,20 +465,14 @@ class CreateContainerTest(BaseAPIClientTest):
args = fake_request.call_args
data = json.loads(args[1]['data'])
- self.assertEqual(
- data['HostConfig']['Memory'], 128.0 * 1024 * 1024 * 1024
- )
+ assert data['HostConfig']['Memory'] == 128.0 * 1024 * 1024 * 1024
def test_create_container_with_mem_limit_as_string_with_wrong_value(self):
- self.assertRaises(
- docker.errors.DockerException,
- self.client.create_host_config, mem_limit='128p'
- )
+ with pytest.raises(docker.errors.DockerException):
+ self.client.create_host_config(mem_limit='128p')
- self.assertRaises(
- docker.errors.DockerException,
- self.client.create_host_config, mem_limit='1f28'
- )
+ with pytest.raises(docker.errors.DockerException):
+ self.client.create_host_config(mem_limit='1f28')
def test_create_container_with_lxc_conf(self):
self.client.create_container(
@@ -531,25 +482,16 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(
- args[0][1],
- url_prefix + 'containers/create'
- )
+ assert args[0][1] == url_prefix + 'containers/create'
expected_payload = self.base_create_payload()
expected_payload['HostConfig'] = self.client.create_host_config()
expected_payload['HostConfig']['LxcConf'] = [
{"Value": "lxc.conf.value", "Key": "lxc.conf.k"}
]
- self.assertEqual(json.loads(args[1]['data']), expected_payload)
- self.assertEqual(
- args[1]['headers'],
- {'Content-Type': 'application/json'}
- )
- self.assertEqual(
- args[1]['timeout'],
- DEFAULT_TIMEOUT_SECONDS
- )
+ assert json.loads(args[1]['data']) == expected_payload
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
+ assert args[1]['timeout'] == DEFAULT_TIMEOUT_SECONDS
def test_create_container_with_lxc_conf_compat(self):
self.client.create_container(
@@ -559,20 +501,15 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1], url_prefix + 'containers/create')
+ assert args[0][1] == url_prefix + 'containers/create'
expected_payload = self.base_create_payload()
expected_payload['HostConfig'] = self.client.create_host_config()
expected_payload['HostConfig']['LxcConf'] = [
{"Value": "lxc.conf.value", "Key": "lxc.conf.k"}
]
- self.assertEqual(
- json.loads(args[1]['data']), expected_payload)
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
- self.assertEqual(
- args[1]['timeout'],
- DEFAULT_TIMEOUT_SECONDS
- )
+ assert json.loads(args[1]['data']) == expected_payload
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
+ assert args[1]['timeout'] == DEFAULT_TIMEOUT_SECONDS
def test_create_container_with_binds_ro(self):
mount_dest = '/mnt'
@@ -588,18 +525,13 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1], url_prefix +
- 'containers/create')
+ assert args[0][1] == url_prefix + 'containers/create'
expected_payload = self.base_create_payload()
expected_payload['HostConfig'] = self.client.create_host_config()
expected_payload['HostConfig']['Binds'] = ["/tmp:/mnt:ro"]
- self.assertEqual(json.loads(args[1]['data']), expected_payload)
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
- self.assertEqual(
- args[1]['timeout'],
- DEFAULT_TIMEOUT_SECONDS
- )
+ assert json.loads(args[1]['data']) == expected_payload
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
+ assert args[1]['timeout'] == DEFAULT_TIMEOUT_SECONDS
def test_create_container_with_binds_rw(self):
mount_dest = '/mnt'
@@ -615,18 +547,13 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1], url_prefix +
- 'containers/create')
+ assert args[0][1] == url_prefix + 'containers/create'
expected_payload = self.base_create_payload()
expected_payload['HostConfig'] = self.client.create_host_config()
expected_payload['HostConfig']['Binds'] = ["/tmp:/mnt:rw"]
- self.assertEqual(json.loads(args[1]['data']), expected_payload)
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
- self.assertEqual(
- args[1]['timeout'],
- DEFAULT_TIMEOUT_SECONDS
- )
+ assert json.loads(args[1]['data']) == expected_payload
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
+ assert args[1]['timeout'] == DEFAULT_TIMEOUT_SECONDS
def test_create_container_with_binds_mode(self):
mount_dest = '/mnt'
@@ -642,18 +569,13 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1], url_prefix +
- 'containers/create')
+ assert args[0][1] == url_prefix + 'containers/create'
expected_payload = self.base_create_payload()
expected_payload['HostConfig'] = self.client.create_host_config()
expected_payload['HostConfig']['Binds'] = ["/tmp:/mnt:z"]
- self.assertEqual(json.loads(args[1]['data']), expected_payload)
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
- self.assertEqual(
- args[1]['timeout'],
- DEFAULT_TIMEOUT_SECONDS
- )
+ assert json.loads(args[1]['data']) == expected_payload
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
+ assert args[1]['timeout'] == DEFAULT_TIMEOUT_SECONDS
def test_create_container_with_binds_mode_and_ro_error(self):
with pytest.raises(ValueError):
@@ -680,21 +602,16 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1], url_prefix +
- 'containers/create')
+ assert args[0][1] == url_prefix + 'containers/create'
expected_payload = self.base_create_payload()
expected_payload['HostConfig'] = self.client.create_host_config()
expected_payload['HostConfig']['Binds'] = [
"/tmp:/mnt/1:ro",
"/tmp:/mnt/2",
]
- self.assertEqual(json.loads(args[1]['data']), expected_payload)
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
- self.assertEqual(
- args[1]['timeout'],
- DEFAULT_TIMEOUT_SECONDS
- )
+ assert json.loads(args[1]['data']) == expected_payload
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
+ assert args[1]['timeout'] == DEFAULT_TIMEOUT_SECONDS
def test_create_container_with_port_binds(self):
self.maxDiff = None
@@ -713,42 +630,31 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1], url_prefix + 'containers/create')
+ assert args[0][1] == url_prefix + 'containers/create'
data = json.loads(args[1]['data'])
port_bindings = data['HostConfig']['PortBindings']
- self.assertTrue('1111/tcp' in port_bindings)
- self.assertTrue('2222/tcp' in port_bindings)
- self.assertTrue('3333/udp' in port_bindings)
- self.assertTrue('4444/tcp' in port_bindings)
- self.assertTrue('5555/tcp' in port_bindings)
- self.assertTrue('6666/tcp' in port_bindings)
- self.assertEqual(
- [{"HostPort": "", "HostIp": ""}],
- port_bindings['1111/tcp']
- )
- self.assertEqual(
- [{"HostPort": "2222", "HostIp": ""}],
- port_bindings['2222/tcp']
- )
- self.assertEqual(
- [{"HostPort": "3333", "HostIp": ""}],
- port_bindings['3333/udp']
- )
- self.assertEqual(
- [{"HostPort": "", "HostIp": "127.0.0.1"}],
- port_bindings['4444/tcp']
- )
- self.assertEqual(
- [{"HostPort": "5555", "HostIp": "127.0.0.1"}],
- port_bindings['5555/tcp']
- )
- self.assertEqual(len(port_bindings['6666/tcp']), 2)
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
- self.assertEqual(
- args[1]['timeout'],
- DEFAULT_TIMEOUT_SECONDS
- )
+ assert '1111/tcp' in port_bindings
+ assert '2222/tcp' in port_bindings
+ assert '3333/udp' in port_bindings
+ assert '4444/tcp' in port_bindings
+ assert '5555/tcp' in port_bindings
+ assert '6666/tcp' in port_bindings
+ assert [{"HostPort": "", "HostIp": ""}] == port_bindings['1111/tcp']
+ assert [
+ {"HostPort": "2222", "HostIp": ""}
+ ] == port_bindings['2222/tcp']
+ assert [
+ {"HostPort": "3333", "HostIp": ""}
+ ] == port_bindings['3333/udp']
+ assert [
+ {"HostPort": "", "HostIp": "127.0.0.1"}
+ ] == port_bindings['4444/tcp']
+ assert [
+ {"HostPort": "5555", "HostIp": "127.0.0.1"}
+ ] == port_bindings['5555/tcp']
+ assert len(port_bindings['6666/tcp']) == 2
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
+ assert args[1]['timeout'] == DEFAULT_TIMEOUT_SECONDS
def test_create_container_with_mac_address(self):
expected = "02:42:ac:11:00:0a"
@@ -760,7 +666,7 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1], url_prefix + 'containers/create')
+ assert args[0][1] == url_prefix + 'containers/create'
data = json.loads(args[1]['data'])
assert data['MacAddress'] == expected
@@ -775,17 +681,13 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(
- args[0][1], url_prefix + 'containers/create'
- )
+ assert args[0][1] == url_prefix + 'containers/create'
expected_payload = self.base_create_payload()
expected_payload['HostConfig'] = self.client.create_host_config()
expected_payload['HostConfig']['Links'] = ['path:alias']
- self.assertEqual(json.loads(args[1]['data']), expected_payload)
- self.assertEqual(
- args[1]['headers'], {'Content-Type': 'application/json'}
- )
+ assert json.loads(args[1]['data']) == expected_payload
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
def test_create_container_with_multiple_links(self):
link_path = 'path'
@@ -801,16 +703,14 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1], url_prefix + 'containers/create')
+ assert args[0][1] == url_prefix + 'containers/create'
expected_payload = self.base_create_payload()
expected_payload['HostConfig'] = self.client.create_host_config()
expected_payload['HostConfig']['Links'] = [
'path1:alias1', 'path2:alias2'
]
- self.assertEqual(json.loads(args[1]['data']), expected_payload)
- self.assertEqual(
- args[1]['headers'], {'Content-Type': 'application/json'}
- )
+ assert json.loads(args[1]['data']) == expected_payload
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
def test_create_container_with_links_as_list_of_tuples(self):
link_path = 'path'
@@ -823,15 +723,13 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1], url_prefix + 'containers/create')
+ assert args[0][1] == url_prefix + 'containers/create'
expected_payload = self.base_create_payload()
expected_payload['HostConfig'] = self.client.create_host_config()
expected_payload['HostConfig']['Links'] = ['path:alias']
- self.assertEqual(json.loads(args[1]['data']), expected_payload)
- self.assertEqual(
- args[1]['headers'], {'Content-Type': 'application/json'}
- )
+ assert json.loads(args[1]['data']) == expected_payload
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
def test_create_container_privileged(self):
self.client.create_container(
@@ -843,14 +741,10 @@ class CreateContainerTest(BaseAPIClientTest):
expected_payload['HostConfig'] = self.client.create_host_config()
expected_payload['HostConfig']['Privileged'] = True
args = fake_request.call_args
- self.assertEqual(args[0][1], url_prefix + 'containers/create')
- self.assertEqual(json.loads(args[1]['data']), expected_payload)
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
- self.assertEqual(
- args[1]['timeout'],
- DEFAULT_TIMEOUT_SECONDS
- )
+ assert args[0][1] == url_prefix + 'containers/create'
+ assert json.loads(args[1]['data']) == expected_payload
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
+ assert args[1]['timeout'] == DEFAULT_TIMEOUT_SECONDS
def test_create_container_with_restart_policy(self):
self.client.create_container(
@@ -863,21 +757,17 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1], url_prefix + 'containers/create')
+ assert args[0][1] == url_prefix + 'containers/create'
expected_payload = self.base_create_payload()
expected_payload['HostConfig'] = self.client.create_host_config()
expected_payload['HostConfig']['RestartPolicy'] = {
"MaximumRetryCount": 0, "Name": "always"
}
- self.assertEqual(json.loads(args[1]['data']), expected_payload)
+ assert json.loads(args[1]['data']) == expected_payload
- self.assertEqual(
- args[1]['headers'], {'Content-Type': 'application/json'}
- )
- self.assertEqual(
- args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS
- )
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
+ assert args[1]['timeout'] == DEFAULT_TIMEOUT_SECONDS
def test_create_container_with_added_capabilities(self):
self.client.create_container(
@@ -886,17 +776,13 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1], url_prefix + 'containers/create')
+ assert args[0][1] == url_prefix + 'containers/create'
expected_payload = self.base_create_payload()
expected_payload['HostConfig'] = self.client.create_host_config()
expected_payload['HostConfig']['CapAdd'] = ['MKNOD']
- self.assertEqual(json.loads(args[1]['data']), expected_payload)
- self.assertEqual(
- args[1]['headers'], {'Content-Type': 'application/json'}
- )
- self.assertEqual(
- args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS
- )
+ assert json.loads(args[1]['data']) == expected_payload
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
+ assert args[1]['timeout'] == DEFAULT_TIMEOUT_SECONDS
def test_create_container_with_dropped_capabilities(self):
self.client.create_container(
@@ -905,17 +791,13 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1], url_prefix + 'containers/create')
+ assert args[0][1] == url_prefix + 'containers/create'
expected_payload = self.base_create_payload()
expected_payload['HostConfig'] = self.client.create_host_config()
expected_payload['HostConfig']['CapDrop'] = ['MKNOD']
- self.assertEqual(json.loads(args[1]['data']), expected_payload)
- self.assertEqual(
- args[1]['headers'], {'Content-Type': 'application/json'}
- )
- self.assertEqual(
- args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS
- )
+ assert json.loads(args[1]['data']) == expected_payload
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
+ assert args[1]['timeout'] == DEFAULT_TIMEOUT_SECONDS
def test_create_container_with_devices(self):
self.client.create_container(
@@ -927,7 +809,7 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1], url_prefix + 'containers/create')
+ assert args[0][1] == url_prefix + 'containers/create'
expected_payload = self.base_create_payload()
expected_payload['HostConfig'] = self.client.create_host_config()
expected_payload['HostConfig']['Devices'] = [
@@ -941,13 +823,9 @@ class CreateContainerTest(BaseAPIClientTest):
'PathInContainer': '/dev/sdc',
'PathOnHost': '/dev/sdc'}
]
- self.assertEqual(json.loads(args[1]['data']), expected_payload)
- self.assertEqual(
- args[1]['headers'], {'Content-Type': 'application/json'}
- )
- self.assertEqual(
- args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS
- )
+ assert json.loads(args[1]['data']) == expected_payload
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
+ assert args[1]['timeout'] == DEFAULT_TIMEOUT_SECONDS
def test_create_container_with_labels_dict(self):
labels_dict = {
@@ -961,14 +839,10 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1], url_prefix + 'containers/create')
- self.assertEqual(json.loads(args[1]['data'])['Labels'], labels_dict)
- self.assertEqual(
- args[1]['headers'], {'Content-Type': 'application/json'}
- )
- self.assertEqual(
- args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS
- )
+ assert args[0][1] == url_prefix + 'containers/create'
+ assert json.loads(args[1]['data'])['Labels'] == labels_dict
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
+ assert args[1]['timeout'] == DEFAULT_TIMEOUT_SECONDS
def test_create_container_with_labels_list(self):
labels_list = [
@@ -986,14 +860,10 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1], url_prefix + 'containers/create')
- self.assertEqual(json.loads(args[1]['data'])['Labels'], labels_dict)
- self.assertEqual(
- args[1]['headers'], {'Content-Type': 'application/json'}
- )
- self.assertEqual(
- args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS
- )
+ assert args[0][1] == url_prefix + 'containers/create'
+ assert json.loads(args[1]['data'])['Labels'] == labels_dict
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
+ assert args[1]['timeout'] == DEFAULT_TIMEOUT_SECONDS
def test_create_container_with_named_volume(self):
mount_dest = '/mnt'
@@ -1010,39 +880,31 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(
- args[0][1], url_prefix + 'containers/create'
- )
+ assert args[0][1] == url_prefix + 'containers/create'
expected_payload = self.base_create_payload()
expected_payload['HostConfig'] = self.client.create_host_config()
expected_payload['HostConfig']['VolumeDriver'] = 'foodriver'
expected_payload['HostConfig']['Binds'] = ["name:/mnt:rw"]
- self.assertEqual(json.loads(args[1]['data']), expected_payload)
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
- self.assertEqual(
- args[1]['timeout'],
- DEFAULT_TIMEOUT_SECONDS
- )
+ assert json.loads(args[1]['data']) == expected_payload
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
+ assert args[1]['timeout'] == DEFAULT_TIMEOUT_SECONDS
def test_create_container_with_stop_signal(self):
self.client.create_container('busybox', 'ls',
stop_signal='SIGINT')
args = fake_request.call_args
- self.assertEqual(args[0][1],
- url_prefix + 'containers/create')
- self.assertEqual(json.loads(args[1]['data']),
- json.loads('''
- {"Tty": false, "Image": "busybox",
- "Cmd": ["ls"], "AttachStdin": false,
- "AttachStderr": true,
- "AttachStdout": true, "OpenStdin": false,
- "StdinOnce": false,
- "NetworkDisabled": false,
- "StopSignal": "SIGINT"}'''))
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
+ assert args[0][1] == url_prefix + 'containers/create'
+ assert json.loads(args[1]['data']) == json.loads('''
+ {"Tty": false, "Image": "busybox",
+ "Cmd": ["ls"], "AttachStdin": false,
+ "AttachStderr": true,
+ "AttachStdout": true, "OpenStdin": false,
+ "StdinOnce": false,
+ "NetworkDisabled": false,
+ "StopSignal": "SIGINT"}
+ ''')
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
@requires_api_version('1.22')
def test_create_container_with_aliases(self):
@@ -1059,22 +921,22 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(json.loads(args[1]['data']),
- json.loads('''
- {"Tty": false, "Image": "busybox",
- "Cmd": ["ls"], "AttachStdin": false,
- "AttachStderr": true,
- "AttachStdout": true, "OpenStdin": false,
- "StdinOnce": false,
- "NetworkDisabled": false,
- "HostConfig": {
- "NetworkMode": "some-network"
- },
- "NetworkingConfig": {
- "EndpointsConfig": {
- "some-network": {"Aliases": ["foo", "bar"]}
- }
- }}'''))
+ assert json.loads(args[1]['data']) == json.loads('''
+ {"Tty": false, "Image": "busybox",
+ "Cmd": ["ls"], "AttachStdin": false,
+ "AttachStderr": true,
+ "AttachStdout": true, "OpenStdin": false,
+ "StdinOnce": false,
+ "NetworkDisabled": false,
+ "HostConfig": {
+ "NetworkMode": "some-network"
+ },
+ "NetworkingConfig": {
+ "EndpointsConfig": {
+ "some-network": {"Aliases": ["foo", "bar"]}
+ }
+ }}
+ ''')
@requires_api_version('1.22')
def test_create_container_with_tmpfs_list(self):
@@ -1089,21 +951,16 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1], url_prefix +
- 'containers/create')
+ assert args[0][1] == url_prefix + 'containers/create'
expected_payload = self.base_create_payload()
expected_payload['HostConfig'] = self.client.create_host_config()
expected_payload['HostConfig']['Tmpfs'] = {
"/tmp": "",
"/mnt": "size=3G,uid=100"
}
- self.assertEqual(json.loads(args[1]['data']), expected_payload)
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
- self.assertEqual(
- args[1]['timeout'],
- DEFAULT_TIMEOUT_SECONDS
- )
+ assert json.loads(args[1]['data']) == expected_payload
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
+ assert args[1]['timeout'] == DEFAULT_TIMEOUT_SECONDS
@requires_api_version('1.22')
def test_create_container_with_tmpfs_dict(self):
@@ -1118,21 +975,16 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1], url_prefix +
- 'containers/create')
+ assert args[0][1] == url_prefix + 'containers/create'
expected_payload = self.base_create_payload()
expected_payload['HostConfig'] = self.client.create_host_config()
expected_payload['HostConfig']['Tmpfs'] = {
"/tmp": "",
"/mnt": "size=3G,uid=100"
}
- self.assertEqual(json.loads(args[1]['data']), expected_payload)
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
- self.assertEqual(
- args[1]['timeout'],
- DEFAULT_TIMEOUT_SECONDS
- )
+ assert json.loads(args[1]['data']) == expected_payload
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
+ assert args[1]['timeout'] == DEFAULT_TIMEOUT_SECONDS
@requires_api_version('1.24')
def test_create_container_with_sysctl(self):
@@ -1147,19 +999,15 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1], url_prefix + 'containers/create')
+ assert args[0][1] == url_prefix + 'containers/create'
expected_payload = self.base_create_payload()
expected_payload['HostConfig'] = self.client.create_host_config()
expected_payload['HostConfig']['Sysctls'] = {
'net.core.somaxconn': '1024', 'net.ipv4.tcp_syncookies': '0',
}
- self.assertEqual(json.loads(args[1]['data']), expected_payload)
- self.assertEqual(
- args[1]['headers'], {'Content-Type': 'application/json'}
- )
- self.assertEqual(
- args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS
- )
+ assert json.loads(args[1]['data']) == expected_payload
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
+ assert args[1]['timeout'] == DEFAULT_TIMEOUT_SECONDS
def test_create_container_with_unicode_envvars(self):
envvars_dict = {
@@ -1176,8 +1024,8 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1], url_prefix + 'containers/create')
- self.assertEqual(json.loads(args[1]['data'])['Env'], expected)
+ assert args[0][1] == url_prefix + 'containers/create'
+ assert json.loads(args[1]['data'])['Env'] == expected
@requires_api_version('1.25')
def test_create_container_with_host_config_cpus(self):
@@ -1190,26 +1038,23 @@ class CreateContainerTest(BaseAPIClientTest):
)
args = fake_request.call_args
- self.assertEqual(args[0][1],
- url_prefix + 'containers/create')
-
- self.assertEqual(json.loads(args[1]['data']),
- json.loads('''
- {"Tty": false, "Image": "busybox",
- "Cmd": ["ls"], "AttachStdin": false,
- "AttachStderr": true,
- "AttachStdout": true, "OpenStdin": false,
- "StdinOnce": false,
- "NetworkDisabled": false,
- "HostConfig": {
- "CpuCount": 1,
- "CpuPercent": 20,
- "NanoCpus": 1000,
- "NetworkMode": "default"
- }}'''))
- self.assertEqual(
- args[1]['headers'], {'Content-Type': 'application/json'}
- )
+ assert args[0][1] == url_prefix + 'containers/create'
+
+ assert json.loads(args[1]['data']) == json.loads('''
+ {"Tty": false, "Image": "busybox",
+ "Cmd": ["ls"], "AttachStdin": false,
+ "AttachStderr": true,
+ "AttachStdout": true, "OpenStdin": false,
+ "StdinOnce": false,
+ "NetworkDisabled": false,
+ "HostConfig": {
+ "CpuCount": 1,
+ "CpuPercent": 20,
+ "NanoCpus": 1000,
+ "NetworkMode": "default"
+ }}
+ ''')
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
class ContainerTest(BaseAPIClientTest):
@@ -1291,10 +1136,7 @@ class ContainerTest(BaseAPIClientTest):
stream=False
)
- self.assertEqual(
- logs,
- 'Flowering Nights\n(Sakuya Iyazoi)\n'.encode('ascii')
- )
+ assert logs == 'Flowering Nights\n(Sakuya Iyazoi)\n'.encode('ascii')
def test_logs_with_dict_instead_of_id(self):
with mock.patch('docker.api.client.APIClient.inspect_container',
@@ -1310,10 +1152,7 @@ class ContainerTest(BaseAPIClientTest):
stream=False
)
- self.assertEqual(
- logs,
- 'Flowering Nights\n(Sakuya Iyazoi)\n'.encode('ascii')
- )
+ assert logs == 'Flowering Nights\n(Sakuya Iyazoi)\n'.encode('ascii')
def test_log_streaming(self):
with mock.patch('docker.api.client.APIClient.inspect_container',
@@ -1426,7 +1265,7 @@ class ContainerTest(BaseAPIClientTest):
def test_log_since_with_invalid_value_raises_error(self):
with mock.patch('docker.api.client.APIClient.inspect_container',
fake_inspect_container):
- with self.assertRaises(docker.errors.InvalidArgument):
+ with pytest.raises(docker.errors.InvalidArgument):
self.client.logs(fake_api.FAKE_CONTAINER_ID, stream=False,
follow=False, since=42.42)
@@ -1439,7 +1278,7 @@ class ContainerTest(BaseAPIClientTest):
self.client.logs(fake_api.FAKE_CONTAINER_ID,
follow=True, stream=True)
- self.assertTrue(m.called)
+ assert m.called
fake_request.assert_called_with(
'GET',
url_prefix + 'containers/3cc2351ab11b/logs',
@@ -1623,9 +1462,7 @@ class ContainerTest(BaseAPIClientTest):
with pytest.raises(docker.errors.NullResource) as excinfo:
self.client.inspect_container(arg)
- self.assertEqual(
- excinfo.value.args[0], 'Resource ID was not provided'
- )
+ assert excinfo.value.args[0] == 'Resource ID was not provided'
def test_container_stats(self):
self.client.stats(fake_api.FAKE_CONTAINER_ID)
@@ -1664,13 +1501,8 @@ class ContainerTest(BaseAPIClientTest):
blkio_weight=345
)
args = fake_request.call_args
- self.assertEqual(
- args[0][1], url_prefix + 'containers/3cc2351ab11b/update'
- )
- self.assertEqual(
- json.loads(args[1]['data']),
- {'Memory': 2 * 1024, 'CpuShares': 124, 'BlkioWeight': 345}
- )
- self.assertEqual(
- args[1]['headers']['Content-Type'], 'application/json'
- )
+ assert args[0][1] == url_prefix + 'containers/3cc2351ab11b/update'
+ assert json.loads(args[1]['data']) == {
+ 'Memory': 2 * 1024, 'CpuShares': 124, 'BlkioWeight': 345
+ }
+ assert args[1]['headers']['Content-Type'] == 'application/json'
diff --git a/tests/unit/api_exec_test.py b/tests/unit/api_exec_test.py
index 41ee940..a9d2dd5 100644
--- a/tests/unit/api_exec_test.py
+++ b/tests/unit/api_exec_test.py
@@ -11,85 +11,65 @@ class ExecTest(BaseAPIClientTest):
self.client.exec_create(fake_api.FAKE_CONTAINER_ID, ['ls', '-1'])
args = fake_request.call_args
- self.assertEqual(
- 'POST',
- args[0][0], url_prefix + 'containers/{0}/exec'.format(
- fake_api.FAKE_CONTAINER_ID
- )
+ assert 'POST' == args[0][0], url_prefix + 'containers/{0}/exec'.format(
+ fake_api.FAKE_CONTAINER_ID
)
- self.assertEqual(
- json.loads(args[1]['data']), {
- 'Tty': False,
- 'AttachStdout': True,
- 'Container': fake_api.FAKE_CONTAINER_ID,
- 'Cmd': ['ls', '-1'],
- 'Privileged': False,
- 'AttachStdin': False,
- 'AttachStderr': True,
- 'User': ''
- }
- )
+ assert json.loads(args[1]['data']) == {
+ 'Tty': False,
+ 'AttachStdout': True,
+ 'Container': fake_api.FAKE_CONTAINER_ID,
+ 'Cmd': ['ls', '-1'],
+ 'Privileged': False,
+ 'AttachStdin': False,
+ 'AttachStderr': True,
+ 'User': ''
+ }
- self.assertEqual(args[1]['headers'],
- {'Content-Type': 'application/json'})
+ assert args[1]['headers'] == {'Content-Type': 'application/json'}
def test_exec_start(self):
self.client.exec_start(fake_api.FAKE_EXEC_ID)
args = fake_request.call_args
- self.assertEqual(
- args[0][1], url_prefix + 'exec/{0}/start'.format(
- fake_api.FAKE_EXEC_ID
- )
+ assert args[0][1] == url_prefix + 'exec/{0}/start'.format(
+ fake_api.FAKE_EXEC_ID
)
- self.assertEqual(
- json.loads(args[1]['data']), {
- 'Tty': False,
- 'Detach': False,
- }
- )
+ assert json.loads(args[1]['data']) == {
+ 'Tty': False,
+ 'Detach': False,
+ }
- self.assertEqual(
- args[1]['headers'], {
- 'Content-Type': 'application/json',
- 'Connection': 'Upgrade',
- 'Upgrade': 'tcp'
- }
- )
+ assert args[1]['headers'] == {
+ 'Content-Type': 'application/json',
+ 'Connection': 'Upgrade',
+ 'Upgrade': 'tcp'
+ }
def test_exec_start_detached(self):
self.client.exec_start(fake_api.FAKE_EXEC_ID, detach=True)
args = fake_request.call_args
- self.assertEqual(
- args[0][1], url_prefix + 'exec/{0}/start'.format(
- fake_api.FAKE_EXEC_ID
- )
+ assert args[0][1] == url_prefix + 'exec/{0}/start'.format(
+ fake_api.FAKE_EXEC_ID
)
- self.assertEqual(
- json.loads(args[1]['data']), {
- 'Tty': False,
- 'Detach': True
- }
- )
+ assert json.loads(args[1]['data']) == {
+ 'Tty': False,
+ 'Detach': True
+ }
- self.assertEqual(
- args[1]['headers'], {
- 'Content-Type': 'application/json'
- }
- )
+ assert args[1]['headers'] == {
+ 'Content-Type': 'application/json'
+ }
def test_exec_inspect(self):
self.client.exec_inspect(fake_api.FAKE_EXEC_ID)
args = fake_request.call_args
- self.assertEqual(
- args[0][1], url_prefix + 'exec/{0}/json'.format(
- fake_api.FAKE_EXEC_ID
- )
+ assert args[0][1] == url_prefix + 'exec/{0}/json'.format(
+ fake_api.FAKE_EXEC_ID
)
def test_exec_resize(self):
diff --git a/tests/unit/api_image_test.py b/tests/unit/api_image_test.py
index f1e42cc..785f887 100644
--- a/tests/unit/api_image_test.py
+++ b/tests/unit/api_image_test.py
@@ -65,29 +65,21 @@ class ImageTest(BaseAPIClientTest):
self.client.pull('joffrey/test001')
args = fake_request.call_args
- self.assertEqual(
- args[0][1],
- url_prefix + 'images/create'
- )
- self.assertEqual(
- args[1]['params'],
- {'tag': None, 'fromImage': 'joffrey/test001'}
- )
- self.assertFalse(args[1]['stream'])
+ assert args[0][1] == url_prefix + 'images/create'
+ assert args[1]['params'] == {
+ 'tag': None, 'fromImage': 'joffrey/test001'
+ }
+ assert not args[1]['stream']
def test_pull_stream(self):
self.client.pull('joffrey/test001', stream=True)
args = fake_request.call_args
- self.assertEqual(
- args[0][1],
- url_prefix + 'images/create'
- )
- self.assertEqual(
- args[1]['params'],
- {'tag': None, 'fromImage': 'joffrey/test001'}
- )
- self.assertTrue(args[1]['stream'])
+ assert args[0][1] == url_prefix + 'images/create'
+ assert args[1]['params'] == {
+ 'tag': None, 'fromImage': 'joffrey/test001'
+ }
+ assert args[1]['stream']
def test_commit(self):
self.client.commit(fake_api.FAKE_CONTAINER_ID)
@@ -203,18 +195,16 @@ class ImageTest(BaseAPIClientTest):
with pytest.raises(docker.errors.NullResource) as excinfo:
self.client.inspect_image(arg)
- self.assertEqual(
- excinfo.value.args[0], 'Resource ID was not provided'
- )
+ assert excinfo.value.args[0] == 'Resource ID was not provided'
def test_insert_image(self):
try:
self.client.insert(fake_api.FAKE_IMAGE_NAME,
fake_api.FAKE_URL, fake_api.FAKE_PATH)
except docker.errors.DeprecatedMethod:
- self.assertTrue(
- docker.utils.compare_version('1.12', self.client._version) >= 0
- )
+ assert docker.utils.compare_version(
+ '1.12', self.client._version
+ ) >= 0
return
fake_request.assert_called_with(
diff --git a/tests/unit/api_network_test.py b/tests/unit/api_network_test.py
index 96cdc4b..fbbc97b 100644
--- a/tests/unit/api_network_test.py
+++ b/tests/unit/api_network_test.py
@@ -34,20 +34,20 @@ class NetworkTest(BaseAPIClientTest):
status_code=200, content=json.dumps(networks).encode('utf-8')))
with mock.patch('docker.api.client.APIClient.get', get):
- self.assertEqual(self.client.networks(), networks)
+ assert self.client.networks() == networks
- self.assertEqual(get.call_args[0][0], url_prefix + 'networks')
+ assert get.call_args[0][0] == url_prefix + 'networks'
filters = json.loads(get.call_args[1]['params']['filters'])
- self.assertFalse(filters)
+ assert not filters
self.client.networks(names=['foo'])
filters = json.loads(get.call_args[1]['params']['filters'])
- self.assertEqual(filters, {'name': ['foo']})
+ assert filters == {'name': ['foo']}
self.client.networks(ids=['123'])
filters = json.loads(get.call_args[1]['params']['filters'])
- self.assertEqual(filters, {'id': ['123']})
+ assert filters == {'id': ['123']}
@requires_api_version('1.21')
def test_create_network(self):
@@ -61,15 +61,11 @@ class NetworkTest(BaseAPIClientTest):
with mock.patch('docker.api.client.APIClient.post', post):
result = self.client.create_network('foo')
- self.assertEqual(result, network_data)
+ assert result == network_data
- self.assertEqual(
- post.call_args[0][0],
- url_prefix + 'networks/create')
+ assert post.call_args[0][0] == url_prefix + 'networks/create'
- self.assertEqual(
- json.loads(post.call_args[1]['data']),
- {"Name": "foo"})
+ assert json.loads(post.call_args[1]['data']) == {"Name": "foo"}
opts = {
'com.docker.network.bridge.enable_icc': False,
@@ -77,9 +73,9 @@ class NetworkTest(BaseAPIClientTest):
}
self.client.create_network('foo', 'bridge', opts)
- self.assertEqual(
- json.loads(post.call_args[1]['data']),
- {"Name": "foo", "Driver": "bridge", "Options": opts})
+ assert json.loads(post.call_args[1]['data']) == {
+ "Name": "foo", "Driver": "bridge", "Options": opts
+ }
ipam_pool_config = IPAMPool(subnet="192.168.52.0/24",
gateway="192.168.52.254")
@@ -88,21 +84,19 @@ class NetworkTest(BaseAPIClientTest):
self.client.create_network("bar", driver="bridge",
ipam=ipam_config)
- self.assertEqual(
- json.loads(post.call_args[1]['data']),
- {
- "Name": "bar",
- "Driver": "bridge",
- "IPAM": {
- "Driver": "default",
- "Config": [{
- "IPRange": None,
- "Gateway": "192.168.52.254",
- "Subnet": "192.168.52.0/24",
- "AuxiliaryAddresses": None,
- }],
- }
- })
+ assert json.loads(post.call_args[1]['data']) == {
+ "Name": "bar",
+ "Driver": "bridge",
+ "IPAM": {
+ "Driver": "default",
+ "Config": [{
+ "IPRange": None,
+ "Gateway": "192.168.52.254",
+ "Subnet": "192.168.52.0/24",
+ "AuxiliaryAddresses": None,
+ }],
+ }
+ }
@requires_api_version('1.21')
def test_remove_network(self):
@@ -113,8 +107,7 @@ class NetworkTest(BaseAPIClientTest):
self.client.remove_network(network_id)
args = delete.call_args
- self.assertEqual(args[0][0],
- url_prefix + 'networks/{0}'.format(network_id))
+ assert args[0][0] == url_prefix + 'networks/{0}'.format(network_id)
@requires_api_version('1.21')
def test_inspect_network(self):
@@ -132,11 +125,10 @@ class NetworkTest(BaseAPIClientTest):
with mock.patch('docker.api.client.APIClient.get', get):
result = self.client.inspect_network(network_id)
- self.assertEqual(result, network_data)
+ assert result == network_data
args = get.call_args
- self.assertEqual(args[0][0],
- url_prefix + 'networks/{0}'.format(network_id))
+ assert args[0][0] == url_prefix + 'networks/{0}'.format(network_id)
@requires_api_version('1.21')
def test_connect_container_to_network(self):
@@ -153,19 +145,17 @@ class NetworkTest(BaseAPIClientTest):
links=[('baz', 'quux')]
)
- self.assertEqual(
- post.call_args[0][0],
- url_prefix + 'networks/{0}/connect'.format(network_id))
+ assert post.call_args[0][0] == (
+ url_prefix + 'networks/{0}/connect'.format(network_id)
+ )
- self.assertEqual(
- json.loads(post.call_args[1]['data']),
- {
- 'Container': container_id,
- 'EndpointConfig': {
- 'Aliases': ['foo', 'bar'],
- 'Links': ['baz:quux'],
- },
- })
+ assert json.loads(post.call_args[1]['data']) == {
+ 'Container': container_id,
+ 'EndpointConfig': {
+ 'Aliases': ['foo', 'bar'],
+ 'Links': ['baz:quux'],
+ },
+ }
@requires_api_version('1.21')
def test_disconnect_container_from_network(self):
@@ -178,10 +168,9 @@ class NetworkTest(BaseAPIClientTest):
self.client.disconnect_container_from_network(
container={'Id': container_id}, net_id=network_id)
- self.assertEqual(
- post.call_args[0][0],
- url_prefix + 'networks/{0}/disconnect'.format(network_id))
-
- self.assertEqual(
- json.loads(post.call_args[1]['data']),
- {'Container': container_id})
+ assert post.call_args[0][0] == (
+ url_prefix + 'networks/{0}/disconnect'.format(network_id)
+ )
+ assert json.loads(post.call_args[1]['data']) == {
+ 'Container': container_id
+ }
diff --git a/tests/unit/api_test.py b/tests/unit/api_test.py
index 6ac92c4..b9e0d52 100644
--- a/tests/unit/api_test.py
+++ b/tests/unit/api_test.py
@@ -128,34 +128,27 @@ class DockerApiTest(BaseAPIClientTest):
with pytest.raises(docker.errors.DockerException) as excinfo:
APIClient(version=1.12)
- self.assertEqual(
- str(excinfo.value),
- 'Version parameter must be a string or None. Found float'
- )
+ assert str(
+ excinfo.value
+ ) == 'Version parameter must be a string or None. Found float'
def test_url_valid_resource(self):
url = self.client._url('/hello/{0}/world', 'somename')
- self.assertEqual(
- url, '{0}{1}'.format(url_prefix, 'hello/somename/world')
- )
+ assert url == '{0}{1}'.format(url_prefix, 'hello/somename/world')
url = self.client._url(
'/hello/{0}/world/{1}', 'somename', 'someothername'
)
- self.assertEqual(
- url,
- '{0}{1}'.format(url_prefix, 'hello/somename/world/someothername')
+ assert url == '{0}{1}'.format(
+ url_prefix, 'hello/somename/world/someothername'
)
url = self.client._url('/hello/{0}/world', 'some?name')
- self.assertEqual(
- url, '{0}{1}'.format(url_prefix, 'hello/some%3Fname/world')
- )
+ assert url == '{0}{1}'.format(url_prefix, 'hello/some%3Fname/world')
url = self.client._url("/images/{0}/push", "localhost:5000/image")
- self.assertEqual(
- url,
- '{0}{1}'.format(url_prefix, 'images/localhost:5000/image/push')
+ assert url == '{0}{1}'.format(
+ url_prefix, 'images/localhost:5000/image/push'
)
def test_url_invalid_resource(self):
@@ -164,15 +157,13 @@ class DockerApiTest(BaseAPIClientTest):
def test_url_no_resource(self):
url = self.client._url('/simple')
- self.assertEqual(url, '{0}{1}'.format(url_prefix, 'simple'))
+ assert url == '{0}{1}'.format(url_prefix, 'simple')
def test_url_unversioned_api(self):
url = self.client._url(
'/hello/{0}/world', 'somename', versioned_api=False
)
- self.assertEqual(
- url, '{0}{1}'.format(url_base, 'hello/somename/world')
- )
+ assert url == '{0}{1}'.format(url_base, 'hello/somename/world')
def test_version(self):
self.client.version()
@@ -194,13 +185,13 @@ class DockerApiTest(BaseAPIClientTest):
def test_retrieve_server_version(self):
client = APIClient(version="auto")
- self.assertTrue(isinstance(client._version, six.string_types))
- self.assertFalse(client._version == "auto")
+ assert isinstance(client._version, six.string_types)
+ assert not (client._version == "auto")
client.close()
def test_auto_retrieve_server_version(self):
version = self.client._retrieve_server_version()
- self.assertTrue(isinstance(version, six.string_types))
+ assert isinstance(version, six.string_types)
def test_info(self):
self.client.info()
@@ -313,11 +304,10 @@ class DockerApiTest(BaseAPIClientTest):
def test_create_host_config_secopt(self):
security_opt = ['apparmor:test_profile']
result = self.client.create_host_config(security_opt=security_opt)
- self.assertIn('SecurityOpt', result)
- self.assertEqual(result['SecurityOpt'], security_opt)
- self.assertRaises(
- TypeError, self.client.create_host_config, security_opt='wrong'
- )
+ assert 'SecurityOpt' in result
+ assert result['SecurityOpt'] == security_opt
+ with pytest.raises(TypeError):
+ self.client.create_host_config(security_opt='wrong')
def test_stream_helper_decoding(self):
status_code, content = fake_api.fake_responses[url_prefix + 'events']()
@@ -335,26 +325,26 @@ class DockerApiTest(BaseAPIClientTest):
raw_resp._fp.seek(0)
resp = response(status_code=status_code, content=content, raw=raw_resp)
result = next(self.client._stream_helper(resp))
- self.assertEqual(result, content_str)
+ assert result == content_str
# pass `decode=True` to the helper
raw_resp._fp.seek(0)
resp = response(status_code=status_code, content=content, raw=raw_resp)
result = next(self.client._stream_helper(resp, decode=True))
- self.assertEqual(result, content)
+ assert result == content
# non-chunked response, pass `decode=False` to the helper
setattr(raw_resp._fp, 'chunked', False)
raw_resp._fp.seek(0)
resp = response(status_code=status_code, content=content, raw=raw_resp)
result = next(self.client._stream_helper(resp))
- self.assertEqual(result, content_str.decode('utf-8'))
+ assert result == content_str.decode('utf-8')
# non-chunked response, pass `decode=True` to the helper
raw_resp._fp.seek(0)
resp = response(status_code=status_code, content=content, raw=raw_resp)
result = next(self.client._stream_helper(resp, decode=True))
- self.assertEqual(result, content)
+ assert result == content
class StreamTest(unittest.TestCase):
@@ -442,8 +432,7 @@ class StreamTest(unittest.TestCase):
b'\r\n'
) + b'\r\n'.join(lines)
- with APIClient(base_url="http+unix://" + self.socket_file) \
- as client:
+ with APIClient(base_url="http+unix://" + self.socket_file) as client:
for i in range(5):
try:
stream = client.build(
@@ -455,8 +444,8 @@ class StreamTest(unittest.TestCase):
if i == 4:
raise e
- self.assertEqual(list(stream), [
- str(i).encode() for i in range(50)])
+ assert list(stream) == [
+ str(i).encode() for i in range(50)]
class UserAgentTest(unittest.TestCase):
@@ -475,18 +464,18 @@ class UserAgentTest(unittest.TestCase):
client = APIClient()
client.version()
- self.assertEqual(self.mock_send.call_count, 1)
+ assert self.mock_send.call_count == 1
headers = self.mock_send.call_args[0][0].headers
expected = 'docker-sdk-python/%s' % docker.__version__
- self.assertEqual(headers['User-Agent'], expected)
+ assert headers['User-Agent'] == expected
def test_custom_user_agent(self):
client = APIClient(user_agent='foo/bar')
client.version()
- self.assertEqual(self.mock_send.call_count, 1)
+ assert self.mock_send.call_count == 1
headers = self.mock_send.call_args[0][0].headers
- self.assertEqual(headers['User-Agent'], 'foo/bar')
+ assert headers['User-Agent'] == 'foo/bar'
class DisableSocketTest(unittest.TestCase):
@@ -509,7 +498,7 @@ class DisableSocketTest(unittest.TestCase):
self.client._disable_socket_timeout(socket)
- self.assertEqual(socket.timeout, None)
+ assert socket.timeout is None
def test_disable_socket_timeout2(self):
"""Test that the timeouts are disabled on a generic socket object
@@ -519,8 +508,8 @@ class DisableSocketTest(unittest.TestCase):
self.client._disable_socket_timeout(socket)
- self.assertEqual(socket.timeout, None)
- self.assertEqual(socket._sock.timeout, None)
+ assert socket.timeout is None
+ assert socket._sock.timeout is None
def test_disable_socket_timout_non_blocking(self):
"""Test that a non-blocking socket does not get set to blocking."""
@@ -529,5 +518,5 @@ class DisableSocketTest(unittest.TestCase):
self.client._disable_socket_timeout(socket)
- self.assertEqual(socket.timeout, None)
- self.assertEqual(socket._sock.timeout, 0.0)
+ assert socket.timeout is None
+ assert socket._sock.timeout == 0.0
diff --git a/tests/unit/api_volume_test.py b/tests/unit/api_volume_test.py
index fc2a556..f5e5001 100644
--- a/tests/unit/api_volume_test.py
+++ b/tests/unit/api_volume_test.py
@@ -10,12 +10,12 @@ class VolumeTest(BaseAPIClientTest):
@requires_api_version('1.21')
def test_list_volumes(self):
volumes = self.client.volumes()
- self.assertIn('Volumes', volumes)
- self.assertEqual(len(volumes['Volumes']), 2)
+ assert 'Volumes' in volumes
+ assert len(volumes['Volumes']) == 2
args = fake_request.call_args
- self.assertEqual(args[0][0], 'GET')
- self.assertEqual(args[0][1], url_prefix + 'volumes')
+ assert args[0][0] == 'GET'
+ assert args[0][1] == url_prefix + 'volumes'
@requires_api_version('1.21')
def test_list_volumes_and_filters(self):
@@ -33,25 +33,25 @@ class VolumeTest(BaseAPIClientTest):
def test_create_volume(self):
name = 'perfectcherryblossom'
result = self.client.create_volume(name)
- self.assertIn('Name', result)
- self.assertEqual(result['Name'], name)
- self.assertIn('Driver', result)
- self.assertEqual(result['Driver'], 'local')
+ assert 'Name' in result
+ assert result['Name'] == name
+ assert 'Driver' in result
+ assert result['Driver'] == 'local'
args = fake_request.call_args
- self.assertEqual(args[0][0], 'POST')
- self.assertEqual(args[0][1], url_prefix + 'volumes/create')
- self.assertEqual(json.loads(args[1]['data']), {'Name': name})
+ assert args[0][0] == 'POST'
+ assert args[0][1] == url_prefix + 'volumes/create'
+ assert json.loads(args[1]['data']) == {'Name': name}
@requires_api_version('1.23')
def test_create_volume_with_labels(self):
name = 'perfectcherryblossom'
result = self.client.create_volume(name, labels={
- 'com.example.some-label': 'some-value'})
- self.assertEqual(
- result["Labels"],
- {'com.example.some-label': 'some-value'}
- )
+ 'com.example.some-label': 'some-value'
+ })
+ assert result["Labels"] == {
+ 'com.example.some-label': 'some-value'
+ }
@requires_api_version('1.23')
def test_create_volume_with_invalid_labels(self):
@@ -66,11 +66,11 @@ class VolumeTest(BaseAPIClientTest):
self.client.create_volume(name, driver=driver_name)
args = fake_request.call_args
- self.assertEqual(args[0][0], 'POST')
- self.assertEqual(args[0][1], url_prefix + 'volumes/create')
+ assert args[0][0] == 'POST'
+ assert args[0][1] == url_prefix + 'volumes/create'
data = json.loads(args[1]['data'])
- self.assertIn('Driver', data)
- self.assertEqual(data['Driver'], driver_name)
+ assert 'Driver' in data
+ assert data['Driver'] == driver_name
@requires_api_version('1.21')
def test_create_volume_invalid_opts_type(self):
@@ -92,25 +92,25 @@ class VolumeTest(BaseAPIClientTest):
@requires_api_version('1.24')
def test_create_volume_with_no_specified_name(self):
result = self.client.create_volume(name=None)
- self.assertIn('Name', result)
- self.assertNotEqual(result['Name'], None)
- self.assertIn('Driver', result)
- self.assertEqual(result['Driver'], 'local')
- self.assertIn('Scope', result)
- self.assertEqual(result['Scope'], 'local')
+ assert 'Name' in result
+ assert result['Name'] is not None
+ assert 'Driver' in result
+ assert result['Driver'] == 'local'
+ assert 'Scope' in result
+ assert result['Scope'] == 'local'
@requires_api_version('1.21')
def test_inspect_volume(self):
name = 'perfectcherryblossom'
result = self.client.inspect_volume(name)
- self.assertIn('Name', result)
- self.assertEqual(result['Name'], name)
- self.assertIn('Driver', result)
- self.assertEqual(result['Driver'], 'local')
+ assert 'Name' in result
+ assert result['Name'] == name
+ assert 'Driver' in result
+ assert result['Driver'] == 'local'
args = fake_request.call_args
- self.assertEqual(args[0][0], 'GET')
- self.assertEqual(args[0][1], '{0}volumes/{1}'.format(url_prefix, name))
+ assert args[0][0] == 'GET'
+ assert args[0][1] == '{0}volumes/{1}'.format(url_prefix, name)
@requires_api_version('1.21')
def test_remove_volume(self):
@@ -118,5 +118,5 @@ class VolumeTest(BaseAPIClientTest):
self.client.remove_volume(name)
args = fake_request.call_args
- self.assertEqual(args[0][0], 'DELETE')
- self.assertEqual(args[0][1], '{0}volumes/{1}'.format(url_prefix, name))
+ assert args[0][0] == 'DELETE'
+ assert args[0][1] == '{0}volumes/{1}'.format(url_prefix, name)
diff --git a/tests/unit/auth_test.py b/tests/unit/auth_test.py
index 56fd50c..1506ccb 100644
--- a/tests/unit/auth_test.py
+++ b/tests/unit/auth_test.py
@@ -13,6 +13,7 @@ from py.test import ensuretemp
from pytest import mark
from docker import auth, errors
+import pytest
try:
from unittest import mock
@@ -33,82 +34,68 @@ class RegressionTest(unittest.TestCase):
class ResolveRepositoryNameTest(unittest.TestCase):
def test_resolve_repository_name_hub_library_image(self):
- self.assertEqual(
- auth.resolve_repository_name('image'),
- ('docker.io', 'image'),
+ assert auth.resolve_repository_name('image') == (
+ 'docker.io', 'image'
)
def test_resolve_repository_name_dotted_hub_library_image(self):
- self.assertEqual(
- auth.resolve_repository_name('image.valid'),
- ('docker.io', 'image.valid')
+ assert auth.resolve_repository_name('image.valid') == (
+ 'docker.io', 'image.valid'
)
def test_resolve_repository_name_hub_image(self):
- self.assertEqual(
- auth.resolve_repository_name('username/image'),
- ('docker.io', 'username/image'),
+ assert auth.resolve_repository_name('username/image') == (
+ 'docker.io', 'username/image'
)
def test_explicit_hub_index_library_image(self):
- self.assertEqual(
- auth.resolve_repository_name('docker.io/image'),
- ('docker.io', 'image')
+ assert auth.resolve_repository_name('docker.io/image') == (
+ 'docker.io', 'image'
)
def test_explicit_legacy_hub_index_library_image(self):
- self.assertEqual(
- auth.resolve_repository_name('index.docker.io/image'),
- ('docker.io', 'image')
+ assert auth.resolve_repository_name('index.docker.io/image') == (
+ 'docker.io', 'image'
)
def test_resolve_repository_name_private_registry(self):
- self.assertEqual(
- auth.resolve_repository_name('my.registry.net/image'),
- ('my.registry.net', 'image'),
+ assert auth.resolve_repository_name('my.registry.net/image') == (
+ 'my.registry.net', 'image'
)
def test_resolve_repository_name_private_registry_with_port(self):
- self.assertEqual(
- auth.resolve_repository_name('my.registry.net:5000/image'),
- ('my.registry.net:5000', 'image'),
+ assert auth.resolve_repository_name('my.registry.net:5000/image') == (
+ 'my.registry.net:5000', 'image'
)
def test_resolve_repository_name_private_registry_with_username(self):
- self.assertEqual(
- auth.resolve_repository_name('my.registry.net/username/image'),
- ('my.registry.net', 'username/image'),
- )
+ assert auth.resolve_repository_name(
+ 'my.registry.net/username/image'
+ ) == ('my.registry.net', 'username/image')
def test_resolve_repository_name_no_dots_but_port(self):
- self.assertEqual(
- auth.resolve_repository_name('hostname:5000/image'),
- ('hostname:5000', 'image'),
+ assert auth.resolve_repository_name('hostname:5000/image') == (
+ 'hostname:5000', 'image'
)
def test_resolve_repository_name_no_dots_but_port_and_username(self):
- self.assertEqual(
- auth.resolve_repository_name('hostname:5000/username/image'),
- ('hostname:5000', 'username/image'),
- )
+ assert auth.resolve_repository_name(
+ 'hostname:5000/username/image'
+ ) == ('hostname:5000', 'username/image')
def test_resolve_repository_name_localhost(self):
- self.assertEqual(
- auth.resolve_repository_name('localhost/image'),
- ('localhost', 'image'),
+ assert auth.resolve_repository_name('localhost/image') == (
+ 'localhost', 'image'
)
def test_resolve_repository_name_localhost_with_username(self):
- self.assertEqual(
- auth.resolve_repository_name('localhost/username/image'),
- ('localhost', 'username/image'),
+ assert auth.resolve_repository_name('localhost/username/image') == (
+ 'localhost', 'username/image'
)
def test_invalid_index_name(self):
- self.assertRaises(
- errors.InvalidRepository,
- lambda: auth.resolve_repository_name('-gecko.com/image')
- )
+ with pytest.raises(errors.InvalidRepository):
+ auth.resolve_repository_name('-gecko.com/image')
def encode_auth(auth_info):
@@ -129,147 +116,100 @@ class ResolveAuthTest(unittest.TestCase):
})
def test_resolve_authconfig_hostname_only(self):
- self.assertEqual(
- auth.resolve_authconfig(
- self.auth_config, 'my.registry.net'
- )['username'],
- 'privateuser'
- )
+ assert auth.resolve_authconfig(
+ self.auth_config, 'my.registry.net'
+ )['username'] == 'privateuser'
def test_resolve_authconfig_no_protocol(self):
- self.assertEqual(
- auth.resolve_authconfig(
- self.auth_config, 'my.registry.net/v1/'
- )['username'],
- 'privateuser'
- )
+ assert auth.resolve_authconfig(
+ self.auth_config, 'my.registry.net/v1/'
+ )['username'] == 'privateuser'
def test_resolve_authconfig_no_path(self):
- self.assertEqual(
- auth.resolve_authconfig(
- self.auth_config, 'http://my.registry.net'
- )['username'],
- 'privateuser'
- )
+ assert auth.resolve_authconfig(
+ self.auth_config, 'http://my.registry.net'
+ )['username'] == 'privateuser'
def test_resolve_authconfig_no_path_trailing_slash(self):
- self.assertEqual(
- auth.resolve_authconfig(
- self.auth_config, 'http://my.registry.net/'
- )['username'],
- 'privateuser'
- )
+ assert auth.resolve_authconfig(
+ self.auth_config, 'http://my.registry.net/'
+ )['username'] == 'privateuser'
def test_resolve_authconfig_no_path_wrong_secure_proto(self):
- self.assertEqual(
- auth.resolve_authconfig(
- self.auth_config, 'https://my.registry.net'
- )['username'],
- 'privateuser'
- )
+ assert auth.resolve_authconfig(
+ self.auth_config, 'https://my.registry.net'
+ )['username'] == 'privateuser'
def test_resolve_authconfig_no_path_wrong_insecure_proto(self):
- self.assertEqual(
- auth.resolve_authconfig(
- self.auth_config, 'http://index.docker.io'
- )['username'],
- 'indexuser'
- )
+ assert auth.resolve_authconfig(
+ self.auth_config, 'http://index.docker.io'
+ )['username'] == 'indexuser'
def test_resolve_authconfig_path_wrong_proto(self):
- self.assertEqual(
- auth.resolve_authconfig(
- self.auth_config, 'https://my.registry.net/v1/'
- )['username'],
- 'privateuser'
- )
+ assert auth.resolve_authconfig(
+ self.auth_config, 'https://my.registry.net/v1/'
+ )['username'] == 'privateuser'
def test_resolve_authconfig_default_registry(self):
- self.assertEqual(
- auth.resolve_authconfig(self.auth_config)['username'],
- 'indexuser'
- )
+ assert auth.resolve_authconfig(
+ self.auth_config
+ )['username'] == 'indexuser'
def test_resolve_authconfig_default_explicit_none(self):
- self.assertEqual(
- auth.resolve_authconfig(self.auth_config, None)['username'],
- 'indexuser'
- )
+ assert auth.resolve_authconfig(
+ self.auth_config, None
+ )['username'] == 'indexuser'
def test_resolve_authconfig_fully_explicit(self):
- self.assertEqual(
- auth.resolve_authconfig(
- self.auth_config, 'http://my.registry.net/v1/'
- )['username'],
- 'privateuser'
- )
+ assert auth.resolve_authconfig(
+ self.auth_config, 'http://my.registry.net/v1/'
+ )['username'] == 'privateuser'
def test_resolve_authconfig_legacy_config(self):
- self.assertEqual(
- auth.resolve_authconfig(
- self.auth_config, 'legacy.registry.url'
- )['username'],
- 'legacyauth'
- )
+ assert auth.resolve_authconfig(
+ self.auth_config, 'legacy.registry.url'
+ )['username'] == 'legacyauth'
def test_resolve_authconfig_no_match(self):
- self.assertTrue(
- auth.resolve_authconfig(self.auth_config, 'does.not.exist') is None
- )
+ assert auth.resolve_authconfig(
+ self.auth_config, 'does.not.exist'
+ ) is None
def test_resolve_registry_and_auth_library_image(self):
image = 'image'
- self.assertEqual(
- auth.resolve_authconfig(
- self.auth_config, auth.resolve_repository_name(image)[0]
- )['username'],
- 'indexuser',
- )
+ assert auth.resolve_authconfig(
+ self.auth_config, auth.resolve_repository_name(image)[0]
+ )['username'] == 'indexuser'
def test_resolve_registry_and_auth_hub_image(self):
image = 'username/image'
- self.assertEqual(
- auth.resolve_authconfig(
- self.auth_config, auth.resolve_repository_name(image)[0]
- )['username'],
- 'indexuser',
- )
+ assert auth.resolve_authconfig(
+ self.auth_config, auth.resolve_repository_name(image)[0]
+ )['username'] == 'indexuser'
def test_resolve_registry_and_auth_explicit_hub(self):
image = 'docker.io/username/image'
- self.assertEqual(
- auth.resolve_authconfig(
- self.auth_config, auth.resolve_repository_name(image)[0]
- )['username'],
- 'indexuser',
- )
+ assert auth.resolve_authconfig(
+ self.auth_config, auth.resolve_repository_name(image)[0]
+ )['username'] == 'indexuser'
def test_resolve_registry_and_auth_explicit_legacy_hub(self):
image = 'index.docker.io/username/image'
- self.assertEqual(
- auth.resolve_authconfig(
- self.auth_config, auth.resolve_repository_name(image)[0]
- )['username'],
- 'indexuser',
- )
+ assert auth.resolve_authconfig(
+ self.auth_config, auth.resolve_repository_name(image)[0]
+ )['username'] == 'indexuser'
def test_resolve_registry_and_auth_private_registry(self):
image = 'my.registry.net/image'
- self.assertEqual(
- auth.resolve_authconfig(
- self.auth_config, auth.resolve_repository_name(image)[0]
- )['username'],
- 'privateuser',
- )
+ assert auth.resolve_authconfig(
+ self.auth_config, auth.resolve_repository_name(image)[0]
+ )['username'] == 'privateuser'
def test_resolve_registry_and_auth_unauthenticated_registry(self):
image = 'other.registry.net/image'
- self.assertEqual(
- auth.resolve_authconfig(
- self.auth_config, auth.resolve_repository_name(image)[0]
- ),
- None,
- )
+ assert auth.resolve_authconfig(
+ self.auth_config, auth.resolve_repository_name(image)[0]
+ ) is None
class CredStoreTest(unittest.TestCase):
@@ -378,7 +318,7 @@ class LoadConfigTest(unittest.TestCase):
folder = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, folder)
cfg = auth.load_config(folder)
- self.assertTrue(cfg is not None)
+ assert cfg is not None
def test_load_config(self):
folder = tempfile.mkdtemp()
@@ -390,12 +330,12 @@ class LoadConfigTest(unittest.TestCase):
f.write('email = sakuya@scarlet.net')
cfg = auth.load_config(dockercfg_path)
assert auth.INDEX_NAME in cfg
- self.assertNotEqual(cfg[auth.INDEX_NAME], None)
+ assert cfg[auth.INDEX_NAME] is not None
cfg = cfg[auth.INDEX_NAME]
- self.assertEqual(cfg['username'], 'sakuya')
- self.assertEqual(cfg['password'], 'izayoi')
- self.assertEqual(cfg['email'], 'sakuya@scarlet.net')
- self.assertEqual(cfg.get('auth'), None)
+ assert cfg['username'] == 'sakuya'
+ assert cfg['password'] == 'izayoi'
+ assert cfg['email'] == 'sakuya@scarlet.net'
+ assert cfg.get('auth') is None
def test_load_config_with_random_name(self):
folder = tempfile.mkdtemp()
@@ -418,12 +358,12 @@ class LoadConfigTest(unittest.TestCase):
cfg = auth.load_config(dockercfg_path)
assert registry in cfg
- self.assertNotEqual(cfg[registry], None)
+ assert cfg[registry] is not None
cfg = cfg[registry]
- self.assertEqual(cfg['username'], 'sakuya')
- self.assertEqual(cfg['password'], 'izayoi')
- self.assertEqual(cfg['email'], 'sakuya@scarlet.net')
- self.assertEqual(cfg.get('auth'), None)
+ assert cfg['username'] == 'sakuya'
+ assert cfg['password'] == 'izayoi'
+ assert cfg['email'] == 'sakuya@scarlet.net'
+ assert cfg.get('auth') is None
def test_load_config_custom_config_env(self):
folder = tempfile.mkdtemp()
@@ -445,12 +385,12 @@ class LoadConfigTest(unittest.TestCase):
with mock.patch.dict(os.environ, {'DOCKER_CONFIG': folder}):
cfg = auth.load_config(None)
assert registry in cfg
- self.assertNotEqual(cfg[registry], None)
+ assert cfg[registry] is not None
cfg = cfg[registry]
- self.assertEqual(cfg['username'], 'sakuya')
- self.assertEqual(cfg['password'], 'izayoi')
- self.assertEqual(cfg['email'], 'sakuya@scarlet.net')
- self.assertEqual(cfg.get('auth'), None)
+ assert cfg['username'] == 'sakuya'
+ assert cfg['password'] == 'izayoi'
+ assert cfg['email'] == 'sakuya@scarlet.net'
+ assert cfg.get('auth') is None
def test_load_config_custom_config_env_with_auths(self):
folder = tempfile.mkdtemp()
@@ -474,12 +414,12 @@ class LoadConfigTest(unittest.TestCase):
with mock.patch.dict(os.environ, {'DOCKER_CONFIG': folder}):
cfg = auth.load_config(None)
assert registry in cfg
- self.assertNotEqual(cfg[registry], None)
+ assert cfg[registry] is not None
cfg = cfg[registry]
- self.assertEqual(cfg['username'], 'sakuya')
- self.assertEqual(cfg['password'], 'izayoi')
- self.assertEqual(cfg['email'], 'sakuya@scarlet.net')
- self.assertEqual(cfg.get('auth'), None)
+ assert cfg['username'] == 'sakuya'
+ assert cfg['password'] == 'izayoi'
+ assert cfg['email'] == 'sakuya@scarlet.net'
+ assert cfg.get('auth') is None
def test_load_config_custom_config_env_utf8(self):
folder = tempfile.mkdtemp()
@@ -504,12 +444,12 @@ class LoadConfigTest(unittest.TestCase):
with mock.patch.dict(os.environ, {'DOCKER_CONFIG': folder}):
cfg = auth.load_config(None)
assert registry in cfg
- self.assertNotEqual(cfg[registry], None)
+ assert cfg[registry] is not None
cfg = cfg[registry]
- self.assertEqual(cfg['username'], b'sakuya\xc3\xa6'.decode('utf8'))
- self.assertEqual(cfg['password'], b'izayoi\xc3\xa6'.decode('utf8'))
- self.assertEqual(cfg['email'], 'sakuya@scarlet.net')
- self.assertEqual(cfg.get('auth'), None)
+ assert cfg['username'] == b'sakuya\xc3\xa6'.decode('utf8')
+ assert cfg['password'] == b'izayoi\xc3\xa6'.decode('utf8')
+ assert cfg['email'] == 'sakuya@scarlet.net'
+ assert cfg.get('auth') is None
def test_load_config_custom_config_env_with_headers(self):
folder = tempfile.mkdtemp()
@@ -529,11 +469,11 @@ class LoadConfigTest(unittest.TestCase):
with mock.patch.dict(os.environ, {'DOCKER_CONFIG': folder}):
cfg = auth.load_config(None)
assert 'HttpHeaders' in cfg
- self.assertNotEqual(cfg['HttpHeaders'], None)
+ assert cfg['HttpHeaders'] is not None
cfg = cfg['HttpHeaders']
- self.assertEqual(cfg['Name'], 'Spike')
- self.assertEqual(cfg['Surname'], 'Spiegel')
+ assert cfg['Name'] == 'Spike'
+ assert cfg['Surname'] == 'Spiegel'
def test_load_config_unknown_keys(self):
folder = tempfile.mkdtemp()
diff --git a/tests/unit/client_test.py b/tests/unit/client_test.py
index c4996f1..cce99c5 100644
--- a/tests/unit/client_test.py
+++ b/tests/unit/client_test.py
@@ -8,6 +8,7 @@ import os
import unittest
from . import fake_api
+import pytest
try:
from unittest import mock
@@ -51,25 +52,25 @@ class ClientTest(unittest.TestCase):
def test_call_api_client_method(self):
client = docker.from_env()
- with self.assertRaises(AttributeError) as cm:
+ with pytest.raises(AttributeError) as cm:
client.create_container()
- s = str(cm.exception)
+ s = cm.exconly()
assert "'DockerClient' object has no attribute 'create_container'" in s
assert "this method is now on the object APIClient" in s
- with self.assertRaises(AttributeError) as cm:
+ with pytest.raises(AttributeError) as cm:
client.abcdef()
- s = str(cm.exception)
+ s = cm.exconly()
assert "'DockerClient' object has no attribute 'abcdef'" in s
assert "this method is now on the object APIClient" not in s
def test_call_containers(self):
client = docker.DockerClient(**kwargs_from_env())
- with self.assertRaises(TypeError) as cm:
+ with pytest.raises(TypeError) as cm:
client.containers()
- s = str(cm.exception)
+ s = cm.exconly()
assert "'ContainerCollection' object is not callable" in s
assert "docker.APIClient" in s
@@ -90,22 +91,22 @@ class FromEnvTest(unittest.TestCase):
DOCKER_CERT_PATH=TEST_CERT_DIR,
DOCKER_TLS_VERIFY='1')
client = docker.from_env()
- self.assertEqual(client.api.base_url, "https://192.168.59.103:2376")
+ assert client.api.base_url == "https://192.168.59.103:2376"
def test_from_env_with_version(self):
os.environ.update(DOCKER_HOST='tcp://192.168.59.103:2376',
DOCKER_CERT_PATH=TEST_CERT_DIR,
DOCKER_TLS_VERIFY='1')
client = docker.from_env(version='2.32')
- self.assertEqual(client.api.base_url, "https://192.168.59.103:2376")
- self.assertEqual(client.api._version, '2.32')
+ assert client.api.base_url == "https://192.168.59.103:2376"
+ assert client.api._version == '2.32'
def test_from_env_without_version_uses_default(self):
client = docker.from_env()
- self.assertEqual(client.api._version, DEFAULT_DOCKER_API_VERSION)
+ assert client.api._version == DEFAULT_DOCKER_API_VERSION
def test_from_env_without_timeout_uses_default(self):
client = docker.from_env()
- self.assertEqual(client.api.timeout, DEFAULT_TIMEOUT_SECONDS)
+ assert client.api.timeout == DEFAULT_TIMEOUT_SECONDS
diff --git a/tests/unit/dockertypes_test.py b/tests/unit/dockertypes_test.py
index 40adbb7..93c1397 100644
--- a/tests/unit/dockertypes_test.py
+++ b/tests/unit/dockertypes_test.py
@@ -25,11 +25,11 @@ def create_host_config(*args, **kwargs):
class HostConfigTest(unittest.TestCase):
def test_create_host_config_no_options(self):
config = create_host_config(version='1.19')
- self.assertFalse('NetworkMode' in config)
+ assert not ('NetworkMode' in config)
def test_create_host_config_no_options_newer_api_version(self):
config = create_host_config(version='1.20')
- self.assertEqual(config['NetworkMode'], 'default')
+ assert config['NetworkMode'] == 'default'
def test_create_host_config_invalid_cpu_cfs_types(self):
with pytest.raises(TypeError):
@@ -46,65 +46,58 @@ class HostConfigTest(unittest.TestCase):
def test_create_host_config_with_cpu_quota(self):
config = create_host_config(version='1.20', cpu_quota=1999)
- self.assertEqual(config.get('CpuQuota'), 1999)
+ assert config.get('CpuQuota') == 1999
def test_create_host_config_with_cpu_period(self):
config = create_host_config(version='1.20', cpu_period=1999)
- self.assertEqual(config.get('CpuPeriod'), 1999)
+ assert config.get('CpuPeriod') == 1999
def test_create_host_config_with_blkio_constraints(self):
blkio_rate = [{"Path": "/dev/sda", "Rate": 1000}]
- config = create_host_config(version='1.22',
- blkio_weight=1999,
- blkio_weight_device=blkio_rate,
- device_read_bps=blkio_rate,
- device_write_bps=blkio_rate,
- device_read_iops=blkio_rate,
- device_write_iops=blkio_rate)
-
- self.assertEqual(config.get('BlkioWeight'), 1999)
- self.assertTrue(config.get('BlkioWeightDevice') is blkio_rate)
- self.assertTrue(config.get('BlkioDeviceReadBps') is blkio_rate)
- self.assertTrue(config.get('BlkioDeviceWriteBps') is blkio_rate)
- self.assertTrue(config.get('BlkioDeviceReadIOps') is blkio_rate)
- self.assertTrue(config.get('BlkioDeviceWriteIOps') is blkio_rate)
- self.assertEqual(blkio_rate[0]['Path'], "/dev/sda")
- self.assertEqual(blkio_rate[0]['Rate'], 1000)
+ config = create_host_config(
+ version='1.22', blkio_weight=1999, blkio_weight_device=blkio_rate,
+ device_read_bps=blkio_rate, device_write_bps=blkio_rate,
+ device_read_iops=blkio_rate, device_write_iops=blkio_rate
+ )
+
+ assert config.get('BlkioWeight') == 1999
+ assert config.get('BlkioWeightDevice') is blkio_rate
+ assert config.get('BlkioDeviceReadBps') is blkio_rate
+ assert config.get('BlkioDeviceWriteBps') is blkio_rate
+ assert config.get('BlkioDeviceReadIOps') is blkio_rate
+ assert config.get('BlkioDeviceWriteIOps') is blkio_rate
+ assert blkio_rate[0]['Path'] == "/dev/sda"
+ assert blkio_rate[0]['Rate'] == 1000
def test_create_host_config_with_shm_size(self):
config = create_host_config(version='1.22', shm_size=67108864)
- self.assertEqual(config.get('ShmSize'), 67108864)
+ assert config.get('ShmSize') == 67108864
def test_create_host_config_with_shm_size_in_mb(self):
config = create_host_config(version='1.22', shm_size='64M')
- self.assertEqual(config.get('ShmSize'), 67108864)
+ assert config.get('ShmSize') == 67108864
def test_create_host_config_with_oom_kill_disable(self):
config = create_host_config(version='1.20', oom_kill_disable=True)
- self.assertEqual(config.get('OomKillDisable'), True)
- self.assertRaises(
- InvalidVersion, lambda: create_host_config(version='1.18.3',
- oom_kill_disable=True))
+ assert config.get('OomKillDisable') is True
+ with pytest.raises(InvalidVersion):
+ create_host_config(version='1.18.3', oom_kill_disable=True)
def test_create_host_config_with_userns_mode(self):
config = create_host_config(version='1.23', userns_mode='host')
- self.assertEqual(config.get('UsernsMode'), 'host')
- self.assertRaises(
- InvalidVersion, lambda: create_host_config(version='1.22',
- userns_mode='host'))
- self.assertRaises(
- ValueError, lambda: create_host_config(version='1.23',
- userns_mode='host12'))
+ assert config.get('UsernsMode') == 'host'
+ with pytest.raises(InvalidVersion):
+ create_host_config(version='1.22', userns_mode='host')
+ with pytest.raises(ValueError):
+ create_host_config(version='1.23', userns_mode='host12')
def test_create_host_config_with_oom_score_adj(self):
config = create_host_config(version='1.22', oom_score_adj=100)
- self.assertEqual(config.get('OomScoreAdj'), 100)
- self.assertRaises(
- InvalidVersion, lambda: create_host_config(version='1.21',
- oom_score_adj=100))
- self.assertRaises(
- TypeError, lambda: create_host_config(version='1.22',
- oom_score_adj='100'))
+ assert config.get('OomScoreAdj') == 100
+ with pytest.raises(InvalidVersion):
+ create_host_config(version='1.21', oom_score_adj=100)
+ with pytest.raises(TypeError):
+ create_host_config(version='1.22', oom_score_adj='100')
def test_create_host_config_with_dns_opt(self):
@@ -112,30 +105,27 @@ class HostConfigTest(unittest.TestCase):
config = create_host_config(version='1.21', dns_opt=tested_opts)
dns_opts = config.get('DnsOptions')
- self.assertTrue('use-vc' in dns_opts)
- self.assertTrue('no-tld-query' in dns_opts)
+ assert 'use-vc' in dns_opts
+ assert 'no-tld-query' in dns_opts
- self.assertRaises(
- InvalidVersion, lambda: create_host_config(version='1.20',
- dns_opt=tested_opts))
+ with pytest.raises(InvalidVersion):
+ create_host_config(version='1.20', dns_opt=tested_opts)
def test_create_host_config_with_mem_reservation(self):
config = create_host_config(version='1.21', mem_reservation=67108864)
- self.assertEqual(config.get('MemoryReservation'), 67108864)
- self.assertRaises(
- InvalidVersion, lambda: create_host_config(
- version='1.20', mem_reservation=67108864))
+ assert config.get('MemoryReservation') == 67108864
+ with pytest.raises(InvalidVersion):
+ create_host_config(version='1.20', mem_reservation=67108864)
def test_create_host_config_with_kernel_memory(self):
config = create_host_config(version='1.21', kernel_memory=67108864)
- self.assertEqual(config.get('KernelMemory'), 67108864)
- self.assertRaises(
- InvalidVersion, lambda: create_host_config(
- version='1.20', kernel_memory=67108864))
+ assert config.get('KernelMemory') == 67108864
+ with pytest.raises(InvalidVersion):
+ create_host_config(version='1.20', kernel_memory=67108864)
def test_create_host_config_with_pids_limit(self):
config = create_host_config(version='1.23', pids_limit=1024)
- self.assertEqual(config.get('PidsLimit'), 1024)
+ assert config.get('PidsLimit') == 1024
with pytest.raises(InvalidVersion):
create_host_config(version='1.22', pids_limit=1024)
@@ -144,7 +134,7 @@ class HostConfigTest(unittest.TestCase):
def test_create_host_config_with_isolation(self):
config = create_host_config(version='1.24', isolation='hyperv')
- self.assertEqual(config.get('Isolation'), 'hyperv')
+ assert config.get('Isolation') == 'hyperv'
with pytest.raises(InvalidVersion):
create_host_config(version='1.23', isolation='hyperv')
@@ -179,10 +169,9 @@ class HostConfigTest(unittest.TestCase):
def test_create_host_config_with_cpu_count(self):
config = create_host_config(version='1.25', cpu_count=2)
- self.assertEqual(config.get('CpuCount'), 2)
- self.assertRaises(
- InvalidVersion, lambda: create_host_config(
- version='1.24', cpu_count=1))
+ assert config.get('CpuCount') == 2
+ with pytest.raises(InvalidVersion):
+ create_host_config(version='1.24', cpu_count=1)
def test_create_host_config_invalid_cpu_percent_types(self):
with pytest.raises(TypeError):
@@ -190,10 +179,9 @@ class HostConfigTest(unittest.TestCase):
def test_create_host_config_with_cpu_percent(self):
config = create_host_config(version='1.25', cpu_percent=15)
- self.assertEqual(config.get('CpuPercent'), 15)
- self.assertRaises(
- InvalidVersion, lambda: create_host_config(
- version='1.24', cpu_percent=10))
+ assert config.get('CpuPercent') == 15
+ with pytest.raises(InvalidVersion):
+ create_host_config(version='1.24', cpu_percent=10)
def test_create_host_config_invalid_nano_cpus_types(self):
with pytest.raises(TypeError):
@@ -201,10 +189,9 @@ class HostConfigTest(unittest.TestCase):
def test_create_host_config_with_nano_cpus(self):
config = create_host_config(version='1.25', nano_cpus=1000)
- self.assertEqual(config.get('NanoCpus'), 1000)
- self.assertRaises(
- InvalidVersion, lambda: create_host_config(
- version='1.24', nano_cpus=1))
+ assert config.get('NanoCpus') == 1000
+ with pytest.raises(InvalidVersion):
+ create_host_config(version='1.24', nano_cpus=1)
def test_create_host_config_with_cpu_rt_period_types(self):
with pytest.raises(TypeError):
@@ -212,10 +199,9 @@ class HostConfigTest(unittest.TestCase):
def test_create_host_config_with_cpu_rt_period(self):
config = create_host_config(version='1.25', cpu_rt_period=1000)
- self.assertEqual(config.get('CPURealtimePeriod'), 1000)
- self.assertRaises(
- InvalidVersion, lambda: create_host_config(
- version='1.24', cpu_rt_period=1000))
+ assert config.get('CPURealtimePeriod') == 1000
+ with pytest.raises(InvalidVersion):
+ create_host_config(version='1.24', cpu_rt_period=1000)
def test_ctrate_host_config_with_cpu_rt_runtime_types(self):
with pytest.raises(TypeError):
@@ -223,10 +209,9 @@ class HostConfigTest(unittest.TestCase):
def test_create_host_config_with_cpu_rt_runtime(self):
config = create_host_config(version='1.25', cpu_rt_runtime=1000)
- self.assertEqual(config.get('CPURealtimeRuntime'), 1000)
- self.assertRaises(
- InvalidVersion, lambda: create_host_config(
- version='1.24', cpu_rt_runtime=1000))
+ assert config.get('CPURealtimeRuntime') == 1000
+ with pytest.raises(InvalidVersion):
+ create_host_config(version='1.24', cpu_rt_runtime=1000)
class ContainerConfigTest(unittest.TestCase):
@@ -264,43 +249,46 @@ class UlimitTest(unittest.TestCase):
config = create_host_config(
ulimits=[ulimit_dct], version=DEFAULT_DOCKER_API_VERSION
)
- self.assertIn('Ulimits', config)
- self.assertEqual(len(config['Ulimits']), 1)
+ assert 'Ulimits' in config
+ assert len(config['Ulimits']) == 1
ulimit_obj = config['Ulimits'][0]
- self.assertTrue(isinstance(ulimit_obj, Ulimit))
- self.assertEqual(ulimit_obj.name, ulimit_dct['name'])
- self.assertEqual(ulimit_obj.soft, ulimit_dct['soft'])
- self.assertEqual(ulimit_obj['Soft'], ulimit_obj.soft)
+ assert isinstance(ulimit_obj, Ulimit)
+ assert ulimit_obj.name == ulimit_dct['name']
+ assert ulimit_obj.soft == ulimit_dct['soft']
+ assert ulimit_obj['Soft'] == ulimit_obj.soft
def test_create_host_config_dict_ulimit_capitals(self):
ulimit_dct = {'Name': 'nofile', 'Soft': 8096, 'Hard': 8096 * 4}
config = create_host_config(
ulimits=[ulimit_dct], version=DEFAULT_DOCKER_API_VERSION
)
- self.assertIn('Ulimits', config)
- self.assertEqual(len(config['Ulimits']), 1)
+ assert 'Ulimits' in config
+ assert len(config['Ulimits']) == 1
ulimit_obj = config['Ulimits'][0]
- self.assertTrue(isinstance(ulimit_obj, Ulimit))
- self.assertEqual(ulimit_obj.name, ulimit_dct['Name'])
- self.assertEqual(ulimit_obj.soft, ulimit_dct['Soft'])
- self.assertEqual(ulimit_obj.hard, ulimit_dct['Hard'])
- self.assertEqual(ulimit_obj['Soft'], ulimit_obj.soft)
+ assert isinstance(ulimit_obj, Ulimit)
+ assert ulimit_obj.name == ulimit_dct['Name']
+ assert ulimit_obj.soft == ulimit_dct['Soft']
+ assert ulimit_obj.hard == ulimit_dct['Hard']
+ assert ulimit_obj['Soft'] == ulimit_obj.soft
def test_create_host_config_obj_ulimit(self):
ulimit_dct = Ulimit(name='nofile', soft=8096)
config = create_host_config(
ulimits=[ulimit_dct], version=DEFAULT_DOCKER_API_VERSION
)
- self.assertIn('Ulimits', config)
- self.assertEqual(len(config['Ulimits']), 1)
+ assert 'Ulimits' in config
+ assert len(config['Ulimits']) == 1
ulimit_obj = config['Ulimits'][0]
- self.assertTrue(isinstance(ulimit_obj, Ulimit))
- self.assertEqual(ulimit_obj, ulimit_dct)
+ assert isinstance(ulimit_obj, Ulimit)
+ assert ulimit_obj == ulimit_dct
def test_ulimit_invalid_type(self):
- self.assertRaises(ValueError, lambda: Ulimit(name=None))
- self.assertRaises(ValueError, lambda: Ulimit(name='hello', soft='123'))
- self.assertRaises(ValueError, lambda: Ulimit(name='hello', hard='456'))
+ with pytest.raises(ValueError):
+ Ulimit(name=None)
+ with pytest.raises(ValueError):
+ Ulimit(name='hello', soft='123')
+ with pytest.raises(ValueError):
+ Ulimit(name='hello', hard='456')
class LogConfigTest(unittest.TestCase):
@@ -309,18 +297,18 @@ class LogConfigTest(unittest.TestCase):
config = create_host_config(
version=DEFAULT_DOCKER_API_VERSION, log_config=dct
)
- self.assertIn('LogConfig', config)
- self.assertTrue(isinstance(config['LogConfig'], LogConfig))
- self.assertEqual(dct['type'], config['LogConfig'].type)
+ assert 'LogConfig' in config
+ assert isinstance(config['LogConfig'], LogConfig)
+ assert dct['type'] == config['LogConfig'].type
def test_create_host_config_obj_logconfig(self):
obj = LogConfig(type=LogConfig.types.SYSLOG, config={'key1': 'val1'})
config = create_host_config(
version=DEFAULT_DOCKER_API_VERSION, log_config=obj
)
- self.assertIn('LogConfig', config)
- self.assertTrue(isinstance(config['LogConfig'], LogConfig))
- self.assertEqual(obj, config['LogConfig'])
+ assert 'LogConfig' in config
+ assert isinstance(config['LogConfig'], LogConfig)
+ assert obj == config['LogConfig']
def test_logconfig_invalid_config_type(self):
with pytest.raises(ValueError):
@@ -342,7 +330,7 @@ class IPAMConfigTest(unittest.TestCase):
gateway='192.168.52.254')
ipam_config = IPAMConfig(pool_configs=[ipam_pool])
- self.assertEqual(ipam_config, {
+ assert ipam_config == {
'Driver': 'default',
'Config': [{
'Subnet': '192.168.52.0/24',
@@ -350,7 +338,7 @@ class IPAMConfigTest(unittest.TestCase):
'AuxiliaryAddresses': None,
'IPRange': None,
}]
- })
+ }
class ServiceModeTest(unittest.TestCase):
diff --git a/tests/unit/models_containers_test.py b/tests/unit/models_containers_test.py
index d7457ba..1fdd7a5 100644
--- a/tests/unit/models_containers_test.py
+++ b/tests/unit/models_containers_test.py
@@ -5,6 +5,7 @@ import unittest
from .fake_api import FAKE_CONTAINER_ID, FAKE_IMAGE_ID, FAKE_EXEC_ID
from .fake_api_client import make_fake_client
+import pytest
class ContainerCollectionTest(unittest.TestCase):
@@ -232,10 +233,10 @@ class ContainerCollectionTest(unittest.TestCase):
client.api.logs.return_value = "some error"
client.api.wait.return_value = 1
- with self.assertRaises(docker.errors.ContainerError) as cm:
+ with pytest.raises(docker.errors.ContainerError) as cm:
client.containers.run('alpine', 'echo hello world')
- assert cm.exception.exit_status == 1
- assert "some error" in str(cm.exception)
+ assert cm.value.exit_status == 1
+ assert "some error" in cm.exconly()
def test_run_with_image_object(self):
client = make_fake_client()
@@ -257,7 +258,7 @@ class ContainerCollectionTest(unittest.TestCase):
client = make_fake_client()
client.api.wait.return_value = 1
- with self.assertRaises(docker.errors.ContainerError):
+ with pytest.raises(docker.errors.ContainerError):
client.containers.run("alpine")
client.api.remove_container.assert_not_called()
@@ -267,18 +268,18 @@ class ContainerCollectionTest(unittest.TestCase):
client = make_fake_client()
client.api.wait.return_value = 1
- with self.assertRaises(docker.errors.ContainerError):
+ with pytest.raises(docker.errors.ContainerError):
client.containers.run("alpine", remove=True)
client.api.remove_container.assert_called_with(FAKE_CONTAINER_ID)
client = make_fake_client()
client.api._version = '1.24'
- with self.assertRaises(RuntimeError):
+ with pytest.raises(RuntimeError):
client.containers.run("alpine", detach=True, remove=True)
client = make_fake_client()
client.api._version = '1.23'
- with self.assertRaises(RuntimeError):
+ with pytest.raises(RuntimeError):
client.containers.run("alpine", detach=True, remove=True)
client = make_fake_client()
diff --git a/tests/unit/ssladapter_test.py b/tests/unit/ssladapter_test.py
index 2b7ce52..73b7336 100644
--- a/tests/unit/ssladapter_test.py
+++ b/tests/unit/ssladapter_test.py
@@ -1,5 +1,6 @@
import unittest
from docker.transport import ssladapter
+import pytest
try:
from backports.ssl_match_hostname import (
@@ -69,11 +70,9 @@ class MatchHostnameTest(unittest.TestCase):
assert match_hostname(self.cert, 'touhou.gensokyo.jp') is None
def test_match_ip_address_failure(self):
- self.assertRaises(
- CertificateError, match_hostname, self.cert, '192.168.0.25'
- )
+ with pytest.raises(CertificateError):
+ match_hostname(self.cert, '192.168.0.25')
def test_match_dns_failure(self):
- self.assertRaises(
- CertificateError, match_hostname, self.cert, 'foobar.co.uk'
- )
+ with pytest.raises(CertificateError):
+ match_hostname(self.cert, 'foobar.co.uk')
diff --git a/tests/unit/swarm_test.py b/tests/unit/swarm_test.py
index 9a66c0c..4385380 100644
--- a/tests/unit/swarm_test.py
+++ b/tests/unit/swarm_test.py
@@ -21,15 +21,11 @@ class SwarmTest(BaseAPIClientTest):
node_id=fake_api.FAKE_NODE_ID, version=1, node_spec=node_spec
)
args = fake_request.call_args
- self.assertEqual(
- args[0][1], url_prefix + 'nodes/24ifsmvkjbyhk/update?version=1'
- )
- self.assertEqual(
- json.loads(args[1]['data']), node_spec
- )
- self.assertEqual(
- args[1]['headers']['Content-Type'], 'application/json'
+ assert args[0][1] == (
+ url_prefix + 'nodes/24ifsmvkjbyhk/update?version=1'
)
+ assert json.loads(args[1]['data']) == node_spec
+ assert args[1]['headers']['Content-Type'] == 'application/json'
@requires_api_version('1.24')
def test_join_swarm(self):
diff --git a/tests/unit/utils_test.py b/tests/unit/utils_test.py
index 2fa1d05..230b1aa 100644
--- a/tests/unit/utils_test.py
+++ b/tests/unit/utils_test.py
@@ -80,25 +80,25 @@ class KwargsFromEnvTest(unittest.TestCase):
os.environ.pop('DOCKER_TLS_VERIFY', None)
kwargs = kwargs_from_env()
- self.assertEqual(None, kwargs.get('base_url'))
- self.assertEqual(None, kwargs.get('tls'))
+ assert kwargs.get('base_url') is None
+ assert kwargs.get('tls') is None
def test_kwargs_from_env_tls(self):
os.environ.update(DOCKER_HOST='tcp://192.168.59.103:2376',
DOCKER_CERT_PATH=TEST_CERT_DIR,
DOCKER_TLS_VERIFY='1')
kwargs = kwargs_from_env(assert_hostname=False)
- self.assertEqual('https://192.168.59.103:2376', kwargs['base_url'])
- self.assertTrue('ca.pem' in kwargs['tls'].ca_cert)
- self.assertTrue('cert.pem' in kwargs['tls'].cert[0])
- self.assertTrue('key.pem' in kwargs['tls'].cert[1])
- self.assertEqual(False, kwargs['tls'].assert_hostname)
- self.assertTrue(kwargs['tls'].verify)
+ assert 'https://192.168.59.103:2376' == kwargs['base_url']
+ assert 'ca.pem' in kwargs['tls'].ca_cert
+ assert 'cert.pem' in kwargs['tls'].cert[0]
+ assert 'key.pem' in kwargs['tls'].cert[1]
+ assert kwargs['tls'].assert_hostname is False
+ assert kwargs['tls'].verify
try:
client = APIClient(**kwargs)
- self.assertEqual(kwargs['base_url'], client.base_url)
- self.assertEqual(kwargs['tls'].ca_cert, client.verify)
- self.assertEqual(kwargs['tls'].cert, client.cert)
+ assert kwargs['base_url'] == client.base_url
+ assert kwargs['tls'].ca_cert == client.verify
+ assert kwargs['tls'].cert == client.cert
except TypeError as e:
self.fail(e)
@@ -107,17 +107,17 @@ class KwargsFromEnvTest(unittest.TestCase):
DOCKER_CERT_PATH=TEST_CERT_DIR,
DOCKER_TLS_VERIFY='')
kwargs = kwargs_from_env(assert_hostname=True)
- self.assertEqual('https://192.168.59.103:2376', kwargs['base_url'])
- self.assertTrue('ca.pem' in kwargs['tls'].ca_cert)
- self.assertTrue('cert.pem' in kwargs['tls'].cert[0])
- self.assertTrue('key.pem' in kwargs['tls'].cert[1])
- self.assertEqual(True, kwargs['tls'].assert_hostname)
- self.assertEqual(False, kwargs['tls'].verify)
+ assert 'https://192.168.59.103:2376' == kwargs['base_url']
+ assert 'ca.pem' in kwargs['tls'].ca_cert
+ assert 'cert.pem' in kwargs['tls'].cert[0]
+ assert 'key.pem' in kwargs['tls'].cert[1]
+ assert kwargs['tls'].assert_hostname is True
+ assert kwargs['tls'].verify is False
try:
client = APIClient(**kwargs)
- self.assertEqual(kwargs['base_url'], client.base_url)
- self.assertEqual(kwargs['tls'].cert, client.cert)
- self.assertFalse(kwargs['tls'].verify)
+ assert kwargs['base_url'] == client.base_url
+ assert kwargs['tls'].cert == client.cert
+ assert not kwargs['tls'].verify
except TypeError as e:
self.fail(e)
@@ -131,7 +131,7 @@ class KwargsFromEnvTest(unittest.TestCase):
DOCKER_TLS_VERIFY='')
os.environ.pop('DOCKER_CERT_PATH', None)
kwargs = kwargs_from_env(assert_hostname=True)
- self.assertEqual('tcp://192.168.59.103:2376', kwargs['base_url'])
+ assert 'tcp://192.168.59.103:2376' == kwargs['base_url']
def test_kwargs_from_env_no_cert_path(self):
try:
@@ -144,10 +144,10 @@ class KwargsFromEnvTest(unittest.TestCase):
DOCKER_TLS_VERIFY='1')
kwargs = kwargs_from_env()
- self.assertTrue(kwargs['tls'].verify)
- self.assertIn(cert_dir, kwargs['tls'].ca_cert)
- self.assertIn(cert_dir, kwargs['tls'].cert[0])
- self.assertIn(cert_dir, kwargs['tls'].cert[1])
+ assert kwargs['tls'].verify
+ assert cert_dir in kwargs['tls'].ca_cert
+ assert cert_dir in kwargs['tls'].cert[0]
+ assert cert_dir in kwargs['tls'].cert[1]
finally:
if temp_dir:
shutil.rmtree(temp_dir)
@@ -169,12 +169,12 @@ class KwargsFromEnvTest(unittest.TestCase):
class ConverVolumeBindsTest(unittest.TestCase):
def test_convert_volume_binds_empty(self):
- self.assertEqual(convert_volume_binds({}), [])
- self.assertEqual(convert_volume_binds([]), [])
+ assert convert_volume_binds({}) == []
+ assert convert_volume_binds([]) == []
def test_convert_volume_binds_list(self):
data = ['/a:/a:ro', '/b:/c:z']
- self.assertEqual(convert_volume_binds(data), data)
+ assert convert_volume_binds(data) == data
def test_convert_volume_binds_complete(self):
data = {
@@ -183,13 +183,13 @@ class ConverVolumeBindsTest(unittest.TestCase):
'mode': 'ro'
}
}
- self.assertEqual(convert_volume_binds(data), ['/mnt/vol1:/data:ro'])
+ assert convert_volume_binds(data) == ['/mnt/vol1:/data:ro']
def test_convert_volume_binds_compact(self):
data = {
'/mnt/vol1': '/data'
}
- self.assertEqual(convert_volume_binds(data), ['/mnt/vol1:/data:rw'])
+ assert convert_volume_binds(data) == ['/mnt/vol1:/data:rw']
def test_convert_volume_binds_no_mode(self):
data = {
@@ -197,7 +197,7 @@ class ConverVolumeBindsTest(unittest.TestCase):
'bind': '/data'
}
}
- self.assertEqual(convert_volume_binds(data), ['/mnt/vol1:/data:rw'])
+ assert convert_volume_binds(data) == ['/mnt/vol1:/data:rw']
def test_convert_volume_binds_unicode_bytes_input(self):
expected = [u'/mnt/지연:/unicode/박:rw']
@@ -208,9 +208,7 @@ class ConverVolumeBindsTest(unittest.TestCase):
'mode': 'rw'
}
}
- self.assertEqual(
- convert_volume_binds(data), expected
- )
+ assert convert_volume_binds(data) == expected
def test_convert_volume_binds_unicode_unicode_input(self):
expected = [u'/mnt/지연:/unicode/박:rw']
@@ -221,9 +219,7 @@ class ConverVolumeBindsTest(unittest.TestCase):
'mode': 'rw'
}
}
- self.assertEqual(
- convert_volume_binds(data), expected
- )
+ assert convert_volume_binds(data) == expected
class ParseEnvFileTest(unittest.TestCase):
@@ -242,38 +238,35 @@ class ParseEnvFileTest(unittest.TestCase):
env_file = self.generate_tempfile(
file_content='USER=jdoe\nPASS=secret')
get_parse_env_file = parse_env_file(env_file)
- self.assertEqual(get_parse_env_file,
- {'USER': 'jdoe', 'PASS': 'secret'})
+ assert get_parse_env_file == {'USER': 'jdoe', 'PASS': 'secret'}
os.unlink(env_file)
def test_parse_env_file_with_equals_character(self):
env_file = self.generate_tempfile(
file_content='USER=jdoe\nPASS=sec==ret')
get_parse_env_file = parse_env_file(env_file)
- self.assertEqual(get_parse_env_file,
- {'USER': 'jdoe', 'PASS': 'sec==ret'})
+ assert get_parse_env_file == {'USER': 'jdoe', 'PASS': 'sec==ret'}
os.unlink(env_file)
def test_parse_env_file_commented_line(self):
env_file = self.generate_tempfile(
file_content='USER=jdoe\n#PASS=secret')
get_parse_env_file = parse_env_file(env_file)
- self.assertEqual(get_parse_env_file, {'USER': 'jdoe'})
+ assert get_parse_env_file == {'USER': 'jdoe'}
os.unlink(env_file)
def test_parse_env_file_newline(self):
env_file = self.generate_tempfile(
file_content='\nUSER=jdoe\n\n\nPASS=secret')
get_parse_env_file = parse_env_file(env_file)
- self.assertEqual(get_parse_env_file,
- {'USER': 'jdoe', 'PASS': 'secret'})
+ assert get_parse_env_file == {'USER': 'jdoe', 'PASS': 'secret'}
os.unlink(env_file)
def test_parse_env_file_invalid_line(self):
env_file = self.generate_tempfile(
file_content='USER jdoe')
- self.assertRaises(
- DockerException, parse_env_file, env_file)
+ with pytest.raises(DockerException):
+ parse_env_file(env_file)
os.unlink(env_file)
@@ -343,46 +336,34 @@ class ParseRepositoryTagTest(unittest.TestCase):
sha = 'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855'
def test_index_image_no_tag(self):
- self.assertEqual(
- parse_repository_tag("root"), ("root", None)
- )
+ assert parse_repository_tag("root") == ("root", None)
def test_index_image_tag(self):
- self.assertEqual(
- parse_repository_tag("root:tag"), ("root", "tag")
- )
+ assert parse_repository_tag("root:tag") == ("root", "tag")
def test_index_user_image_no_tag(self):
- self.assertEqual(
- parse_repository_tag("user/repo"), ("user/repo", None)
- )
+ assert parse_repository_tag("user/repo") == ("user/repo", None)
def test_index_user_image_tag(self):
- self.assertEqual(
- parse_repository_tag("user/repo:tag"), ("user/repo", "tag")
- )
+ assert parse_repository_tag("user/repo:tag") == ("user/repo", "tag")
def test_private_reg_image_no_tag(self):
- self.assertEqual(
- parse_repository_tag("url:5000/repo"), ("url:5000/repo", None)
- )
+ assert parse_repository_tag("url:5000/repo") == ("url:5000/repo", None)
def test_private_reg_image_tag(self):
- self.assertEqual(
- parse_repository_tag("url:5000/repo:tag"), ("url:5000/repo", "tag")
+ assert parse_repository_tag("url:5000/repo:tag") == (
+ "url:5000/repo", "tag"
)
def test_index_image_sha(self):
- self.assertEqual(
- parse_repository_tag("root@sha256:{0}".format(self.sha)),
- ("root", "sha256:{0}".format(self.sha))
+ assert parse_repository_tag("root@sha256:{0}".format(self.sha)) == (
+ "root", "sha256:{0}".format(self.sha)
)
def test_private_reg_image_sha(self):
- self.assertEqual(
- parse_repository_tag("url:5000/repo@sha256:{0}".format(self.sha)),
- ("url:5000/repo", "sha256:{0}".format(self.sha))
- )
+ assert parse_repository_tag(
+ "url:5000/repo@sha256:{0}".format(self.sha)
+ ) == ("url:5000/repo", "sha256:{0}".format(self.sha))
class ParseDeviceTest(unittest.TestCase):
@@ -392,35 +373,35 @@ class ParseDeviceTest(unittest.TestCase):
'PathInContainer': '/dev/mnt1',
'CgroupPermissions': 'r'
}])
- self.assertEqual(devices[0], {
+ assert devices[0] == {
'PathOnHost': '/dev/sda1',
'PathInContainer': '/dev/mnt1',
'CgroupPermissions': 'r'
- })
+ }
def test_partial_string_definition(self):
devices = parse_devices(['/dev/sda1'])
- self.assertEqual(devices[0], {
+ assert devices[0] == {
'PathOnHost': '/dev/sda1',
'PathInContainer': '/dev/sda1',
'CgroupPermissions': 'rwm'
- })
+ }
def test_permissionless_string_definition(self):
devices = parse_devices(['/dev/sda1:/dev/mnt1'])
- self.assertEqual(devices[0], {
+ assert devices[0] == {
'PathOnHost': '/dev/sda1',
'PathInContainer': '/dev/mnt1',
'CgroupPermissions': 'rwm'
- })
+ }
def test_full_string_definition(self):
devices = parse_devices(['/dev/sda1:/dev/mnt1:r'])
- self.assertEqual(devices[0], {
+ assert devices[0] == {
'PathOnHost': '/dev/sda1',
'PathInContainer': '/dev/mnt1',
'CgroupPermissions': 'r'
- })
+ }
def test_hybrid_list(self):
devices = parse_devices([
@@ -432,36 +413,38 @@ class ParseDeviceTest(unittest.TestCase):
}
])
- self.assertEqual(devices[0], {
+ assert devices[0] == {
'PathOnHost': '/dev/sda1',
'PathInContainer': '/dev/mnt1',
'CgroupPermissions': 'rw'
- })
- self.assertEqual(devices[1], {
+ }
+ assert devices[1] == {
'PathOnHost': '/dev/sda2',
'PathInContainer': '/dev/mnt2',
'CgroupPermissions': 'r'
- })
+ }
class ParseBytesTest(unittest.TestCase):
def test_parse_bytes_valid(self):
- self.assertEqual(parse_bytes("512MB"), 536870912)
- self.assertEqual(parse_bytes("512M"), 536870912)
- self.assertEqual(parse_bytes("512m"), 536870912)
+ assert parse_bytes("512MB") == 536870912
+ assert parse_bytes("512M") == 536870912
+ assert parse_bytes("512m") == 536870912
def test_parse_bytes_invalid(self):
- self.assertRaises(DockerException, parse_bytes, "512MK")
- self.assertRaises(DockerException, parse_bytes, "512L")
- self.assertRaises(DockerException, parse_bytes, "127.0.0.1K")
+ with pytest.raises(DockerException):
+ parse_bytes("512MK")
+ with pytest.raises(DockerException):
+ parse_bytes("512L")
+ with pytest.raises(DockerException):
+ parse_bytes("127.0.0.1K")
def test_parse_bytes_float(self):
- self.assertRaises(DockerException, parse_bytes, "1.5k")
+ with pytest.raises(DockerException):
+ parse_bytes("1.5k")
def test_parse_bytes_maxint(self):
- self.assertEqual(
- parse_bytes("{0}k".format(sys.maxsize)), sys.maxsize * 1024
- )
+ assert parse_bytes("{0}k".format(sys.maxsize)) == sys.maxsize * 1024
class UtilsTest(unittest.TestCase):
@@ -476,7 +459,7 @@ class UtilsTest(unittest.TestCase):
]
for filters, expected in tests:
- self.assertEqual(convert_filters(filters), expected)
+ assert convert_filters(filters) == expected
def test_decode_json_header(self):
obj = {'a': 'b', 'c': 1}
@@ -486,144 +469,144 @@ class UtilsTest(unittest.TestCase):
else:
data = base64.urlsafe_b64encode(json.dumps(obj))
decoded_data = decode_json_header(data)
- self.assertEqual(obj, decoded_data)
+ assert obj == decoded_data
class SplitCommandTest(unittest.TestCase):
def test_split_command_with_unicode(self):
- self.assertEqual(split_command(u'echo μμ'), ['echo', 'μμ'])
+ assert split_command(u'echo μμ') == ['echo', 'μμ']
@pytest.mark.skipif(six.PY3, reason="shlex doesn't support bytes in py3")
def test_split_command_with_bytes(self):
- self.assertEqual(split_command('echo μμ'), ['echo', 'μμ'])
+ assert split_command('echo μμ') == ['echo', 'μμ']
class PortsTest(unittest.TestCase):
def test_split_port_with_host_ip(self):
internal_port, external_port = split_port("127.0.0.1:1000:2000")
- self.assertEqual(internal_port, ["2000"])
- self.assertEqual(external_port, [("127.0.0.1", "1000")])
+ assert internal_port == ["2000"]
+ assert external_port == [("127.0.0.1", "1000")]
def test_split_port_with_protocol(self):
internal_port, external_port = split_port("127.0.0.1:1000:2000/udp")
- self.assertEqual(internal_port, ["2000/udp"])
- self.assertEqual(external_port, [("127.0.0.1", "1000")])
+ assert internal_port == ["2000/udp"]
+ assert external_port == [("127.0.0.1", "1000")]
def test_split_port_with_host_ip_no_port(self):
internal_port, external_port = split_port("127.0.0.1::2000")
- self.assertEqual(internal_port, ["2000"])
- self.assertEqual(external_port, [("127.0.0.1", None)])
+ assert internal_port == ["2000"]
+ assert external_port == [("127.0.0.1", None)]
def test_split_port_range_with_host_ip_no_port(self):
internal_port, external_port = split_port("127.0.0.1::2000-2001")
- self.assertEqual(internal_port, ["2000", "2001"])
- self.assertEqual(external_port,
- [("127.0.0.1", None), ("127.0.0.1", None)])
+ assert internal_port == ["2000", "2001"]
+ assert external_port == [("127.0.0.1", None), ("127.0.0.1", None)]
def test_split_port_with_host_port(self):
internal_port, external_port = split_port("1000:2000")
- self.assertEqual(internal_port, ["2000"])
- self.assertEqual(external_port, ["1000"])
+ assert internal_port == ["2000"]
+ assert external_port == ["1000"]
def test_split_port_range_with_host_port(self):
internal_port, external_port = split_port("1000-1001:2000-2001")
- self.assertEqual(internal_port, ["2000", "2001"])
- self.assertEqual(external_port, ["1000", "1001"])
+ assert internal_port == ["2000", "2001"]
+ assert external_port == ["1000", "1001"]
def test_split_port_random_port_range_with_host_port(self):
internal_port, external_port = split_port("1000-1001:2000")
- self.assertEqual(internal_port, ["2000"])
- self.assertEqual(external_port, ["1000-1001"])
+ assert internal_port == ["2000"]
+ assert external_port == ["1000-1001"]
def test_split_port_no_host_port(self):
internal_port, external_port = split_port("2000")
- self.assertEqual(internal_port, ["2000"])
- self.assertEqual(external_port, None)
+ assert internal_port == ["2000"]
+ assert external_port is None
def test_split_port_range_no_host_port(self):
internal_port, external_port = split_port("2000-2001")
- self.assertEqual(internal_port, ["2000", "2001"])
- self.assertEqual(external_port, None)
+ assert internal_port == ["2000", "2001"]
+ assert external_port is None
def test_split_port_range_with_protocol(self):
internal_port, external_port = split_port(
"127.0.0.1:1000-1001:2000-2001/udp")
- self.assertEqual(internal_port, ["2000/udp", "2001/udp"])
- self.assertEqual(external_port,
- [("127.0.0.1", "1000"), ("127.0.0.1", "1001")])
+ assert internal_port == ["2000/udp", "2001/udp"]
+ assert external_port == [("127.0.0.1", "1000"), ("127.0.0.1", "1001")]
def test_split_port_with_ipv6_address(self):
internal_port, external_port = split_port(
"2001:abcd:ef00::2:1000:2000")
- self.assertEqual(internal_port, ["2000"])
- self.assertEqual(external_port, [("2001:abcd:ef00::2", "1000")])
+ assert internal_port == ["2000"]
+ assert external_port == [("2001:abcd:ef00::2", "1000")]
def test_split_port_invalid(self):
- self.assertRaises(ValueError,
- lambda: split_port("0.0.0.0:1000:2000:tcp"))
+ with pytest.raises(ValueError):
+ split_port("0.0.0.0:1000:2000:tcp")
def test_non_matching_length_port_ranges(self):
- self.assertRaises(
- ValueError,
- lambda: split_port("0.0.0.0:1000-1010:2000-2002/tcp")
- )
+ with pytest.raises(ValueError):
+ split_port("0.0.0.0:1000-1010:2000-2002/tcp")
def test_port_and_range_invalid(self):
- self.assertRaises(ValueError,
- lambda: split_port("0.0.0.0:1000:2000-2002/tcp"))
+ with pytest.raises(ValueError):
+ split_port("0.0.0.0:1000:2000-2002/tcp")
def test_port_only_with_colon(self):
- self.assertRaises(ValueError,
- lambda: split_port(":80"))
+ with pytest.raises(ValueError):
+ split_port(":80")
def test_host_only_with_colon(self):
- self.assertRaises(ValueError,
- lambda: split_port("localhost:"))
+ with pytest.raises(ValueError):
+ split_port("localhost:")
def test_with_no_container_port(self):
- self.assertRaises(ValueError,
- lambda: split_port("localhost:80:"))
+ with pytest.raises(ValueError):
+ split_port("localhost:80:")
def test_split_port_empty_string(self):
- self.assertRaises(ValueError, lambda: split_port(""))
+ with pytest.raises(ValueError):
+ split_port("")
def test_split_port_non_string(self):
assert split_port(1243) == (['1243'], None)
def test_build_port_bindings_with_one_port(self):
port_bindings = build_port_bindings(["127.0.0.1:1000:1000"])
- self.assertEqual(port_bindings["1000"], [("127.0.0.1", "1000")])
+ assert port_bindings["1000"] == [("127.0.0.1", "1000")]
def test_build_port_bindings_with_matching_internal_ports(self):
port_bindings = build_port_bindings(
["127.0.0.1:1000:1000", "127.0.0.1:2000:1000"])
- self.assertEqual(port_bindings["1000"],
- [("127.0.0.1", "1000"), ("127.0.0.1", "2000")])
+ assert port_bindings["1000"] == [
+ ("127.0.0.1", "1000"), ("127.0.0.1", "2000")
+ ]
def test_build_port_bindings_with_nonmatching_internal_ports(self):
port_bindings = build_port_bindings(
["127.0.0.1:1000:1000", "127.0.0.1:2000:2000"])
- self.assertEqual(port_bindings["1000"], [("127.0.0.1", "1000")])
- self.assertEqual(port_bindings["2000"], [("127.0.0.1", "2000")])
+ assert port_bindings["1000"] == [("127.0.0.1", "1000")]
+ assert port_bindings["2000"] == [("127.0.0.1", "2000")]
def test_build_port_bindings_with_port_range(self):
port_bindings = build_port_bindings(["127.0.0.1:1000-1001:1000-1001"])
- self.assertEqual(port_bindings["1000"], [("127.0.0.1", "1000")])
- self.assertEqual(port_bindings["1001"], [("127.0.0.1", "1001")])
+ assert port_bindings["1000"] == [("127.0.0.1", "1000")]
+ assert port_bindings["1001"] == [("127.0.0.1", "1001")]
def test_build_port_bindings_with_matching_internal_port_ranges(self):
port_bindings = build_port_bindings(
["127.0.0.1:1000-1001:1000-1001", "127.0.0.1:2000-2001:1000-1001"])
- self.assertEqual(port_bindings["1000"],
- [("127.0.0.1", "1000"), ("127.0.0.1", "2000")])
- self.assertEqual(port_bindings["1001"],
- [("127.0.0.1", "1001"), ("127.0.0.1", "2001")])
+ assert port_bindings["1000"] == [
+ ("127.0.0.1", "1000"), ("127.0.0.1", "2000")
+ ]
+ assert port_bindings["1001"] == [
+ ("127.0.0.1", "1001"), ("127.0.0.1", "2001")
+ ]
def test_build_port_bindings_with_nonmatching_internal_port_ranges(self):
port_bindings = build_port_bindings(
["127.0.0.1:1000:1000", "127.0.0.1:2000:2000"])
- self.assertEqual(port_bindings["1000"], [("127.0.0.1", "1000")])
- self.assertEqual(port_bindings["2000"], [("127.0.0.1", "2000")])
+ assert port_bindings["1000"] == [("127.0.0.1", "1000")]
+ assert port_bindings["2000"] == [("127.0.0.1", "2000")]
def convert_paths(collection):
@@ -708,11 +691,13 @@ class ExcludePathsTest(unittest.TestCase):
If we're using a custom Dockerfile, make sure that's not
excluded.
"""
- assert self.exclude(['*'], dockerfile='Dockerfile.alt') == \
- set(['Dockerfile.alt', '.dockerignore'])
+ assert self.exclude(['*'], dockerfile='Dockerfile.alt') == set(
+ ['Dockerfile.alt', '.dockerignore']
+ )
- assert self.exclude(['*'], dockerfile='foo/Dockerfile3') == \
- convert_paths(set(['foo/Dockerfile3', '.dockerignore']))
+ assert self.exclude(
+ ['*'], dockerfile='foo/Dockerfile3'
+ ) == convert_paths(set(['foo/Dockerfile3', '.dockerignore']))
def test_exclude_dockerfile_child(self):
includes = self.exclude(['foo/'], dockerfile='foo/Dockerfile3')
@@ -946,7 +931,7 @@ class TarTest(unittest.TestCase):
os.makedirs(os.path.join(base, d))
with tar(base) as archive:
tar_data = tarfile.open(fileobj=archive)
- self.assertEqual(sorted(tar_data.getnames()), ['bar', 'foo'])
+ assert sorted(tar_data.getnames()) == ['bar', 'foo']
@pytest.mark.skipif(IS_WINDOWS_PLATFORM, reason='No symlinks on Windows')
def test_tar_with_file_symlinks(self):
@@ -958,9 +943,7 @@ class TarTest(unittest.TestCase):
os.symlink('../foo', os.path.join(base, 'bar/foo'))
with tar(base) as archive:
tar_data = tarfile.open(fileobj=archive)
- self.assertEqual(
- sorted(tar_data.getnames()), ['bar', 'bar/foo', 'foo']
- )
+ assert sorted(tar_data.getnames()) == ['bar', 'bar/foo', 'foo']
@pytest.mark.skipif(IS_WINDOWS_PLATFORM, reason='No symlinks on Windows')
def test_tar_with_directory_symlinks(self):
@@ -971,9 +954,7 @@ class TarTest(unittest.TestCase):
os.symlink('../foo', os.path.join(base, 'bar/foo'))
with tar(base) as archive:
tar_data = tarfile.open(fileobj=archive)
- self.assertEqual(
- sorted(tar_data.getnames()), ['bar', 'bar/foo', 'foo']
- )
+ assert sorted(tar_data.getnames()) == ['bar', 'bar/foo', 'foo']
@pytest.mark.skipif(IS_WINDOWS_PLATFORM, reason='No UNIX sockets on Win32')
def test_tar_socket_file(self):
@@ -986,9 +967,7 @@ class TarTest(unittest.TestCase):
sock.bind(os.path.join(base, 'test.sock'))
with tar(base) as archive:
tar_data = tarfile.open(fileobj=archive)
- self.assertEqual(
- sorted(tar_data.getnames()), ['bar', 'foo']
- )
+ assert sorted(tar_data.getnames()) == ['bar', 'foo']
class ShouldCheckDirectoryTest(unittest.TestCase):
View Lorry Controller Status