blob: a8d9193f75994b434c2b07e23b2c788e29561412 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
|
import json
import pytest
from ..helpers import requires_api_version
from .api_test import BaseAPIClientTest, url_prefix, fake_request
class VolumeTest(BaseAPIClientTest):
def test_list_volumes(self):
volumes = self.client.volumes()
assert 'Volumes' in volumes
assert len(volumes['Volumes']) == 2
args = fake_request.call_args
assert args[0][0] == 'GET'
assert args[0][1] == url_prefix + 'volumes'
def test_list_volumes_and_filters(self):
volumes = self.client.volumes(filters={'dangling': True})
assert 'Volumes' in volumes
assert len(volumes['Volumes']) == 2
args = fake_request.call_args
assert args[0][0] == 'GET'
assert args[0][1] == url_prefix + 'volumes'
assert args[1] == {'params': {'filters': '{"dangling": ["true"]}'},
'timeout': 60}
def test_create_volume(self):
name = 'perfectcherryblossom'
result = self.client.create_volume(name)
assert 'Name' in result
assert result['Name'] == name
assert 'Driver' in result
assert result['Driver'] == 'local'
args = fake_request.call_args
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'
})
assert result["Labels"] == {
'com.example.some-label': 'some-value'
}
@requires_api_version('1.23')
def test_create_volume_with_invalid_labels(self):
name = 'perfectcherryblossom'
with pytest.raises(TypeError):
self.client.create_volume(name, labels=1)
def test_create_volume_with_driver(self):
name = 'perfectcherryblossom'
driver_name = 'sshfs'
self.client.create_volume(name, driver=driver_name)
args = fake_request.call_args
assert args[0][0] == 'POST'
assert args[0][1] == url_prefix + 'volumes/create'
data = json.loads(args[1]['data'])
assert 'Driver' in data
assert data['Driver'] == driver_name
def test_create_volume_invalid_opts_type(self):
with pytest.raises(TypeError):
self.client.create_volume(
'perfectcherryblossom', driver_opts='hello=world'
)
with pytest.raises(TypeError):
self.client.create_volume(
'perfectcherryblossom', driver_opts=['hello=world']
)
with pytest.raises(TypeError):
self.client.create_volume(
'perfectcherryblossom', driver_opts=''
)
@requires_api_version('1.24')
def test_create_volume_with_no_specified_name(self):
result = self.client.create_volume(name=None)
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'
def test_inspect_volume(self):
name = 'perfectcherryblossom'
result = self.client.inspect_volume(name)
assert 'Name' in result
assert result['Name'] == name
assert 'Driver' in result
assert result['Driver'] == 'local'
args = fake_request.call_args
assert args[0][0] == 'GET'
assert args[0][1] == f'{url_prefix}volumes/{name}'
def test_remove_volume(self):
name = 'perfectcherryblossom'
self.client.remove_volume(name)
args = fake_request.call_args
assert args[0][0] == 'DELETE'
assert args[0][1] == f'{url_prefix}volumes/{name}'
|
