openstack/shade
Code Issues Proposed changes

Merge "Cleanup volumes in functional tests in parallel"

Browse Source
This commit is contained in:
Jenkins 2017-06-14 03:51:57 +00:00 committed by Gerrit Code Review
commit cb70320595
1 changed files with 20 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
shade/tests/functional/test_volume.py
View File

@ -19,6 +19,7 @@ Functional tests for `shade` block storage methods.
from testtools import content from testtools import content
from shade import _utils
from shade.tests.functional import base from shade.tests.functional import base
@ -77,7 +78,7 @@ class TestVolume(base.BaseFunctionalTestCase):
self.user_cloud.delete_image(image_name, wait=True) self.user_cloud.delete_image(image_name, wait=True)
self.user_cloud.delete_volume(volume_name, wait=True) self.user_cloud.delete_volume(volume_name, wait=True)
def cleanup(self, volume_name, snapshot_name=None, image_name=None): def cleanup(self, volume, snapshot_name=None, image_name=None):
# Need to delete snapshots before volumes # Need to delete snapshots before volumes
if snapshot_name: if snapshot_name:
snapshot = self.user_cloud.get_volume_snapshot(snapshot_name) snapshot = self.user_cloud.get_volume_snapshot(snapshot_name)
@ -88,9 +89,23 @@ class TestVolume(base.BaseFunctionalTestCase):
image = self.user_cloud.get_image(image_name) image = self.user_cloud.get_image(image_name)
if image: if image:
self.user_cloud.delete_image(image_name, wait=True) self.user_cloud.delete_image(image_name, wait=True)
volume = self.user_cloud.get_volume(volume_name) if not isinstance(volume, list):
if volume: self.user_cloud.delete_volume(volume, wait=True)
self.user_cloud.delete_volume(volume_name, wait=True) else:
# We have more than one volume to clean up - submit all of the
# deletes without wait, then poll until none of them are found
# in the volume list anymore
for v in volume:
self.user_cloud.delete_volume(v, wait=False)
for count in _utils._iterate_timeout(
180, "Timeout waiting for volume cleanup"):
found = False
for existing in self.user_cloud.list_volumes():
for v in volume:
if v['id'] == existing['id']:
found = True
if not found:
break
def test_list_volumes_pagination(self): def test_list_volumes_pagination(self):
'''Test pagination for list volumes functionality''' '''Test pagination for list volumes functionality'''
@ -101,9 +116,9 @@ class TestVolume(base.BaseFunctionalTestCase):
num_volumes = 8 num_volumes = 8
for i in range(num_volumes): for i in range(num_volumes):
name = self.getUniqueString() name = self.getUniqueString()
self.addCleanup(self.cleanup, name)
v = self.user_cloud.create_volume(display_name=name, size=1) v = self.user_cloud.create_volume(display_name=name, size=1)
volumes.append(v) volumes.append(v)
self.addCleanup(self.cleanup, volumes)
result = [] result = []
for i in self.user_cloud.list_volumes(): for i in self.user_cloud.list_volumes():
if i['name'] and i['name'].startswith(self.id()): if i['name'] and i['name'].startswith(self.id()):