Codebase list cinder-tempest-plugin / 4cb9068
Merge "[srbac] Test srbac on snapshots" Zuul authored 2 years ago Gerrit Code Review committed 2 years ago
2 changed file(s) with 464 addition(s) and 0 deletion(s). Raw diff Collapse all Expand all
+90
-0
cinder_tempest_plugin/rbac/v3/base.py less more
99 # License for the specific language governing permissions and limitations
1010 # under the License.
1111
12 from tempest.common import waiters
1213 from tempest import config
1314 from tempest.lib.common import api_microversion_fixture
1415 from tempest.lib.common import api_version_utils
16 from tempest.lib.common.utils import data_utils
17 from tempest.lib.common.utils import test_utils
18 from tempest.lib.decorators import cleanup_order
1519 from tempest import test
1620
1721 CONF = config.CONF
5963 CONF.volume.min_microversion))
6064
6165 def do_request(self, method, expected_status=200, client=None, **payload):
66 """Perform API call
67
68 Args:
69 method: Name of the API call
70 expected_status: HTTP desired response code
71 client: Client object if exists, None otherwise
72 payload: API call required parameters
73
74 Returns:
75 HTTP response
76 """
6277 if not client:
6378 client = self.client
6479 if isinstance(expected_status, type(Exception)):
6984 response = getattr(client, method)(**payload)
7085 self.assertEqual(response.response.status, expected_status)
7186 return response
87
88 @cleanup_order
89 def create_volume(self, client, **kwargs):
90 """Wrapper utility that returns a test volume
91
92 Args:
93 client: Client object
94
95 Returns:
96 ID of the created volume
97 """
98 kwargs['size'] = CONF.volume.volume_size
99 kwargs['name'] = data_utils.rand_name(
100 VolumeV3RbacBaseTests.__name__ + '-Volume'
101 )
102
103 volume_id = client.create_volume(**kwargs)['volume']['id']
104 self.cleanup(
105 test_utils.call_and_ignore_notfound_exc, func=self.delete_resource,
106 client=client, volume_id=volume_id
107 )
108 waiters.wait_for_volume_resource_status(
109 client=client, resource_id=volume_id, status='available'
110 )
111
112 return volume_id
113
114 @cleanup_order
115 def create_snapshot(self, client, volume_id, cleanup=True, **kwargs):
116 """Wrapper utility that returns a test snapshot.
117
118 Args:
119 client: Client object
120 volume_id: ID of the volume
121 cleanup: Boolean if should delete the snapshot
122
123 Returns:
124 ID of the created snapshot
125 """
126 kwargs['name'] = data_utils.rand_name(
127 VolumeV3RbacBaseTests.__name__ + '-Snapshot'
128 )
129
130 snapshot_id = client.create_snapshot(
131 volume_id=volume_id, **kwargs)['snapshot']['id']
132 if cleanup:
133 self.cleanup(
134 test_utils.call_and_ignore_notfound_exc,
135 func=self.delete_resource,
136 client=client, snapshot_id=snapshot_id
137 )
138 waiters.wait_for_volume_resource_status(
139 client=client, resource_id=snapshot_id, status='available'
140 )
141
142 return snapshot_id
143
144 @classmethod
145 def delete_resource(cls, client, **kwargs):
146 """Delete a resource by a given client
147
148 Args:
149 client: Client object
150
151 Keyword Args:
152 snapshot_id: ID of a snapshot
153 volume_id: ID of a volume
154 """
155 key, resource_id = list(kwargs.items())[0]
156 resource_name = key.split('_')[0]
157
158 del_action = getattr(client, f'delete_{resource_name}')
159 test_utils.call_and_ignore_notfound_exc(del_action, resource_id)
160 test_utils.call_and_ignore_notfound_exc(
161 client.wait_for_resource_deletion, resource_id)
+374
-0
cinder_tempest_plugin/rbac/v3/test_snapshots.py less more
0 # Licensed under the Apache License, Version 2.0 (the "License"); you may
1 # not use this file except in compliance with the License. You may obtain
2 # a copy of the License at
3 #
4 # http://www.apache.org/licenses/LICENSE-2.0
5 #
6 # Unless required by applicable law or agreed to in writing, software
7 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
8 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
9 # License for the specific language governing permissions and limitations
10 # under the License.
11
12 from tempest.common import waiters
13 from tempest import config
14 from tempest.lib.common.utils import data_utils
15 from tempest.lib.common.utils import test_utils
16 from tempest.lib import decorators
17 from tempest.lib import exceptions
18
19 from cinder_tempest_plugin.rbac.v3 import base as rbac_base
20
21 CONF = config.CONF
22
23
24 class VolumeV3RbacSnapshotsTests(rbac_base.VolumeV3RbacBaseTests):
25
26 @classmethod
27 def setup_clients(cls):
28 super().setup_clients()
29 cls.vol_other_client = cls.os_project_admin.volumes_client_latest
30 cls.snap_other_client = cls.os_project_admin.snapshots_client_latest
31
32 def _list_snapshots(self, expected_status):
33 """Test list_snapshots operation
34
35 Args:
36 expected_status: The expected HTTP response code
37 """
38 volume_id = self.create_volume(client=self.vol_other_client)
39 self.create_snapshot(
40 client=self.snap_other_client, volume_id=volume_id
41 )
42 self.do_request(
43 expected_status=expected_status, method='list_snapshots'
44 )
45
46 def _show_snapshot(self, expected_status):
47 """Test show_snapshot operation
48
49 Args:
50 expected_status: The expected HTTP response code
51 """
52 volume_id = self.create_volume(client=self.vol_other_client)
53 snapshot_id = self.create_snapshot(
54 client=self.snap_other_client, volume_id=volume_id
55 )
56 self.do_request(
57 expected_status=expected_status, method='show_snapshot',
58 snapshot_id=snapshot_id
59 )
60
61 def _create_snapshot(self, expected_status):
62 """Test create_snapshot operation.
63
64 Args:
65 expected_status: The expected HTTP response code
66 """
67 volume_id = self.create_volume(client=self.vol_other_client)
68 snap_name = data_utils.rand_name(
69 self.__name__ + '-Snapshot'
70 )
71 if expected_status == 202:
72 snapshot_id = self.do_request(
73 method='create_snapshot', expected_status=202,
74 volume_id=volume_id, name=snap_name
75 )['snapshot']['id']
76 self.addCleanup(
77 test_utils.call_and_ignore_notfound_exc, self.delete_resource,
78 client=self.client, snapshot_id=snapshot_id
79 )
80 waiters.wait_for_volume_resource_status(
81 client=self.client, resource_id=snapshot_id, status='available'
82 )
83 elif expected_status == exceptions.Forbidden:
84 self.do_request(
85 method='create_snapshot', expected_status=expected_status,
86 volume_id=volume_id, name=snap_name
87 )
88
89 def _remove_snapshot(self, expected_status):
90 """Test create_snapshot operation.
91
92 Args:
93 expected_status: The expected HTTP response code
94 """
95 volume_id = self.create_volume(client=self.vol_other_client)
96 snapshot_id = self.create_snapshot(
97 client=self.snap_other_client, volume_id=volume_id
98 )
99
100 self.do_request(
101 method='delete_snapshot', snapshot_id=snapshot_id,
102 expected_status=expected_status
103 )
104 if expected_status == 202:
105 self.client.wait_for_resource_deletion(id=snapshot_id)
106
107 def _reset_snapshot_status(self, expected_status):
108 """Test reset_snapshot_status operation.
109
110 Args:
111 expected_status: The expected HTTP response code
112 """
113 volume_id = self.create_volume(client=self.vol_other_client)
114 snapshot_id = self.create_snapshot(
115 client=self.snap_other_client, volume_id=volume_id
116 )
117 self.do_request(
118 'reset_snapshot_status', expected_status=expected_status,
119 snapshot_id=snapshot_id, status='error'
120 )
121
122 def _update_snapshot(self, expected_status):
123 """Test update_snapshot operation.
124
125 Args:
126 expected_status: The expected HTTP response code
127 """
128 volume_id = self.create_volume(client=self.vol_other_client)
129 snapshot_id = self.create_snapshot(
130 client=self.snap_other_client, volume_id=volume_id
131 )
132 new_desc = self.__name__ + '-update_test'
133 self.do_request(
134 method='update_snapshot', expected_status=expected_status,
135 snapshot_id=snapshot_id, description=new_desc
136 )
137
138 def _update_snapshot_status(self, expected_status):
139 """Test update_snapshot_status operation.
140
141 Args:
142 expected_status: The expected HTTP response code
143 """
144 volume_id = self.create_volume(client=self.vol_other_client)
145 snapshot_id = self.create_snapshot(
146 client=self.snap_other_client, volume_id=volume_id
147 )
148
149 reset_status = 'creating' if expected_status == 202 else 'error'
150 request_status = 'error' if expected_status == 202 else 'creating'
151 self.os_project_admin.snapshots_client_latest.reset_snapshot_status(
152 snapshot_id=snapshot_id, status=reset_status
153 )
154 waiters.wait_for_volume_resource_status(
155 client=self.os_project_admin.snapshots_client_latest,
156 resource_id=snapshot_id, status=reset_status
157 )
158
159 self.do_request(
160 'update_snapshot_status', expected_status=expected_status,
161 snapshot_id=snapshot_id, status=request_status, progress='80%'
162 )
163
164 def _force_delete_snapshot(self, expected_status):
165 """Test force_delete_snapshot operation.
166
167 Args:
168 expected_status: The expected HTTP response code
169 """
170 volume_id = self.create_volume(client=self.vol_other_client)
171 snapshot_id = self.create_snapshot(
172 client=self.snap_other_client, volume_id=volume_id
173 )
174 self.do_request(
175 method='force_delete_snapshot', snapshot_id=snapshot_id,
176 expected_status=expected_status
177 )
178 if expected_status != exceptions.Forbidden:
179 self.client.wait_for_resource_deletion(id=snapshot_id)
180 waiters.wait_for_volume_resource_status(
181 client=self.os_project_admin.volumes_client_latest,
182 resource_id=volume_id, status='available'
183 )
184
185 def _unmanage_snapshot(self, expected_status):
186 """Test unmanage_snapshot operation.
187
188 Args:
189 expected_status: The expected HTTP response code
190 """
191 volume_id = self.create_volume(client=self.vol_other_client)
192 snapshot_id = self.create_snapshot(
193 client=self.snap_other_client, volume_id=volume_id
194 )
195 self.do_request(
196 method='unmanage_snapshot',
197 expected_status=expected_status, snapshot_id=snapshot_id
198 )
199 if expected_status != exceptions.Forbidden:
200 self.client.wait_for_resource_deletion(id=snapshot_id)
201
202 def _manage_snapshot(self, client, expected_status):
203 """Test reset_snapshot_status operation.
204
205 Args:
206 client: The client to perform the needed request
207 expected_status: The expected HTTP response code
208 """
209 # Create a volume
210 volume_id = self.create_volume(client=self.vol_other_client)
211
212 # Create a snapshot
213 snapshot_id = self.create_snapshot(
214 client=self.snap_other_client,
215 volume_id=volume_id,
216 cleanup=False
217 )
218 # Unmanage the snapshot
219 # Unmanage snapshot function works almost the same as delete snapshot,
220 # but it does not delete the snapshot data
221 self.snap_other_client.unmanage_snapshot(snapshot_id)
222 self.client.wait_for_resource_deletion(snapshot_id)
223
224 # Verify the original snapshot does not exist in snapshot list
225 params = {'all_tenants': 1}
226 all_snapshots = self.snap_other_client.list_snapshots(
227 detail=True, **params)['snapshots']
228 self.assertNotIn(snapshot_id, [v['id'] for v in all_snapshots])
229
230 # Manage the snapshot
231 name = data_utils.rand_name(
232 self.__class__.__name__ + '-Managed-Snapshot'
233 )
234 description = data_utils.rand_name(
235 self.__class__.__name__ + '-Managed-Snapshot-Description'
236 )
237 metadata = {"manage-snap-meta1": "value1",
238 "manage-snap-meta2": "value2",
239 "manage-snap-meta3": "value3"}
240 snapshot_ref = {
241 'volume_id': volume_id,
242 'ref': {CONF.volume.manage_snapshot_ref[0]:
243 CONF.volume.manage_snapshot_ref[1] % snapshot_id},
244 'name': name,
245 'description': description,
246 'metadata': metadata
247 }
248
249 new_snapshot = self.do_request(
250 client=client,
251 method='manage_snapshot', expected_status=expected_status,
252 volume_id=volume_id, ref=snapshot_ref
253 )
254 if expected_status != exceptions.Forbidden:
255 snapshot = new_snapshot['snapshot']
256 waiters.wait_for_volume_resource_status(
257 client=self.snap_other_client,
258 resource_id=snapshot['id'],
259 status='available'
260 )
261 self.delete_resource(
262 client=self.snap_other_client, snapshot_id=snapshot['id']
263 )
264
265
266 class ProjectReaderTests(VolumeV3RbacSnapshotsTests):
267
268 credentials = ['project_reader', 'project_admin']
269
270 @classmethod
271 def setup_clients(cls):
272 super().setup_clients()
273 cls.client = cls.os_project_reader.snapshots_client_latest
274
275 @decorators.idempotent_id('dd8e19dc-c8fd-443c-8aed-cdffe07fa6be')
276 def test_list_snapshots(self):
277 self._list_snapshots(expected_status=200)
278
279 @decorators.idempotent_id('6f69e8ed-4e11-40a1-9620-258cf3c45872')
280 def test_show_snapshot(self):
281 self._show_snapshot(expected_status=200)
282
283 @decorators.skip_because(bug="2017108")
284 @decorators.idempotent_id('13ae344f-fa01-44cc-b9f1-d04452940dc1')
285 def test_create_snapshot(self):
286 self._create_snapshot(expected_status=exceptions.Forbidden)
287
288 @decorators.skip_because(bug="2017108")
289 @decorators.idempotent_id('5b58f647-da0f-4d2a-bf68-680fc692efb4')
290 def test_delete_snapshot(self):
291 self._remove_snapshot(expected_status=exceptions.Forbidden)
292
293 @decorators.idempotent_id('809d8c8c-25bf-4f1f-9b77-1a81ce4292d1')
294 def test_reset_snapshot_status(self):
295 self._reset_snapshot_status(expected_status=exceptions.Forbidden)
296
297 @decorators.skip_because(bug="2017108")
298 @decorators.idempotent_id('c46f5df8-9a6f-4ed6-b94c-3b65ef05ee9e')
299 def test_update_snapshot(self):
300 self._update_snapshot(expected_status=exceptions.Forbidden)
301
302 @decorators.skip_because(bug="2017108")
303 @decorators.idempotent_id('c90f98d7-3665-4c9f-820f-3f4c2adfdbf5')
304 def test_update_snapshot_status(self):
305 self._update_snapshot_status(expected_status=exceptions.Forbidden)
306
307 @decorators.idempotent_id('63aa8184-897d-4e00-9b80-d2e7828f1b13')
308 def test_force_delete_snapshot(self):
309 self._force_delete_snapshot(expected_status=exceptions.Forbidden)
310
311 @decorators.idempotent_id('35495666-b663-4c68-ba44-0695e30a6838')
312 def test_unmanage_snapshot(self):
313 self._unmanage_snapshot(expected_status=exceptions.Forbidden)
314
315 @decorators.idempotent_id('d2d1326d-fb47-4448-a1e1-2d1219d30fd5')
316 def test_manage_snapshot(self):
317 self._manage_snapshot(
318 expected_status=exceptions.Forbidden,
319 client=self.os_project_reader.snapshot_manage_client_latest
320 )
321
322
323 class ProjectMemberTests(VolumeV3RbacSnapshotsTests):
324
325 credentials = ['project_member', 'project_admin']
326
327 @classmethod
328 def setup_clients(cls):
329 super().setup_clients()
330 cls.client = cls.os_project_member.snapshots_client_latest
331
332 @decorators.idempotent_id('5b3ec87f-443f-42f7-bd3c-ab05ea30c5e1')
333 def test_list_snapshots(self):
334 self._list_snapshots(expected_status=200)
335
336 @decorators.idempotent_id('6fee8967-951c-4957-b51b-97b83c13c7c3')
337 def test_show_snapshot(self):
338 self._show_snapshot(expected_status=200)
339
340 @decorators.idempotent_id('43f77b31-aab4-46d0-b76f-e17000d23589')
341 def test_create_snapshot(self):
342 self._create_snapshot(expected_status=202)
343
344 @decorators.idempotent_id('22939122-8b4e-47d5-abaa-774bc55c07fc')
345 def test_delete_snapshot(self):
346 self._remove_snapshot(expected_status=202)
347
348 @decorators.idempotent_id('da391afd-8baa-458b-b222-f6ab42ab47c3')
349 def test_reset_snapshot_status(self):
350 self._reset_snapshot_status(expected_status=exceptions.Forbidden)
351
352 @decorators.idempotent_id('a774bdca-bfbe-477d-9711-5fb64d7e34ea')
353 def test_update_snapshot(self):
354 self._update_snapshot(expected_status=200)
355
356 @decorators.idempotent_id('12e00e1b-bf84-41c1-8a1e-8625d1317789')
357 def test_update_snapshot_status(self):
358 self._update_snapshot_status(expected_status=202)
359
360 @decorators.idempotent_id('e7cb3eb0-d607-4c90-995d-df82d030eca8')
361 def test_force_delete_snapshot(self):
362 self._force_delete_snapshot(expected_status=exceptions.Forbidden)
363
364 @decorators.idempotent_id('dd7da3da-68ef-42f5-af1d-29803a4a04fd')
365 def test_unmanage_snapshot(self):
366 self._unmanage_snapshot(expected_status=exceptions.Forbidden)
367
368 @decorators.idempotent_id('c2501d05-9bca-42d7-9ab5-c0d9133e762f')
369 def test_manage_snapshot(self):
370 self._manage_snapshot(
371 expected_status=exceptions.Forbidden,
372 client=self.os_project_member.snapshot_manage_client_latest
373 )