port nova v2 images related tests into nova v3 part2
In nova v3 api, there is no image api. So in nova v3
api tests, we just keep these tests related to create_image
action of the server api. Other tests are moved into glance
testing, also some tests is no longer available, as some
actions are just nova v2 api for glance proxy behaviour.
eg: update the specified item of image properties. In glance
we can only update all values of image properties.
Partially implements blueprint nova-v3-api-tests
Change-Id: I89cec53f30f607b39ff2b4762d35c0f307cc634b
diff --git a/tempest/api/compute/v3/images/test_image_metadata.py b/tempest/api/compute/v3/images/test_image_metadata.py
deleted file mode 100644
index cd4e5e7..0000000
--- a/tempest/api/compute/v3/images/test_image_metadata.py
+++ /dev/null
@@ -1,111 +0,0 @@
-# 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.
-
-from tempest.api.compute import base
-from tempest.common.utils import data_utils
-from tempest import config
-from tempest.test import attr
-
-CONF = config.CONF
-
-
-class ImagesMetadataTest(base.BaseV2ComputeTest):
- _interface = 'json'
-
- @classmethod
- def setUpClass(cls):
- super(ImagesMetadataTest, cls).setUpClass()
- if not CONF.service_available.glance:
- skip_msg = ("%s skipped as glance is not available" % cls.__name__)
- raise cls.skipException(skip_msg)
-
- cls.servers_client = cls.servers_client
- cls.client = cls.images_client
-
- resp, server = cls.create_test_server(wait_until='ACTIVE')
- cls.server_id = server['id']
-
- # Snapshot the server once to save time
- name = data_utils.rand_name('image')
- resp, _ = cls.client.create_image(cls.server_id, name, {})
- cls.image_id = resp['location'].rsplit('/', 1)[1]
-
- cls.client.wait_for_image_status(cls.image_id, 'ACTIVE')
-
- @classmethod
- def tearDownClass(cls):
- cls.client.delete_image(cls.image_id)
- super(ImagesMetadataTest, cls).tearDownClass()
-
- def setUp(self):
- super(ImagesMetadataTest, self).setUp()
- meta = {'key1': 'value1', 'key2': 'value2'}
- resp, _ = self.client.set_image_metadata(self.image_id, meta)
- self.assertEqual(resp.status, 200)
-
- @attr(type='gate')
- def test_list_image_metadata(self):
- # All metadata key/value pairs for an image should be returned
- resp, resp_metadata = self.client.list_image_metadata(self.image_id)
- expected = {'key1': 'value1', 'key2': 'value2'}
- self.assertEqual(expected, resp_metadata)
-
- @attr(type='gate')
- def test_set_image_metadata(self):
- # The metadata for the image should match the new values
- req_metadata = {'meta2': 'value2', 'meta3': 'value3'}
- resp, body = self.client.set_image_metadata(self.image_id,
- req_metadata)
-
- resp, resp_metadata = self.client.list_image_metadata(self.image_id)
- self.assertEqual(req_metadata, resp_metadata)
-
- @attr(type='gate')
- def test_update_image_metadata(self):
- # The metadata for the image should match the updated values
- req_metadata = {'key1': 'alt1', 'key3': 'value3'}
- resp, metadata = self.client.update_image_metadata(self.image_id,
- req_metadata)
-
- resp, resp_metadata = self.client.list_image_metadata(self.image_id)
- expected = {'key1': 'alt1', 'key2': 'value2', 'key3': 'value3'}
- self.assertEqual(expected, resp_metadata)
-
- @attr(type='gate')
- def test_get_image_metadata_item(self):
- # The value for a specific metadata key should be returned
- resp, meta = self.client.get_image_metadata_item(self.image_id,
- 'key2')
- self.assertEqual('value2', meta['key2'])
-
- @attr(type='gate')
- def test_set_image_metadata_item(self):
- # The value provided for the given meta item should be set for
- # the image
- meta = {'key1': 'alt'}
- resp, body = self.client.set_image_metadata_item(self.image_id,
- 'key1', meta)
- resp, resp_metadata = self.client.list_image_metadata(self.image_id)
- expected = {'key1': 'alt', 'key2': 'value2'}
- self.assertEqual(expected, resp_metadata)
-
- @attr(type='gate')
- def test_delete_image_metadata_item(self):
- # The metadata value/key pair should be deleted from the image
- resp, body = self.client.delete_image_metadata_item(self.image_id,
- 'key1')
- resp, resp_metadata = self.client.list_image_metadata(self.image_id)
- expected = {'key2': 'value2'}
- self.assertEqual(expected, resp_metadata)
diff --git a/tempest/api/compute/v3/images/test_image_metadata_negative.py b/tempest/api/compute/v3/images/test_image_metadata_negative.py
deleted file mode 100644
index e76af2c..0000000
--- a/tempest/api/compute/v3/images/test_image_metadata_negative.py
+++ /dev/null
@@ -1,75 +0,0 @@
-# Copyright 2013 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.
-
-from tempest.api.compute import base
-from tempest.common.utils import data_utils
-from tempest import exceptions
-from tempest.test import attr
-
-
-class ImagesMetadataTest(base.BaseV2ComputeTest):
- _interface = 'json'
-
- @classmethod
- def setUpClass(cls):
- super(ImagesMetadataTest, cls).setUpClass()
- cls.client = cls.images_client
-
- @attr(type=['negative', 'gate'])
- def test_list_nonexistent_image_metadata(self):
- # Negative test: List on nonexistent image
- # metadata should not happen
- self.assertRaises(exceptions.NotFound, self.client.list_image_metadata,
- data_utils.rand_uuid())
-
- @attr(type=['negative', 'gate'])
- def test_update_nonexistent_image_metadata(self):
- # Negative test:An update should not happen for a non-existent image
- meta = {'key1': 'alt1', 'key2': 'alt2'}
- self.assertRaises(exceptions.NotFound,
- self.client.update_image_metadata,
- data_utils.rand_uuid(), meta)
-
- @attr(type=['negative', 'gate'])
- def test_get_nonexistent_image_metadata_item(self):
- # Negative test: Get on non-existent image should not happen
- self.assertRaises(exceptions.NotFound,
- self.client.get_image_metadata_item,
- data_utils.rand_uuid(), 'key2')
-
- @attr(type=['negative', 'gate'])
- def test_set_nonexistent_image_metadata(self):
- # Negative test: Metadata should not be set to a non-existent image
- meta = {'key1': 'alt1', 'key2': 'alt2'}
- self.assertRaises(exceptions.NotFound, self.client.set_image_metadata,
- data_utils.rand_uuid(), meta)
-
- @attr(type=['negative', 'gate'])
- def test_set_nonexistent_image_metadata_item(self):
- # Negative test: Metadata item should not be set to a
- # nonexistent image
- meta = {'key1': 'alt'}
- self.assertRaises(exceptions.NotFound,
- self.client.set_image_metadata_item,
- data_utils.rand_uuid(), 'key1',
- meta)
-
- @attr(type=['negative', 'gate'])
- def test_delete_nonexistent_image_metadata_item(self):
- # Negative test: Shouldn't be able to delete metadata
- # item from non-existent image
- self.assertRaises(exceptions.NotFound,
- self.client.delete_image_metadata_item,
- data_utils.rand_uuid(), 'key1')
diff --git a/tempest/api/compute/v3/images/test_images_oneserver.py b/tempest/api/compute/v3/images/test_images_oneserver.py
index 18772df..cd40948 100644
--- a/tempest/api/compute/v3/images/test_images_oneserver.py
+++ b/tempest/api/compute/v3/images/test_images_oneserver.py
@@ -20,13 +20,13 @@
from tempest.common.utils import data_utils
from tempest import config
from tempest.openstack.common import log as logging
-from tempest.test import attr
+from tempest import test
CONF = config.CONF
LOG = logging.getLogger(__name__)
-class ImagesOneServerTest(base.BaseV2ComputeTest):
+class ImagesOneServerV3Test(base.BaseV3ComputeTest):
_interface = 'json'
def tearDown(self):
@@ -34,12 +34,12 @@
for image_id in self.image_ids:
self.client.delete_image(image_id)
self.image_ids.remove(image_id)
- super(ImagesOneServerTest, self).tearDown()
+ super(ImagesOneServerV3Test, self).tearDown()
def setUp(self):
# NOTE(afazekas): Normally we use the same server with all test cases,
# but if it has an issue, we build a new one
- super(ImagesOneServerTest, self).setUp()
+ super(ImagesOneServerV3Test, self).setUp()
# Check if the server is in a clean state after test
try:
self.servers_client.wait_for_server_status(self.server_id,
@@ -53,7 +53,7 @@
@classmethod
def setUpClass(cls):
- super(ImagesOneServerTest, cls).setUpClass()
+ super(ImagesOneServerV3Test, cls).setUpClass()
cls.client = cls.images_client
if not CONF.service_available.glance:
skip_msg = ("%s skipped as glance is not available" % cls.__name__)
@@ -86,13 +86,14 @@
@testtools.skipUnless(CONF.compute_feature_enabled.create_image,
'Environment unable to create images.')
- @attr(type='smoke')
+ @test.attr(type='smoke')
def test_create_delete_image(self):
# Create a new image
name = data_utils.rand_name('image')
meta = {'image_type': 'test'}
- resp, body = self.client.create_image(self.server_id, name, meta)
+ resp, body = self.servers_client.create_image(self.server_id,
+ name, meta)
self.assertEqual(202, resp.status)
image_id = data_utils.parse_image_id(resp['location'])
self.client.wait_for_image_status(image_id, 'ACTIVE')
@@ -117,12 +118,13 @@
self.assertEqual('204', resp['status'])
self.client.wait_for_resource_deletion(image_id)
- @attr(type=['gate'])
+ @test.attr(type=['gate'])
def test_create_image_specify_multibyte_character_image_name(self):
# prefix character is:
# http://www.fileformat.info/info/unicode/char/1F4A9/index.htm
utf8_name = data_utils.rand_name(u'\xF0\x9F\x92\xA9')
- resp, body = self.client.create_image(self.server_id, utf8_name)
+ resp, body = self.servers_client.create_image(self.server_id,
+ utf8_name)
image_id = data_utils.parse_image_id(resp['location'])
self.addCleanup(self.client.delete_image, image_id)
self.assertEqual('202', resp['status'])
diff --git a/tempest/api/compute/v3/images/test_images_oneserver_negative.py b/tempest/api/compute/v3/images/test_images_oneserver_negative.py
index bc276d1..f2f2375 100644
--- a/tempest/api/compute/v3/images/test_images_oneserver_negative.py
+++ b/tempest/api/compute/v3/images/test_images_oneserver_negative.py
@@ -20,15 +20,14 @@
from tempest import config
from tempest import exceptions
from tempest.openstack.common import log as logging
-from tempest.test import attr
-from tempest.test import skip_because
+from tempest import test
CONF = config.CONF
LOG = logging.getLogger(__name__)
-class ImagesOneServerNegativeTest(base.BaseV2ComputeTest):
+class ImagesOneServerNegativeV3Test(base.BaseV3ComputeTest):
_interface = 'json'
def tearDown(self):
@@ -36,12 +35,12 @@
for image_id in self.image_ids:
self.client.delete_image(image_id)
self.image_ids.remove(image_id)
- super(ImagesOneServerNegativeTest, self).tearDown()
+ super(ImagesOneServerNegativeV3Test, self).tearDown()
def setUp(self):
# NOTE(afazekas): Normally we use the same server with all test cases,
# but if it has an issue, we build a new one
- super(ImagesOneServerNegativeTest, self).setUp()
+ super(ImagesOneServerNegativeV3Test, self).setUp()
# Check if the server is in a clean state after test
try:
self.servers_client.wait_for_server_status(self.server_id,
@@ -58,7 +57,7 @@
@classmethod
def setUpClass(cls):
- super(ImagesOneServerNegativeTest, cls).setUpClass()
+ super(ImagesOneServerNegativeV3Test, cls).setUpClass()
cls.client = cls.images_client
if not CONF.service_available.glance:
skip_msg = ("%s skipped as glance is not available" % cls.__name__)
@@ -85,41 +84,44 @@
cls.alt_manager = clients.AltManager()
cls.alt_client = cls.alt_manager.images_client
- @skip_because(bug="1006725")
- @attr(type=['negative', 'gate'])
+ @test.skip_because(bug="1006725")
+ @test.attr(type=['negative', 'gate'])
def test_create_image_specify_multibyte_character_image_name(self):
# invalid multibyte sequence from:
# http://stackoverflow.com/questions/1301402/
# example-invalid-utf8-string
invalid_name = data_utils.rand_name(u'\xc3\x28')
self.assertRaises(exceptions.BadRequest,
- self.client.create_image, self.server_id,
+ self.servers_client.create_image,
+ self.server_id,
invalid_name)
- @attr(type=['negative', 'gate'])
+ @test.attr(type=['negative', 'gate'])
def test_create_image_specify_invalid_metadata(self):
# Return an error when creating image with invalid metadata
snapshot_name = data_utils.rand_name('test-snap-')
meta = {'': ''}
- self.assertRaises(exceptions.BadRequest, self.client.create_image,
+ self.assertRaises(exceptions.BadRequest,
+ self.servers_client.create_image,
self.server_id, snapshot_name, meta)
- @attr(type=['negative', 'gate'])
+ @test.attr(type=['negative', 'gate'])
def test_create_image_specify_metadata_over_limits(self):
# Return an error when creating image with meta data over 256 chars
snapshot_name = data_utils.rand_name('test-snap-')
meta = {'a' * 260: 'b' * 260}
- self.assertRaises(exceptions.BadRequest, self.client.create_image,
+ self.assertRaises(exceptions.BadRequest,
+ self.servers_client.create_image,
self.server_id, snapshot_name, meta)
- @attr(type=['negative', 'gate'])
+ @test.attr(type=['negative', 'gate'])
def test_create_second_image_when_first_image_is_being_saved(self):
# Disallow creating another image when first image is being saved
# Create first snapshot
snapshot_name = data_utils.rand_name('test-snap-')
- resp, body = self.client.create_image(self.server_id,
- snapshot_name)
+ resp, body = self.servers_client.create_image(self.server_id,
+ snapshot_name)
self.assertEqual(202, resp.status)
image_id = data_utils.parse_image_id(resp['location'])
self.image_ids.append(image_id)
@@ -127,23 +129,26 @@
# Create second snapshot
alt_snapshot_name = data_utils.rand_name('test-snap-')
- self.assertRaises(exceptions.Conflict, self.client.create_image,
+ self.assertRaises(exceptions.Conflict,
+ self.servers_client.create_image,
self.server_id, alt_snapshot_name)
- @attr(type=['negative', 'gate'])
+ @test.attr(type=['negative', 'gate'])
def test_create_image_specify_name_over_256_chars(self):
# Return an error if snapshot name over 256 characters is passed
snapshot_name = data_utils.rand_name('a' * 260)
- self.assertRaises(exceptions.BadRequest, self.client.create_image,
+ self.assertRaises(exceptions.BadRequest,
+ self.servers_client.create_image,
self.server_id, snapshot_name)
- @attr(type=['negative', 'gate'])
+ @test.attr(type=['negative', 'gate'])
def test_delete_image_that_is_not_yet_active(self):
# Return an error while trying to delete an image what is creating
snapshot_name = data_utils.rand_name('test-snap-')
- resp, body = self.client.create_image(self.server_id, snapshot_name)
+ resp, body = self.servers_client.create_image(self.server_id,
+ snapshot_name)
self.assertEqual(202, resp.status)
image_id = data_utils.parse_image_id(resp['location'])
self.image_ids.append(image_id)
@@ -151,7 +156,7 @@
# Do not wait, attempt to delete the image, ensure it's successful
resp, body = self.client.delete_image(image_id)
- self.assertEqual('204', resp['status'])
+ self.assertEqual('200', resp['status'])
self.image_ids.remove(image_id)
self.assertRaises(exceptions.NotFound, self.client.get_image, image_id)
diff --git a/tempest/api/compute/v3/images/test_list_image_filters.py b/tempest/api/compute/v3/images/test_list_image_filters.py
deleted file mode 100644
index 457ca53..0000000
--- a/tempest/api/compute/v3/images/test_list_image_filters.py
+++ /dev/null
@@ -1,225 +0,0 @@
-# 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.
-
-from tempest.api.compute import base
-from tempest import config
-from tempest import exceptions
-from tempest.openstack.common import log as logging
-from tempest.test import attr
-
-CONF = config.CONF
-
-LOG = logging.getLogger(__name__)
-
-
-class ListImageFiltersTest(base.BaseV2ComputeTest):
- _interface = 'json'
-
- @classmethod
- def setUpClass(cls):
- super(ListImageFiltersTest, cls).setUpClass()
- if not CONF.service_available.glance:
- skip_msg = ("%s skipped as glance is not available" % cls.__name__)
- raise cls.skipException(skip_msg)
- cls.client = cls.images_client
- cls.image_ids = []
-
- try:
- resp, cls.server1 = cls.create_test_server()
- resp, cls.server2 = cls.create_test_server(wait_until='ACTIVE')
- # NOTE(sdague) this is faster than doing the sync wait_util on both
- cls.servers_client.wait_for_server_status(cls.server1['id'],
- 'ACTIVE')
-
- # Create images to be used in the filter tests
- resp, cls.image1 = cls.create_image_from_server(
- cls.server1['id'], wait_until='ACTIVE')
- cls.image1_id = cls.image1['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
- resp, cls.image3 = cls.create_image_from_server(
- cls.server2['id'], wait_until='ACTIVE')
- cls.image3_id = cls.image3['id']
-
- # Wait for the server to be active after the image upload
- resp, cls.image2 = cls.create_image_from_server(
- cls.server1['id'], wait_until='ACTIVE')
- cls.image2_id = cls.image2['id']
- except Exception:
- LOG.exception('setUpClass failed')
- cls.tearDownClass()
- raise
-
- @attr(type=['negative', 'gate'])
- def test_get_image_not_existing(self):
- # Check raises a NotFound
- self.assertRaises(exceptions.NotFound, self.client.get_image,
- "nonexistingimageid")
-
- @attr(type='gate')
- def test_list_images_filter_by_status(self):
- # The list of images should contain only images with the
- # provided status
- params = {'status': 'ACTIVE'}
- resp, images = self.client.list_images(params)
-
- self.assertTrue(any([i for i in images if i['id'] == self.image1_id]))
- self.assertTrue(any([i for i in images if i['id'] == self.image2_id]))
- self.assertTrue(any([i for i in images if i['id'] == self.image3_id]))
-
- @attr(type='gate')
- def test_list_images_filter_by_name(self):
- # List of all images should contain the expected images filtered
- # by name
- params = {'name': self.image1['name']}
- resp, images = self.client.list_images(params)
-
- self.assertTrue(any([i for i in images if i['id'] == self.image1_id]))
- self.assertFalse(any([i for i in images if i['id'] == self.image2_id]))
- self.assertFalse(any([i for i in images if i['id'] == self.image3_id]))
-
- @attr(type='gate')
- def test_list_images_filter_by_server_id(self):
- # The images should contain images filtered by server id
- params = {'server': self.server1['id']}
- resp, images = self.client.list_images(params)
-
- self.assertTrue(any([i for i in images if i['id'] == self.image1_id]),
- "Failed to find image %s in images. Got images %s" %
- (self.image1_id, images))
- self.assertTrue(any([i for i in images if i['id'] == self.image2_id]))
- self.assertFalse(any([i for i in images if i['id'] == self.image3_id]))
-
- @attr(type='gate')
- def test_list_images_filter_by_server_ref(self):
- # The list of servers should be filtered by server ref
- server_links = self.server2['links']
-
- # Try all server link types
- for link in server_links:
- params = {'server': link['href']}
- resp, images = self.client.list_images(params)
-
- self.assertFalse(any([i for i in images
- if i['id'] == self.image1_id]))
- self.assertFalse(any([i for i in images
- if i['id'] == self.image2_id]))
- self.assertTrue(any([i for i in images
- if i['id'] == self.image3_id]))
-
- @attr(type='gate')
- def test_list_images_filter_by_type(self):
- # The list of servers should be filtered by image type
- params = {'type': 'snapshot'}
- resp, images = self.client.list_images(params)
-
- self.assertTrue(any([i for i in images if i['id'] == self.image1_id]))
- self.assertTrue(any([i for i in images if i['id'] == self.image2_id]))
- self.assertTrue(any([i for i in images if i['id'] == self.image3_id]))
- self.assertFalse(any([i for i in images if i['id'] == self.image_ref]))
-
- @attr(type='gate')
- def test_list_images_limit_results(self):
- # Verify only the expected number of results are returned
- params = {'limit': '1'}
- resp, images = self.client.list_images(params)
- self.assertEqual(1, len([x for x in images if 'id' in x]))
-
- @attr(type='gate')
- def test_list_images_filter_by_changes_since(self):
- # 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']}
- resp, images = self.client.list_images(params)
- found = any([i for i in images if i['id'] == self.image3_id])
- self.assertTrue(found)
-
- @attr(type='gate')
- def test_list_images_with_detail_filter_by_status(self):
- # Detailed list of all images should only contain images
- # with the provided status
- params = {'status': 'ACTIVE'}
- resp, images = self.client.list_images_with_detail(params)
-
- self.assertTrue(any([i for i in images if i['id'] == self.image1_id]))
- self.assertTrue(any([i for i in images if i['id'] == self.image2_id]))
- self.assertTrue(any([i for i in images if i['id'] == self.image3_id]))
-
- @attr(type='gate')
- def test_list_images_with_detail_filter_by_name(self):
- # Detailed list of all images should contain the expected
- # images filtered by name
- params = {'name': self.image1['name']}
- resp, images = self.client.list_images_with_detail(params)
-
- self.assertTrue(any([i for i in images if i['id'] == self.image1_id]))
- self.assertFalse(any([i for i in images if i['id'] == self.image2_id]))
- self.assertFalse(any([i for i in images if i['id'] == self.image3_id]))
-
- @attr(type='gate')
- def test_list_images_with_detail_limit_results(self):
- # Verify only the expected number of results (with full details)
- # are returned
- params = {'limit': '1'}
- resp, images = self.client.list_images_with_detail(params)
- self.assertEqual(1, len(images))
-
- @attr(type='gate')
- def test_list_images_with_detail_filter_by_server_ref(self):
- # Detailed list of servers should be filtered by server ref
- server_links = self.server2['links']
-
- # Try all server link types
- for link in server_links:
- params = {'server': link['href']}
- resp, images = self.client.list_images_with_detail(params)
-
- self.assertFalse(any([i for i in images
- if i['id'] == self.image1_id]))
- self.assertFalse(any([i for i in images
- if i['id'] == self.image2_id]))
- self.assertTrue(any([i for i in images
- if i['id'] == self.image3_id]))
-
- @attr(type='gate')
- def test_list_images_with_detail_filter_by_type(self):
- # The detailed list of servers should be filtered by image type
- params = {'type': 'snapshot'}
- resp, images = self.client.list_images_with_detail(params)
- resp, image4 = self.client.get_image(self.image_ref)
-
- self.assertTrue(any([i for i in images if i['id'] == self.image1_id]))
- self.assertTrue(any([i for i in images if i['id'] == self.image2_id]))
- self.assertTrue(any([i for i in images if i['id'] == self.image3_id]))
- self.assertFalse(any([i for i in images if i['id'] == self.image_ref]))
-
- @attr(type='gate')
- def test_list_images_with_detail_filter_by_changes_since(self):
- # Verify an update image is returned
-
- # Becoming ACTIVE will modify the updated time
- # Filter by the image's created time
- params = {'changes-since': self.image1['created']}
- resp, images = self.client.list_images_with_detail(params)
- self.assertTrue(any([i for i in images if i['id'] == self.image1_id]))
-
- @attr(type=['negative', 'gate'])
- def test_get_nonexistent_image(self):
- # Negative test: GET on non-existent image should fail
- self.assertRaises(exceptions.NotFound, self.client.get_image, 999)
diff --git a/tempest/api/image/v1/test_images.py b/tempest/api/image/v1/test_images.py
index d8b79ca..517123d 100644
--- a/tempest/api/image/v1/test_images.py
+++ b/tempest/api/image/v1/test_images.py
@@ -16,8 +16,9 @@
import cStringIO as StringIO
from tempest.api.image import base
+from tempest.common.utils import data_utils
from tempest import config
-from tempest.test import attr
+from tempest import test
CONF = config.CONF
@@ -25,7 +26,7 @@
class CreateRegisterImagesTest(base.BaseV1ImageTest):
"""Here we test the registration and creation of images."""
- @attr(type='gate')
+ @test.attr(type='gate')
def test_register_then_upload(self):
# Register, then upload an image
properties = {'prop1': 'val1'}
@@ -48,7 +49,7 @@
self.assertIn('size', body)
self.assertEqual(1024, body.get('size'))
- @attr(type='gate')
+ @test.attr(type='gate')
def test_register_remote_image(self):
# Register a new remote image
resp, body = self.create_image(name='New Remote Image',
@@ -66,7 +67,7 @@
self.assertEqual(properties['key1'], 'value1')
self.assertEqual(properties['key2'], 'value2')
- @attr(type='gate')
+ @test.attr(type='gate')
def test_register_http_image(self):
resp, body = self.create_image(name='New Http Image',
container_format='bare',
@@ -80,7 +81,7 @@
resp, body = self.client.get_image(image_id)
self.assertEqual(resp['status'], '200')
- @attr(type='gate')
+ @test.attr(type='gate')
def test_register_image_with_min_ram(self):
# Register an image with min ram
properties = {'prop1': 'val1'}
@@ -110,7 +111,6 @@
@classmethod
def setUpClass(cls):
super(ListImagesTest, cls).setUpClass()
-
# We add a few images here to test the listing functionality of
# the images API
img1 = cls._create_remote_image('one', 'bare', 'raw')
@@ -131,7 +131,7 @@
# 1x with size 42
cls.size42_set = set((img5,))
# 3x with size 142
- cls.size142_set = set((img6, img7, img8))
+ cls.size142_set = set((img6, img7, img8,))
# dup named
cls.dup_set = set((img3, img4))
@@ -168,7 +168,7 @@
image_id = image['id']
return image_id
- @attr(type='gate')
+ @test.attr(type='gate')
def test_index_no_params(self):
# Simple test to see all fixture images returned
resp, images_list = self.client.image_list()
@@ -177,7 +177,7 @@
for image_id in self.created_images:
self.assertIn(image_id, image_list)
- @attr(type='gate')
+ @test.attr(type='gate')
def test_index_disk_format(self):
resp, images_list = self.client.image_list(disk_format='ami')
self.assertEqual(resp['status'], '200')
@@ -187,7 +187,7 @@
self.assertTrue(self.ami_set <= result_set)
self.assertFalse(self.created_set - self.ami_set <= result_set)
- @attr(type='gate')
+ @test.attr(type='gate')
def test_index_container_format(self):
resp, images_list = self.client.image_list(container_format='bare')
self.assertEqual(resp['status'], '200')
@@ -197,7 +197,7 @@
self.assertTrue(self.bare_set <= result_set)
self.assertFalse(self.created_set - self.bare_set <= result_set)
- @attr(type='gate')
+ @test.attr(type='gate')
def test_index_max_size(self):
resp, images_list = self.client.image_list(size_max=42)
self.assertEqual(resp['status'], '200')
@@ -207,7 +207,7 @@
self.assertTrue(self.size42_set <= result_set)
self.assertFalse(self.created_set - self.size42_set <= result_set)
- @attr(type='gate')
+ @test.attr(type='gate')
def test_index_min_size(self):
resp, images_list = self.client.image_list(size_min=142)
self.assertEqual(resp['status'], '200')
@@ -217,7 +217,7 @@
self.assertTrue(self.size142_set <= result_set)
self.assertFalse(self.size42_set <= result_set)
- @attr(type='gate')
+ @test.attr(type='gate')
def test_index_status_active_detail(self):
resp, images_list = self.client.image_list_detail(status='active',
sort_key='size',
@@ -230,7 +230,7 @@
top_size = size
self.assertEqual(image['status'], 'active')
- @attr(type='gate')
+ @test.attr(type='gate')
def test_index_name(self):
resp, images_list = self.client.image_list_detail(
name='New Remote Image dup')
@@ -240,3 +240,139 @@
self.assertEqual(image['name'], 'New Remote Image dup')
self.assertTrue(self.dup_set <= result_set)
self.assertFalse(self.created_set - self.dup_set <= result_set)
+
+
+class ListSnapshotImagesTest(base.BaseV1ImageTest):
+ @classmethod
+ def setUpClass(cls):
+ super(ListSnapshotImagesTest, cls).setUpClass()
+ if not CONF.compute_feature_enabled.api_v3:
+ cls.servers_client = cls.os.servers_client
+ else:
+ cls.servers_client = cls.os.servers_v3_client
+ cls.servers = []
+ # We add a few images here to test the listing functionality of
+ # the images API
+ cls.snapshot = cls._create_snapshot(
+ 'snapshot', CONF.compute.image_ref,
+ CONF.compute.flavor_ref)
+ cls.snapshot_set = set((cls.snapshot,))
+
+ image_file = StringIO.StringIO('*' * 42)
+ resp, image = cls.create_image(name="Standard Image",
+ container_format='ami',
+ disk_format='ami',
+ is_public=True, data=image_file)
+ cls.image_id = image['id']
+ cls.client.wait_for_image_status(image['id'], 'active')
+
+ @classmethod
+ def tearDownClass(cls):
+ for server in cls.servers:
+ cls.servers_client.delete_server(server['id'])
+ super(ListSnapshotImagesTest, cls).tearDownClass()
+
+ @classmethod
+ def _create_snapshot(cls, name, image_id, flavor, **kwargs):
+ resp, server = cls.servers_client.create_server(
+ name, image_id, flavor, **kwargs)
+ cls.servers.append(server)
+ cls.servers_client.wait_for_server_status(
+ server['id'], 'ACTIVE')
+ resp, image = cls.servers_client.create_image(
+ server['id'], name)
+ image_id = data_utils.parse_image_id(resp['location'])
+ cls.created_images.append(image_id)
+ cls.client.wait_for_image_status(image_id,
+ 'active')
+ return image_id
+
+ @test.attr(type='gate')
+ def test_index_server_id(self):
+ # The images should contain images filtered by server id
+ resp, images = self.client.image_list_detail(
+ {'instance_uuid': self.servers[0]['id']})
+ self.assertEqual(200, resp.status)
+ result_set = set(map(lambda x: x['id'], images))
+ self.assertEqual(self.snapshot_set, result_set)
+
+ @test.attr(type='gate')
+ def test_index_type(self):
+ # The list of servers should be filtered by image type
+ params = {'image_type': 'snapshot'}
+ resp, images = self.client.image_list_detail(params)
+
+ self.assertEqual(200, resp.status)
+ result_set = set(map(lambda x: x['id'], images))
+ self.assertIn(self.snapshot, result_set)
+
+ @test.attr(type='gate')
+ def test_index_limit(self):
+ # Verify only the expected number of results are returned
+ resp, images = self.client.image_list_detail(limit=1)
+
+ self.assertEqual(200, resp.status)
+ self.assertEqual(1, len(images))
+
+ @test.attr(type='gate')
+ def test_index_by_change_since(self):
+ # Verify an update image is returned
+ # Becoming ACTIVE will modify the updated time
+ # Filter by the image's created time
+ resp, image = self.client.get_image_meta(self.snapshot)
+ self.assertEqual(200, resp.status)
+ self.assertEqual(self.snapshot, image['id'])
+ resp, images = self.client.image_list_detail(
+ changes_since=image['updated_at'])
+
+ self.assertEqual(200, resp.status)
+ result_set = set(map(lambda x: x['id'], images))
+ self.assertIn(self.image_id, result_set)
+ self.assertNotIn(self.snapshot, result_set)
+
+
+class UpdateImageMetaTest(base.BaseV1ImageTest):
+ @classmethod
+ def setUpClass(cls):
+ super(UpdateImageMetaTest, cls).setUpClass()
+ cls.image_id = cls._create_standard_image('1', 'ami', 'ami', 42)
+
+ @classmethod
+ def _create_standard_image(cls, name, container_format,
+ disk_format, size):
+ """
+ Create a new standard image and return the ID of the newly-registered
+ image. Note that the size of the new image is a random number between
+ 1024 and 4096
+ """
+ image_file = StringIO.StringIO('*' * size)
+ name = 'New Standard Image %s' % name
+ resp, image = cls.create_image(name=name,
+ container_format=container_format,
+ disk_format=disk_format,
+ is_public=True, data=image_file,
+ properties={'key1': 'value1'})
+ image_id = image['id']
+ return image_id
+
+ @test.attr(type='gate')
+ def test_list_image_metadata(self):
+ # All metadata key/value pairs for an image should be returned
+ resp, resp_metadata = self.client.get_image_meta(self.image_id)
+ expected = {'key1': 'value1'}
+ self.assertEqual(expected, resp_metadata['properties'])
+
+ @test.attr(type='gate')
+ def test_update_image_metadata(self):
+ # The metadata for the image should match the updated values
+ req_metadata = {'key1': 'alt1', 'key2': 'value2'}
+ resp, metadata = self.client.get_image_meta(self.image_id)
+ self.assertEqual(200, resp.status)
+ self.assertEqual(metadata['properties'], {'key1': 'value1'})
+ metadata['properties'].update(req_metadata)
+ resp, metadata = self.client.update_image(
+ self.image_id, properties=metadata['properties'])
+
+ resp, resp_metadata = self.client.get_image_meta(self.image_id)
+ expected = {'key1': 'alt1', 'key2': 'value2'}
+ self.assertEqual(expected, resp_metadata['properties'])
diff --git a/tempest/services/image/v1/json/image_client.py b/tempest/services/image/v1/json/image_client.py
index bc9db38..3724ac6 100644
--- a/tempest/services/image/v1/json/image_client.py
+++ b/tempest/services/image/v1/json/image_client.py
@@ -157,7 +157,7 @@
return resp, body['image']
def update_image(self, image_id, name=None, container_format=None,
- data=None):
+ data=None, properties=None):
params = {}
headers = {}
if name is not None:
@@ -166,6 +166,9 @@
if container_format is not None:
params['container_format'] = container_format
+ if properties is not None:
+ params['properties'] = properties
+
headers.update(self._image_meta_to_headers(params))
if data is not None:
@@ -190,7 +193,8 @@
body = json.loads(body)
return resp, body['images']
- def image_list_detail(self, properties=dict(), **kwargs):
+ def image_list_detail(self, properties=dict(), changes_since=None,
+ **kwargs):
url = 'v1/images/detail'
params = {}
@@ -199,6 +203,9 @@
kwargs.update(params)
+ if changes_since is not None:
+ kwargs['changes-since'] = changes_since
+
if len(kwargs) > 0:
url += '?%s' % urllib.urlencode(kwargs)