mirror of
https://github.com/docker/compose.git
synced 2025-10-31 11:14:02 +01:00
127 lines
4.3 KiB
Python
127 lines
4.3 KiB
Python
from __future__ import absolute_import
|
|
from __future__ import unicode_literals
|
|
|
|
import six
|
|
from docker.errors import DockerException
|
|
|
|
from .testcases import DockerClientTestCase
|
|
from .testcases import no_cluster
|
|
from compose.const import LABEL_PROJECT
|
|
from compose.const import LABEL_VOLUME
|
|
from compose.volume import Volume
|
|
|
|
|
|
class VolumeTest(DockerClientTestCase):
|
|
def setUp(self):
|
|
self.tmp_volumes = []
|
|
|
|
def tearDown(self):
|
|
for volume in self.tmp_volumes:
|
|
try:
|
|
self.client.remove_volume(volume.full_name)
|
|
except DockerException:
|
|
pass
|
|
del self.tmp_volumes
|
|
super(VolumeTest, self).tearDown()
|
|
|
|
def create_volume(self, name, driver=None, opts=None, external=None, custom_name=False):
|
|
if external:
|
|
custom_name = True
|
|
if isinstance(external, six.text_type):
|
|
name = external
|
|
|
|
vol = Volume(
|
|
self.client, 'composetest', name, driver=driver, driver_opts=opts,
|
|
external=bool(external), custom_name=custom_name
|
|
)
|
|
self.tmp_volumes.append(vol)
|
|
return vol
|
|
|
|
def test_create_volume(self):
|
|
vol = self.create_volume('volume01')
|
|
vol.create()
|
|
info = self.get_volume_data(vol.full_name)
|
|
assert info['Name'].split('/')[-1] == vol.full_name
|
|
|
|
def test_create_volume_custom_name(self):
|
|
vol = self.create_volume('volume01', custom_name=True)
|
|
assert vol.name == vol.full_name
|
|
vol.create()
|
|
info = self.get_volume_data(vol.full_name)
|
|
assert info['Name'].split('/')[-1] == vol.name
|
|
|
|
def test_recreate_existing_volume(self):
|
|
vol = self.create_volume('volume01')
|
|
|
|
vol.create()
|
|
info = self.get_volume_data(vol.full_name)
|
|
assert info['Name'].split('/')[-1] == vol.full_name
|
|
|
|
vol.create()
|
|
info = self.get_volume_data(vol.full_name)
|
|
assert info['Name'].split('/')[-1] == vol.full_name
|
|
|
|
@no_cluster('inspect volume by name defect on Swarm Classic')
|
|
def test_inspect_volume(self):
|
|
vol = self.create_volume('volume01')
|
|
vol.create()
|
|
info = vol.inspect()
|
|
assert info['Name'] == vol.full_name
|
|
|
|
@no_cluster('remove volume by name defect on Swarm Classic')
|
|
def test_remove_volume(self):
|
|
vol = Volume(self.client, 'composetest', 'volume01')
|
|
vol.create()
|
|
vol.remove()
|
|
volumes = self.client.volumes()['Volumes']
|
|
assert len([v for v in volumes if v['Name'] == vol.full_name]) == 0
|
|
|
|
@no_cluster('inspect volume by name defect on Swarm Classic')
|
|
def test_external_volume(self):
|
|
vol = self.create_volume('composetest_volume_ext', external=True)
|
|
assert vol.external is True
|
|
assert vol.full_name == vol.name
|
|
vol.create()
|
|
info = vol.inspect()
|
|
assert info['Name'] == vol.name
|
|
|
|
@no_cluster('inspect volume by name defect on Swarm Classic')
|
|
def test_external_aliased_volume(self):
|
|
alias_name = 'composetest_alias01'
|
|
vol = self.create_volume('volume01', external=alias_name)
|
|
assert vol.external is True
|
|
assert vol.full_name == alias_name
|
|
vol.create()
|
|
info = vol.inspect()
|
|
assert info['Name'] == alias_name
|
|
|
|
@no_cluster('inspect volume by name defect on Swarm Classic')
|
|
def test_exists(self):
|
|
vol = self.create_volume('volume01')
|
|
assert vol.exists() is False
|
|
vol.create()
|
|
assert vol.exists() is True
|
|
|
|
@no_cluster('inspect volume by name defect on Swarm Classic')
|
|
def test_exists_external(self):
|
|
vol = self.create_volume('volume01', external=True)
|
|
assert vol.exists() is False
|
|
vol.create()
|
|
assert vol.exists() is True
|
|
|
|
@no_cluster('inspect volume by name defect on Swarm Classic')
|
|
def test_exists_external_aliased(self):
|
|
vol = self.create_volume('volume01', external='composetest_alias01')
|
|
assert vol.exists() is False
|
|
vol.create()
|
|
assert vol.exists() is True
|
|
|
|
@no_cluster('inspect volume by name defect on Swarm Classic')
|
|
def test_volume_default_labels(self):
|
|
vol = self.create_volume('volume01')
|
|
vol.create()
|
|
vol_data = vol.inspect()
|
|
labels = vol_data['Labels']
|
|
assert labels[LABEL_VOLUME] == vol.name
|
|
assert labels[LABEL_PROJECT] == vol.project
|