port nova v2 images related tests into nova v3 part1
This changeset only copies the v2 files into the appropriate v3
directories unchanged. This is being tried in order to make
reviewing of the porting easier as gerrit will display only what
is actually changed for v3 rather than entirely new files.
Partially implements blueprint nova-v3-api-tests
Change-Id: I4355adee713bb217da7f12d4c2fe9b8abf015b05
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'