Merge "Encryption Test Scenario for clone volume"
diff --git a/cinder_tempest_plugin/scenario/test_volume_encrypted.py b/cinder_tempest_plugin/scenario/test_volume_encrypted.py
new file mode 100644
index 0000000..2366cb8
--- /dev/null
+++ b/cinder_tempest_plugin/scenario/test_volume_encrypted.py
@@ -0,0 +1,183 @@
+# 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.common import utils
+from tempest.common import waiters
+from tempest import config
+from tempest.lib.common.utils import data_utils
+from tempest.lib import decorators
+from tempest.scenario import manager
+
+CONF = config.CONF
+
+
+class TestEncryptedCinderVolumes(manager.EncryptionScenarioTest,
+ manager.ScenarioTest):
+
+ @classmethod
+ def skip_checks(cls):
+ super(TestEncryptedCinderVolumes, cls).skip_checks()
+ if not CONF.compute_feature_enabled.attach_encrypted_volume:
+ raise cls.skipException('Encrypted volume attach is not supported')
+
+ @classmethod
+ def resource_setup(cls):
+ super(TestEncryptedCinderVolumes, cls).resource_setup()
+
+ @classmethod
+ def resource_cleanup(cls):
+ super(TestEncryptedCinderVolumes, cls).resource_cleanup()
+
+ def launch_instance(self):
+ image = self.image_create()
+ keypair = self.create_keypair()
+
+ return self.create_server(image_id=image, key_name=keypair['name'])
+
+ def attach_detach_volume(self, server, volume):
+ attached_volume = self.nova_volume_attach(server, volume)
+ self.nova_volume_detach(server, attached_volume)
+
+ def _delete_server(self, server):
+ self.servers_client.delete_server(server['id'])
+ waiters.wait_for_server_termination(self.servers_client, server['id'])
+
+ def create_encrypted_volume_from_image(self, encryption_provider,
+ volume_type='luks',
+ key_size=256,
+ cipher='aes-xts-plain64',
+ control_location='front-end',
+ **kwargs):
+ """Create an encrypted volume from image.
+
+ :param image_id: ID of the image to create volume from,
+ CONF.compute.image_ref by default
+ :param name: name of the volume,
+ '$classname-volume-origin' by default
+ :param **kwargs: additional parameters
+ """
+ volume_type = self.create_volume_type(name=volume_type)
+ self.create_encryption_type(type_id=volume_type['id'],
+ provider=encryption_provider,
+ key_size=key_size,
+ cipher=cipher,
+ control_location=control_location)
+ image_id = kwargs.pop('image_id', CONF.compute.image_ref)
+ name = kwargs.pop('name', None)
+ if not name:
+ namestart = self.__class__.__name__ + '-volume-origin'
+ name = data_utils.rand_name(namestart)
+ return self.create_volume(volume_type=volume_type['name'],
+ name=name, imageRef=image_id,
+ **kwargs)
+
+ @decorators.idempotent_id('5bb622ab-5060-48a8-8840-d589a548b9e4')
+ @utils.services('volume')
+ @utils.services('compute')
+ def test_attach_cloned_encrypted_volume(self):
+
+ """This test case attempts to reproduce the following steps:
+
+ * Create an encrypted volume
+ * Create clone from volume
+ * Boot an instance and attach/dettach cloned volume
+
+ """
+
+ volume = self.create_encrypted_volume('luks', volume_type='luks')
+ kwargs = {
+ 'display_name': data_utils.rand_name(self.__class__.__name__),
+ 'source_volid': volume['id'],
+ 'volume_type': volume['volume_type'],
+ 'size': volume['size']
+ }
+ volume_s = self.volumes_client.create_volume(**kwargs)['volume']
+ self.addCleanup(self.volumes_client.wait_for_resource_deletion,
+ volume_s['id'])
+ self.addCleanup(self.volumes_client.delete_volume, volume_s['id'])
+ waiters.wait_for_volume_resource_status(
+ self.volumes_client, volume_s['id'], 'available')
+ volume_source = self.volumes_client.show_volume(
+ volume_s['id'])['volume']
+ server = self.launch_instance()
+ self.attach_detach_volume(server, volume_source)
+
+ @decorators.idempotent_id('5bb622ab-5060-48a8-8840-d589a548b7e4')
+ @utils.services('volume')
+ @utils.services('compute')
+ @utils.services('image')
+ def test_boot_cloned_encrypted_volume(self):
+
+ """This test case attempts to reproduce the following steps:
+
+ * Create an encrypted volume from image
+ * Boot an instance from the volume
+ * Write data to the volume
+ * Detach volume
+ * Create a clone from the first volume
+ * Create another encrypted volume from source_volumeid
+ * Boot an instance from cloned volume
+ * Verify the data
+ """
+
+ keypair = self.create_keypair()
+ security_group = self._create_security_group()
+
+ volume = self.create_encrypted_volume_from_image('luks')
+
+ # create an instance from volume
+ instance_1st = self.boot_instance_from_resource(
+ source_id=volume['id'],
+ source_type='volume',
+ keypair=keypair,
+ security_group=security_group)
+
+ # write content to volume on instance
+ ip_instance_1st = self.get_server_ip(instance_1st)
+ timestamp = self.create_timestamp(ip_instance_1st,
+ private_key=keypair['private_key'],
+ server=instance_1st)
+ # delete instance
+ self._delete_server(instance_1st)
+
+ # create clone
+ kwargs = {
+ 'display_name': data_utils.rand_name(self.__class__.__name__),
+ 'source_volid': volume['id'],
+ 'volume_type': volume['volume_type'],
+ 'size': volume['size']
+ }
+ volume_s = self.volumes_client.create_volume(**kwargs)['volume']
+
+ self.addCleanup(self.volumes_client.wait_for_resource_deletion,
+ volume_s['id'])
+ self.addCleanup(self.volumes_client.delete_volume, volume_s['id'])
+ waiters.wait_for_volume_resource_status(
+ self.volumes_client, volume_s['id'], 'available')
+
+ # create an instance from volume
+ instance_2nd = self.boot_instance_from_resource(
+ source_id=volume_s['id'],
+ source_type='volume',
+ keypair=keypair,
+ security_group=security_group)
+
+ # check the content of written file
+ ip_instance_2nd = self.get_server_ip(instance_2nd)
+ timestamp2 = self.get_timestamp(ip_instance_2nd,
+ private_key=keypair['private_key'],
+ server=instance_2nd)
+
+ self.assertEqual(timestamp, timestamp2)
+
+ # delete instance
+ self._delete_server(instance_2nd)