Source code for compute.images.test_list_image_filters
# Copyright 2012 OpenStack Foundation
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import time
import io
import testtools
from tempest.api.compute import base
from tempest.common import waiters
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib.common.utils import test_utils
from tempest.lib import decorators
from tempest.lib import exceptions
CONF = config.CONF
[docs]
class ListImageFiltersTestJSON(base.BaseV2ComputeTest):
"""Test listing server images with compute microversion less than 2.36"""
max_microversion = '2.35'
@classmethod
def skip_checks(cls):
super(ListImageFiltersTestJSON, cls).skip_checks()
if not CONF.service_available.glance:
skip_msg = ("%s skipped as glance is not available" % cls.__name__)
raise cls.skipException(skip_msg)
@classmethod
def setup_clients(cls):
super(ListImageFiltersTestJSON, cls).setup_clients()
cls.client = cls.compute_images_client
if CONF.image_feature_enabled.api_v2:
cls.glance_client = cls.os_primary.image_client_v2
else:
raise exceptions.InvalidConfiguration(
'api_v2 must be True in [image-feature-enabled].')
@classmethod
def resource_setup(cls):
super(ListImageFiltersTestJSON, cls).resource_setup()
def _create_image():
image_name_kwargs = {
'prefix': CONF.resource_name_prefix,
'name': cls.__name__ + '-image'
}
params = {
'name': data_utils.rand_name(**image_name_kwargs),
'container_format': 'bare',
'disk_format': 'raw',
'visibility': 'private'
}
body = cls.glance_client.create_image(**params)
body = body['image'] if 'image' in body else body
image_id = body['id']
cls.addClassResourceCleanup(
test_utils.call_and_ignore_notfound_exc,
cls.compute_images_client.delete_image,
image_id)
# Wait 1 second between creation and upload to ensure a delta
# between created_at and updated_at.
time.sleep(1)
image_file = io.BytesIO((b'*' * 1024))
cls.glance_client.store_image_file(image_id, data=image_file)
waiters.wait_for_image_status(cls.client, image_id, 'ACTIVE')
body = cls.client.show_image(image_id)['image']
return body
# Create non-snapshot images via glance
cls.image1 = _create_image()
cls.image1_id = cls.image1['id']
cls.image2 = _create_image()
cls.image2_id = cls.image2['id']
cls.image3 = _create_image()
cls.image3_id = cls.image3['id']
if not CONF.compute_feature_enabled.snapshot:
return
# Create instances and snapshots via nova
cls.server1 = cls.create_test_server()
cls.server2 = cls.create_test_server(wait_until='ACTIVE')
# NOTE(sdague) this is faster than doing the sync wait_util on both
waiters.wait_for_server_status(cls.servers_client,
cls.server1['id'], 'ACTIVE')
# Create images to be used in the filter tests
cls.snapshot1 = cls.create_image_from_server(
cls.server1['id'], wait_until='ACTIVE')
cls.snapshot1_id = cls.snapshot1['id']
# Servers have a hidden property for when they are being imaged
# Performing back-to-back create image calls on a single
# server will sometimes cause failures
cls.snapshot3 = cls.create_image_from_server(
cls.server2['id'], wait_until='ACTIVE')
cls.snapshot3_id = cls.snapshot3['id']
# Wait for the server to be active after the image upload
cls.snapshot2 = cls.create_image_from_server(
cls.server1['id'], wait_until='ACTIVE')
cls.snapshot2_id = cls.snapshot2['id']
[docs]
@decorators.idempotent_id('a3f5b513-aeb3-42a9-b18e-f091ef73254d')
def test_list_images_filter_by_status(self):
"""Test listing server images filtered by image status
The list of images should contain only images with the
provided image status.
"""
params = {'status': 'ACTIVE'}
images = self.client.list_images(**params)['images']
self.assertNotEmpty([i for i in images if i['id'] == self.image1_id])
self.assertNotEmpty([i for i in images if i['id'] == self.image2_id])
self.assertNotEmpty([i for i in images if i['id'] == self.image3_id])
[docs]
@decorators.idempotent_id('33163b73-79f5-4d07-a7ea-9213bcc468ff')
def test_list_images_filter_by_name(self):
"""Test listing server images filtered by image name
The list of images should contain only images with the
provided image name.
"""
params = {'name': self.image1['name']}
images = self.client.list_images(**params)['images']
self.assertNotEmpty([i for i in images if i['id'] == self.image1_id])
self.assertEmpty([i for i in images if i['id'] == self.image2_id])
self.assertEmpty([i for i in images if i['id'] == self.image3_id])
[docs]
@decorators.idempotent_id('9f238683-c763-45aa-b848-232ec3ce3105')
@testtools.skipUnless(CONF.compute_feature_enabled.snapshot,
'Snapshotting is not available.')
def test_list_images_filter_by_server_id(self):
"""Test listing images filtered by server id
The list of images should contain only images with the
provided server id.
"""
params = {'server': self.server1['id']}
images = self.client.list_images(**params)['images']
self.assertNotEmpty([i for i in images
if i['id'] == self.snapshot1_id],
"Failed to find image %s in images. "
"Got images %s" % (self.image1_id, images))
self.assertNotEmpty([i for i in images
if i['id'] == self.snapshot2_id])
self.assertEmpty([i for i in images if i['id'] == self.snapshot3_id])
[docs]
@decorators.idempotent_id('05a377b8-28cf-4734-a1e6-2ab5c38bf606')
@testtools.skipUnless(CONF.compute_feature_enabled.snapshot,
'Snapshotting is not available.')
def test_list_images_filter_by_server_ref(self):
"""Test listing images filtered by server link href
The list of images should contain only images with the
provided server link href.
"""
server_links = self.server2['links']
# Try all server link types
for link in server_links:
params = {'server': link['href']}
images = self.client.list_images(**params)['images']
self.assertEmpty([i for i in images
if i['id'] == self.snapshot1_id])
self.assertEmpty([i for i in images
if i['id'] == self.snapshot2_id])
self.assertNotEmpty([i for i in images
if i['id'] == self.snapshot3_id])
[docs]
@decorators.idempotent_id('e3356918-4d3e-4756-81d5-abc4524ba29f')
@testtools.skipUnless(CONF.compute_feature_enabled.snapshot,
'Snapshotting is not available.')
def test_list_images_filter_by_type(self):
"""Test listing images filtered by image type
The list of images should contain only images with the
provided image type.
"""
params = {'type': 'snapshot'}
images = self.client.list_images(**params)['images']
self.assertNotEmpty([i for i in images
if i['id'] == self.snapshot1_id])
self.assertNotEmpty([i for i in images
if i['id'] == self.snapshot2_id])
self.assertNotEmpty([i for i in images
if i['id'] == self.snapshot3_id])
self.assertEmpty([i for i in images if i['id'] == self.image_ref])
[docs]
@decorators.idempotent_id('3a484ca9-67ba-451e-b494-7fcf28d32d62')
def test_list_images_limit_results(self):
"""Test listing images with limited count
If we use limit=1 when listing images, then only 1 image should be
returned.
"""
params = {'limit': '1'}
images = self.client.list_images(**params)['images']
self.assertEqual(1, len([x for x in images if 'id' in x]))
[docs]
@decorators.idempotent_id('18bac3ae-da27-436c-92a9-b22474d13aab')
def test_list_images_filter_by_changes_since(self):
"""Test listing images filtered by changes-since
The list of images should contain only images updated since the
provided changes-since value.
"""
# Verify only updated images are returned in the detailed list
# Becoming ACTIVE will modify the updated time
# Filter by the image's created time
params = {'changes-since': self.image3['created']}
images = self.client.list_images(**params)['images']
found = [i for i in images if i['id'] == self.image3_id]
self.assertNotEmpty(found)
[docs]
@decorators.idempotent_id('9b0ea018-6185-4f71-948a-a123a107988e')
def test_list_images_with_detail_filter_by_status(self):
"""Test listing server images details filtered by image status
The list of images should contain only images with the
provided image status.
"""
params = {'status': 'ACTIVE'}
images = self.client.list_images(detail=True, **params)['images']
self.assertNotEmpty([i for i in images if i['id'] == self.image1_id])
self.assertNotEmpty([i for i in images if i['id'] == self.image2_id])
self.assertNotEmpty([i for i in images if i['id'] == self.image3_id])
[docs]
@decorators.idempotent_id('644ea267-9bd9-4f3b-af9f-dffa02396a17')
def test_list_images_with_detail_filter_by_name(self):
"""Test listing server images details filtered by image name
The list of images should contain only images with the
provided image name.
"""
params = {'name': self.image1['name']}
images = self.client.list_images(detail=True, **params)['images']
self.assertNotEmpty([i for i in images if i['id'] == self.image1_id])
self.assertEmpty([i for i in images if i['id'] == self.image2_id])
self.assertEmpty([i for i in images if i['id'] == self.image3_id])
[docs]
@decorators.idempotent_id('ba2fa9a9-b672-47cc-b354-3b4c0600e2cb')
def test_list_images_with_detail_limit_results(self):
"""Test listing images details with limited count
If we use limit=1 when listing images with full details, then only 1
image should be returned.
"""
params = {'limit': '1'}
images = self.client.list_images(detail=True, **params)['images']
self.assertEqual(1, len(images))
[docs]
@decorators.idempotent_id('8c78f822-203b-4bf6-8bba-56ebd551cf84')
@testtools.skipUnless(CONF.compute_feature_enabled.snapshot,
'Snapshotting is not available.')
def test_list_images_with_detail_filter_by_server_ref(self):
"""Test listing images details filtered by server link href
The list of images should contain only images with the
provided server link href.
"""
server_links = self.server2['links']
# Try all server link types
for link in server_links:
params = {'server': link['href']}
images = self.client.list_images(detail=True, **params)['images']
self.assertEmpty([i for i in images
if i['id'] == self.snapshot1_id])
self.assertEmpty([i for i in images
if i['id'] == self.snapshot2_id])
self.assertNotEmpty([i for i in images
if i['id'] == self.snapshot3_id])
[docs]
@decorators.idempotent_id('888c0cc0-7223-43c5-9db0-b125fd0a393b')
@testtools.skipUnless(CONF.compute_feature_enabled.snapshot,
'Snapshotting is not available.')
def test_list_images_with_detail_filter_by_type(self):
"""Test listing images details filtered by image type
The list of images should contain only images with the
provided image type.
"""
params = {'type': 'snapshot'}
images = self.client.list_images(detail=True, **params)['images']
self.client.show_image(self.image_ref)
self.assertNotEmpty([i for i in images
if i['id'] == self.snapshot1_id])
self.assertNotEmpty([i for i in images
if i['id'] == self.snapshot2_id])
self.assertNotEmpty([i for i in images
if i['id'] == self.snapshot3_id])
self.assertEmpty([i for i in images if i['id'] == self.image_ref])
[docs]
@decorators.idempotent_id('7d439e18-ac2e-4827-b049-7e18004712c4')
def test_list_images_with_detail_filter_by_changes_since(self):
"""Test listing images details filtered by changes-since
The list of images should contain only images updated since the
provided changes-since value.
"""
# Becoming ACTIVE will modify the updated time
# Filter by the image's created time
params = {'changes-since': self.image1['created']}
images = self.client.list_images(detail=True, **params)['images']
self.assertNotEmpty([i for i in images if i['id'] == self.image1_id])
