Merge "port nova v2 images related tests into nova v3 part1"
diff --git a/tempest/api/compute/v3/images/test_image_metadata.py b/tempest/api/compute/v3/images/test_image_metadata.py
new file mode 100644
index 0000000..76e0cae
--- /dev/null
+++ b/tempest/api/compute/v3/images/test_image_metadata.py
@@ -0,0 +1,114 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# 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.test import attr
+
+
+class ImagesMetadataTestJSON(base.BaseV2ComputeTest):
+ _interface = 'json'
+
+ @classmethod
+ def setUpClass(cls):
+ super(ImagesMetadataTestJSON, cls).setUpClass()
+ if not cls.config.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(ImagesMetadataTestJSON, cls).tearDownClass()
+
+ def setUp(self):
+ super(ImagesMetadataTestJSON, 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)
+
+
+class ImagesMetadataTestXML(ImagesMetadataTestJSON):
+ _interface = 'xml'
diff --git a/tempest/api/compute/v3/images/test_image_metadata_negative.py b/tempest/api/compute/v3/images/test_image_metadata_negative.py
new file mode 100644
index 0000000..1767e5d
--- /dev/null
+++ b/tempest/api/compute/v3/images/test_image_metadata_negative.py
@@ -0,0 +1,81 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# 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 ImagesMetadataTestJSON(base.BaseV2ComputeTest):
+ _interface = 'json'
+
+ @classmethod
+ def setUpClass(cls):
+ super(ImagesMetadataTestJSON, 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')
+
+
+class ImagesMetadataTestXML(ImagesMetadataTestJSON):
+ _interface = 'xml'
diff --git a/tempest/api/compute/v3/images/test_images_oneserver.py b/tempest/api/compute/v3/images/test_images_oneserver.py
new file mode 100644
index 0000000..26cc3f6
--- /dev/null
+++ b/tempest/api/compute/v3/images/test_images_oneserver.py
@@ -0,0 +1,138 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# 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 testtools
+
+from tempest.api.compute import base
+from tempest import clients
+from tempest.common.utils import data_utils
+from tempest import config
+from tempest.openstack.common import log as logging
+from tempest.test import attr
+
+CONF = config.CONF
+LOG = logging.getLogger(__name__)
+
+
+class ImagesOneServerTestJSON(base.BaseV2ComputeTest):
+ _interface = 'json'
+
+ def tearDown(self):
+ """Terminate test instances created after a test is executed."""
+ for image_id in self.image_ids:
+ self.client.delete_image(image_id)
+ self.image_ids.remove(image_id)
+ super(ImagesOneServerTestJSON, 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(ImagesOneServerTestJSON, self).setUp()
+ # Check if the server is in a clean state after test
+ try:
+ self.servers_client.wait_for_server_status(self.server_id,
+ 'ACTIVE')
+ except Exception:
+ LOG.exception('server %s timed out to become ACTIVE. rebuilding'
+ % self.server_id)
+ # Rebuild server if cannot reach the ACTIVE state
+ # Usually it means the server had a serious accident
+ self.__class__.server_id = self.rebuild_server(self.server_id)
+
+ @classmethod
+ def setUpClass(cls):
+ super(ImagesOneServerTestJSON, cls).setUpClass()
+ cls.client = cls.images_client
+ if not cls.config.service_available.glance:
+ skip_msg = ("%s skipped as glance is not available" % cls.__name__)
+ raise cls.skipException(skip_msg)
+
+ try:
+ resp, server = cls.create_test_server(wait_until='ACTIVE')
+ cls.server_id = server['id']
+ except Exception:
+ cls.tearDownClass()
+ raise
+
+ cls.image_ids = []
+
+ if cls.multi_user:
+ if cls.config.compute.allow_tenant_isolation:
+ creds = cls.isolated_creds.get_alt_creds()
+ username, tenant_name, password = creds
+ cls.alt_manager = clients.Manager(username=username,
+ password=password,
+ tenant_name=tenant_name)
+ else:
+ # Use the alt_XXX credentials in the config file
+ cls.alt_manager = clients.AltManager()
+ cls.alt_client = cls.alt_manager.images_client
+
+ def _get_default_flavor_disk_size(self, flavor_id):
+ resp, flavor = self.flavors_client.get_flavor_details(flavor_id)
+ return flavor['disk']
+
+ @testtools.skipUnless(CONF.compute_feature_enabled.create_image,
+ 'Environment unable to create images.')
+ @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)
+ self.assertEqual(202, resp.status)
+ image_id = data_utils.parse_image_id(resp['location'])
+ self.client.wait_for_image_status(image_id, 'ACTIVE')
+
+ # Verify the image was created correctly
+ resp, image = self.client.get_image(image_id)
+ self.assertEqual(name, image['name'])
+ self.assertEqual('test', image['metadata']['image_type'])
+
+ resp, original_image = self.client.get_image(self.image_ref)
+
+ # Verify minRAM is the same as the original image
+ self.assertEqual(image['minRam'], original_image['minRam'])
+
+ # Verify minDisk is the same as the original image or the flavor size
+ flavor_disk_size = self._get_default_flavor_disk_size(self.flavor_ref)
+ self.assertIn(str(image['minDisk']),
+ (str(original_image['minDisk']), str(flavor_disk_size)))
+
+ # Verify the image was deleted correctly
+ resp, body = self.client.delete_image(image_id)
+ self.assertEqual('204', resp['status'])
+ self.client.wait_for_resource_deletion(image_id)
+
+ @attr(type=['gate'])
+ def test_create_image_specify_multibyte_character_image_name(self):
+ if self.__class__._interface == "xml":
+ # NOTE(sdague): not entirely accurage, but we'd need a ton of work
+ # in our XML client to make this good
+ raise self.skipException("Not testable in XML")
+ # 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)
+ image_id = data_utils.parse_image_id(resp['location'])
+ self.addCleanup(self.client.delete_image, image_id)
+ self.assertEqual('202', resp['status'])
+
+
+class ImagesOneServerTestXML(ImagesOneServerTestJSON):
+ _interface = 'xml'
diff --git a/tempest/api/compute/v3/images/test_images_oneserver_negative.py b/tempest/api/compute/v3/images/test_images_oneserver_negative.py
new file mode 100644
index 0000000..5e235d1
--- /dev/null
+++ b/tempest/api/compute/v3/images/test_images_oneserver_negative.py
@@ -0,0 +1,162 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2012 OpenStack Foundation
+# Copyright 2013 IBM Corp.
+# 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 clients
+from tempest.common.utils import data_utils
+from tempest import exceptions
+from tempest.openstack.common import log as logging
+from tempest.test import attr
+from tempest.test import skip_because
+
+LOG = logging.getLogger(__name__)
+
+
+class ImagesOneServerNegativeTestJSON(base.BaseV2ComputeTest):
+ _interface = 'json'
+
+ def tearDown(self):
+ """Terminate test instances created after a test is executed."""
+ for image_id in self.image_ids:
+ self.client.delete_image(image_id)
+ self.image_ids.remove(image_id)
+ super(ImagesOneServerNegativeTestJSON, 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(ImagesOneServerNegativeTestJSON, self).setUp()
+ # Check if the server is in a clean state after test
+ try:
+ self.servers_client.wait_for_server_status(self.server_id,
+ 'ACTIVE')
+ except Exception:
+ LOG.exception('server %s timed out to become ACTIVE. rebuilding'
+ % self.server_id)
+ # Rebuild server if cannot reach the ACTIVE state
+ # Usually it means the server had a serious accident
+ self._reset_server()
+
+ def _reset_server(self):
+ self.__class__.server_id = self.rebuild_server(self.server_id)
+
+ @classmethod
+ def setUpClass(cls):
+ super(ImagesOneServerNegativeTestJSON, cls).setUpClass()
+ cls.client = cls.images_client
+ if not cls.config.service_available.glance:
+ skip_msg = ("%s skipped as glance is not available" % cls.__name__)
+ raise cls.skipException(skip_msg)
+
+ try:
+ resp, server = cls.create_test_server(wait_until='ACTIVE')
+ cls.server_id = server['id']
+ except Exception:
+ cls.tearDownClass()
+ raise
+
+ cls.image_ids = []
+
+ if cls.multi_user:
+ if cls.config.compute.allow_tenant_isolation:
+ creds = cls.isolated_creds.get_alt_creds()
+ username, tenant_name, password = creds
+ cls.alt_manager = clients.Manager(username=username,
+ password=password,
+ tenant_name=tenant_name)
+ else:
+ # Use the alt_XXX credentials in the config file
+ cls.alt_manager = clients.AltManager()
+ cls.alt_client = cls.alt_manager.images_client
+
+ @skip_because(bug="1006725")
+ @attr(type=['negative', 'gate'])
+ def test_create_image_specify_multibyte_character_image_name(self):
+ if self.__class__._interface == "xml":
+ raise self.skipException("Not testable in XML")
+ # 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,
+ invalid_name)
+
+ @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.server_id, snapshot_name, meta)
+
+ @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.server_id, snapshot_name, meta)
+
+ @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)
+ self.assertEqual(202, resp.status)
+ image_id = data_utils.parse_image_id(resp['location'])
+ self.image_ids.append(image_id)
+ self.addCleanup(self._reset_server)
+
+ # Create second snapshot
+ alt_snapshot_name = data_utils.rand_name('test-snap-')
+ self.assertRaises(exceptions.Conflict, self.client.create_image,
+ self.server_id, alt_snapshot_name)
+
+ @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.server_id, snapshot_name)
+
+ @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)
+ self.assertEqual(202, resp.status)
+ image_id = data_utils.parse_image_id(resp['location'])
+ self.image_ids.append(image_id)
+ self.addCleanup(self._reset_server)
+
+ # 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.image_ids.remove(image_id)
+
+ self.assertRaises(exceptions.NotFound, self.client.get_image, image_id)
+
+
+class ImagesOneServerNegativeTestXML(ImagesOneServerNegativeTestJSON):
+ _interface = 'xml'
diff --git a/tempest/api/compute/v3/images/test_list_image_filters.py b/tempest/api/compute/v3/images/test_list_image_filters.py
new file mode 100644
index 0000000..bfdd8b2
--- /dev/null
+++ b/tempest/api/compute/v3/images/test_list_image_filters.py
@@ -0,0 +1,231 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# 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 exceptions
+from tempest.openstack.common import log as logging
+from tempest.test import attr
+
+
+LOG = logging.getLogger(__name__)
+
+
+class ListImageFiltersTestJSON(base.BaseV2ComputeTest):
+ _interface = 'json'
+
+ @classmethod
+ def setUpClass(cls):
+ super(ListImageFiltersTestJSON, cls).setUpClass()
+ if not cls.config.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)
+ # when _interface='xml', one element for images_links in images
+ # ref: Question #224349
+ 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_nonexistant_image(self):
+ # Negative test: GET on non-existent image should fail
+ self.assertRaises(exceptions.NotFound, self.client.get_image, 999)
+
+
+class ListImageFiltersTestXML(ListImageFiltersTestJSON):
+ _interface = 'xml'