Merge "Add glance image import copy-image tests"

Zuul 2 years ago committed by Gerrit Code Review
commit ae784996c9

@ -13,10 +13,16 @@
# License for the specific language governing permissions and limitations
# under the License.
import six
from tempest.api.image import base
from tempest.common import waiters
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
CONF = config.CONF
class BasicOperationsImagesAdminTest(base.BaseV2ImageAdminTest):
""""Test image operations about image owner"""
@ -52,3 +58,65 @@ class BasicOperationsImagesAdminTest(base.BaseV2ImageAdminTest):
self.assertEqual(random_id_2, updated_image_info['owner'])
class ImportCopyImagesTest(base.BaseV2ImageAdminTest):
"""Test the import copy-image operations"""
def skip_checks(cls):
super(ImportCopyImagesTest, cls).skip_checks()
if not CONF.image_feature_enabled.import_image:
skip_msg = (
"%s skipped as image import is not available" % cls.__name__)
raise cls.skipException(skip_msg)
def test_image_copy_image_import(self):
"""Test 'copy-image' import functionalities
Create image, import image with copy-image method and
verify that import succeeded.
available_stores = self.get_available_stores()
available_import_methods = self.client.info_import()[
# NOTE(gmann): Skip if copy-image import method and multistore
# are not available.
if ('copy-image' not in available_import_methods or
not available_stores):
raise self.skipException('Either copy-image import method or '
'multistore is not available')
uuid = data_utils.rand_uuid()
image_name = data_utils.rand_name('copy-image')
container_format = CONF.image.container_formats[0]
disk_format = CONF.image.disk_formats[0]
image = self.create_image(name=image_name,
self.assertEqual('queued', image['status'])
file_content = data_utils.random_bytes()
image_file = six.BytesIO(file_content)
self.client.store_image_file(image['id'], image_file)
body = self.client.show_image(image['id'])
self.assertEqual(image['id'], body['id'])
self.assertEqual(len(file_content), body.get('size'))
self.assertEqual('active', body['status'])
# Copy image to all the stores. In case of all_stores request
# glance will skip the stores where image is already available.
self.admin_client.image_import(image['id'], method='copy-image',
# Wait for copy to finished on all stores.
failed_stores = waiters.wait_for_image_copied_to_stores(
self.client, image['id'])
# Assert if copy is failed on any store.
self.assertEqual(0, len(failed_stores),
"Failed to copy the following stores: %s" %

@ -215,6 +215,37 @@ def wait_for_image_imported_to_stores(client, image_id, stores):
raise lib_exc.TimeoutException(message)
def wait_for_image_copied_to_stores(client, image_id):
"""Waits for an image to be copied on all requested stores.
The client should also have build_interval and build_timeout attributes.
This return the list of stores where copy is failed.
start = int(time.time())
store_left = []
while int(time.time()) - start < client.build_timeout:
image = client.show_image(image_id)
store_left = image.get('os_glance_importing_to_stores')
# NOTE(danms): If os_glance_importing_to_stores is None, then
# we've raced with the startup of the task and should continue
# to wait.
if store_left is not None and not store_left:
return image['os_glance_failed_import']
if image['status'].lower() == 'killed':
raise exceptions.ImageKilledException(image_id=image_id,
message = ('Image %(image_id)s failed to finish the copy operation '
'on stores: %s' % str(store_left))
caller = test_utils.find_test_caller()
if caller:
message = '(%s) %s' % (caller, message)
raise lib_exc.TimeoutException(message)
def wait_for_volume_resource_status(client, resource_id, status):
"""Waits for a volume resource to reach a given status.