Codebase list cinder-tempest-plugin / e2fbcad
Merge "Test the ability to transfer encrypted volumes" Zuul authored 3 years ago Gerrit Code Review committed 3 years ago
3 changed file(s) with 202 addition(s) and 0 deletion(s). Raw diff Collapse all Expand all
+12
-0
cinder_tempest_plugin/config.py less more
2222 default=False,
2323 help='Enable to run Cinder volume revert tests'),
2424 ]
25
26 # The barbican service is discovered by config_tempest [1], and will appear
27 # in the [service_available] group in tempest.conf. However, the 'barbican'
28 # option isn't registered by tempest itself, and so we do it here. This adds
29 # the ability to test CONF.service_available.barbican.
30 #
31 # [1] I96800a95f844ce7675d266e456e01620e63e347a
32 service_available_option = [
33 cfg.BoolOpt('barbican',
34 default=False,
35 help="Whether or not barbican is expected to be available"),
36 ]
+5
-0
cinder_tempest_plugin/plugin.py less more
4545 config.register_opt_group(conf, config.volume_feature_group,
4646 project_config.cinder_option)
4747
48 config.register_opt_group(conf, config.service_available_group,
49 project_config.service_available_option)
50
4851 def get_opt_lists(self):
4952 """Get a list of options for sample config generation.
5053
5356 """
5457 return [
5558 (config.volume_feature_group.name, project_config.cinder_option),
59 (config.service_available_group.name,
60 project_config.service_available_option),
5661 ]
+185
-0
cinder_tempest_plugin/scenario/test_encrypted_volume_transfer.py less more
0 # Copyright 2022 Red Hat, Inc.
1 # All rights reserved.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License"); you may
4 # not use this file except in compliance with the License. You may obtain
5 # a copy of the License at
6 #
7 # http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12 # License for the specific language governing permissions and limitations
13 # under the License.
14
15 from tempest.common import utils
16 from tempest.common import waiters
17 from tempest import config
18 from tempest.lib import decorators
19 from tempest.scenario import manager
20
21 CONF = config.CONF
22
23
24 class TransferEncryptedVolumeTest(manager.EncryptionScenarioTest):
25
26 volume_min_microversion = '3.70'
27 volume_max_microversion = 'latest'
28
29 credentials = ['primary', 'alt', 'admin']
30
31 @classmethod
32 def setup_clients(cls):
33 super(TransferEncryptedVolumeTest, cls).setup_clients()
34
35 # We need the "mv355" volume transfers client
36 cls.client = cls.os_primary.volume_transfers_mv355_client_latest
37 cls.alt_client = cls.os_alt.volume_transfers_mv355_client_latest
38 cls.alt_volumes_client = cls.os_alt.volumes_client_latest
39
40 @classmethod
41 def skip_checks(cls):
42 super(TransferEncryptedVolumeTest, cls).skip_checks()
43 if not CONF.service_available.barbican:
44 raise cls.skipException('Barbican is required')
45
46 def setUp(self):
47 super(TransferEncryptedVolumeTest, self).setUp()
48 self.keypair = self.create_keypair()
49 self.security_group = self.create_security_group()
50
51 def _create_encrypted_volume_from_image(self):
52 volume_type = self.create_volume_type()
53 self.create_encryption_type(type_id=volume_type['id'],
54 provider='luks',
55 key_size=256,
56 cipher='aes-xts-plain64',
57 control_location='front-end')
58 return self.create_volume_from_image(volume_type=volume_type['id'])
59
60 def _create_or_get_timestamp(self, volume, timestamp_fn):
61 server = self.boot_instance_from_resource(
62 source_id=volume['id'],
63 source_type='volume',
64 keypair=self.keypair,
65 security_group=self.security_group)
66 server_ip = self.get_server_ip(server)
67 timestamp = timestamp_fn(server_ip,
68 private_key=self.keypair['private_key'],
69 server=server)
70 self.servers_client.delete_server(server['id'])
71 waiters.wait_for_server_termination(self.servers_client, server['id'])
72 return timestamp
73
74 def _create_transfer(self, volume, transfer_client, volumes_client):
75 body = transfer_client.create_volume_transfer(volume_id=volume['id'])
76 transfer = body['transfer']
77 waiters.wait_for_volume_resource_status(volumes_client,
78 volume['id'],
79 'awaiting-transfer')
80 return transfer
81
82 def _accept_transfer(self, transfer, transfer_client, volumes_client):
83 _ = transfer_client.accept_volume_transfer(
84 transfer['id'], auth_key=transfer['auth_key'])
85 waiters.wait_for_volume_resource_status(volumes_client,
86 transfer['volume_id'],
87 'available')
88
89 def _delete_transfer(self, transfer, transfer_client, volumes_client):
90 _ = transfer_client.delete_volume_transfer(transfer['id'])
91 waiters.wait_for_volume_resource_status(volumes_client,
92 transfer['volume_id'],
93 'available')
94
95 @decorators.idempotent_id('a694dc4d-d11b-45cb-b268-62e76cc1b4f4')
96 @utils.services('compute', 'volume', 'image', 'network')
97 def test_create_accept_volume_transfer(self):
98 """Verify the ability to transfer an encrypted volume:
99
100 * Create an encrypted volume from image
101 * Boot an instance from the volume and write a timestamp
102 * Transfer the volume to another project, then transfer it back
103 again to the original project (see comments in the code for why
104 this is done).
105 * Boot annother instance from the volume and read the timestamp
106 * Verify the timestamps match, and the volume has a new
107 encryption_key_id.
108 """
109
110 # Create a bootable encrypted volume.
111 volume = self._create_encrypted_volume_from_image()
112
113 # Create an instance from the volume and write a timestamp.
114 timestamp_1 = self._create_or_get_timestamp(volume,
115 self.create_timestamp)
116
117 # Transfer the volume to another project.
118 transfer = self._create_transfer(volume,
119 self.client,
120 self.volumes_client)
121 self._accept_transfer(transfer,
122 self.alt_client,
123 self.alt_volumes_client)
124
125 # Transfer the volume back to the original project. This is done
126 # only because it's awkward in tempest to boot an instance and
127 # access it (to read the timestamp) in another project without
128 # setting up another security group and group rules.
129 transfer = self._create_transfer(volume,
130 self.alt_client,
131 self.alt_volumes_client)
132 self._accept_transfer(transfer, self.client, self.volumes_client)
133
134 # Create another instance from the volume and read the timestamp.
135 timestamp_2 = self._create_or_get_timestamp(volume,
136 self.get_timestamp)
137
138 self.assertEqual(timestamp_1, timestamp_2)
139
140 # Verify the volume has a new encryption_key_id.
141 encryption_key_id_1 = volume['encryption_key_id']
142 volume = self.volumes_client.show_volume(volume['id'])['volume']
143 encryption_key_id_2 = volume['encryption_key_id']
144
145 self.assertNotEqual(encryption_key_id_1, encryption_key_id_2)
146
147 @decorators.idempotent_id('00c04d27-b3c6-454c-a0b4-223a195c4a89')
148 @utils.services('compute', 'volume', 'image', 'network')
149 def test_create_delete_volume_transfer(self):
150 """Verify the ability to cancel an encrypted volume transfer:
151
152 * Create an encrypted volume from image
153 * Boot an instance from the volume and write a timestamp
154 * Create and delete a volume transfer
155 * Boot annother instance from the volume and read the timestamp
156 * Verify the timestamps match, and the volume has a new
157 encryption_key_id.
158 """
159
160 # Create a bootable encrypted volume.
161 volume = self._create_encrypted_volume_from_image()
162
163 # Create an instance from the volume and write a timestamp.
164 timestamp_1 = self._create_or_get_timestamp(volume,
165 self.create_timestamp)
166
167 # Create and then delete a transfer of the volume
168 transfer = self._create_transfer(volume,
169 self.client,
170 self.volumes_client)
171 self._delete_transfer(transfer, self.client, self.volumes_client)
172
173 # Create another instance from the volume and read the timestamp.
174 timestamp_2 = self._create_or_get_timestamp(volume,
175 self.get_timestamp)
176
177 self.assertEqual(timestamp_1, timestamp_2)
178
179 # Verify the volume has a new encryption_key_id.
180 encryption_key_id_1 = volume['encryption_key_id']
181 volume = self.volumes_client.show_volume(volume['id'])['volume']
182 encryption_key_id_2 = volume['encryption_key_id']
183
184 self.assertNotEqual(encryption_key_id_1, encryption_key_id_2)