Coverage for manila/tests/api/v2/test_share_group_snapshots.py: 97%
336 statements
« prev ^ index » next coverage.py v7.11.0, created at 2026-02-18 22:19 +0000
1# Copyright 2016 Alex Meade
2# All Rights Reserved.
3#
4# Licensed under the Apache License, Version 2.0 (the "License"); you may
5# not use this file except in compliance with the License. You may obtain
6# a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13# License for the specific language governing permissions and limitations
14# under the License.
16import copy
17import datetime
18from unittest import mock
20import ddt
21from oslo_config import cfg
22from oslo_serialization import jsonutils
23from oslo_utils import uuidutils
24import webob
26from manila.api.openstack import wsgi
27from manila.api.v2 import share_group_snapshots
28from manila.common import constants
29from manila import context
30from manila import db
31from manila import exception
32from manila import policy
33from manila import test
34from manila.tests.api import fakes
35from manila.tests import db_utils
37CONF = cfg.CONF
38SG_GRADUATION_VERSION = '2.55'
41@ddt.ddt
42class ShareGroupSnapshotAPITest(test.TestCase):
44 def setUp(self):
45 super(ShareGroupSnapshotAPITest, self).setUp()
46 self.controller = share_group_snapshots.ShareGroupSnapshotController()
47 self.resource_name = self.controller.resource_name
48 self.api_version = '2.31'
49 self.mock_policy_check = self.mock_object(
50 policy, 'check_policy', mock.Mock(return_value=True))
51 self.request = fakes.HTTPRequest.blank(
52 '/share-groups', version=self.api_version, experimental=True)
53 self.context = self.request.environ['manila.context']
54 self.admin_context = context.RequestContext('admin', 'fake', True)
55 self.member_context = context.RequestContext('fake', 'fake')
56 self.flags(transport_url='rabbit://fake:fake@mqhost:5672')
58 def _get_fake_share_group_snapshot(self, **values):
59 snap = {
60 'id': 'fake_id',
61 'user_id': 'fakeuser',
62 'project_id': 'fakeproject',
63 'status': constants.STATUS_CREATING,
64 'name': None,
65 'description': None,
66 'share_group_id': None,
67 'created_at': datetime.datetime(1, 1, 1, 1, 1, 1),
68 'members': [],
69 }
71 snap.update(**values)
73 expected_snap = copy.deepcopy(snap)
74 del expected_snap['user_id']
75 return snap, expected_snap
77 def _get_fake_simple_share_group_snapshot(self, **values):
78 snap = {
79 'id': 'fake_id',
80 'name': None,
81 }
83 snap.update(**values)
84 expected_snap = copy.deepcopy(snap)
85 return snap, expected_snap
87 def _get_fake_share_group_snapshot_member(self, **values):
88 member = {
89 'id': 'fake_id',
90 'user_id': 'fakeuser',
91 'project_id': 'fakeproject',
92 'status': constants.STATUS_CREATING,
93 'share_group_snapshot_id': None,
94 'share_proto': None,
95 'share_id': None,
96 'size': None,
97 'created_at': datetime.datetime(1, 1, 1, 1, 1, 1),
98 }
100 member.update(**values)
102 expected_member = copy.deepcopy(member)
103 del expected_member['user_id']
104 del expected_member['status']
105 expected_member['share_protocol'] = member['share_proto']
106 del expected_member['share_proto']
107 return member, expected_member
109 def _get_fake_custom_request_and_context(self, microversion, experimental):
110 req = fakes.HTTPRequest.blank(
111 '/share-group-snapshots', version=microversion,
112 experimental=experimental)
113 req_context = req.environ['manila.context']
114 return req, req_context
116 def test_create_invalid_body(self):
117 body = {"not_group_snapshot": {}}
119 self.assertRaises(
120 webob.exc.HTTPBadRequest,
121 self.controller.create, self.request, body)
123 self.mock_policy_check.assert_called_once_with(
124 self.context, self.resource_name, 'create')
126 def test_create_no_share_group_id(self):
127 body = {"share_group_snapshot": {}}
129 self.assertRaises(
130 webob.exc.HTTPBadRequest,
131 self.controller.create, self.request, body)
133 self.mock_policy_check.assert_called_once_with(
134 self.context, self.resource_name, 'create')
136 @ddt.data({'microversion': '2.31', 'experimental': True},
137 {'microversion': SG_GRADUATION_VERSION, 'experimental': False})
138 @ddt.unpack
139 def test_create(self, microversion, experimental):
140 fake_snap, expected_snap = self._get_fake_share_group_snapshot()
141 fake_id = uuidutils.generate_uuid()
142 body = {"share_group_snapshot": {"share_group_id": fake_id}}
143 mock_create = self.mock_object(
144 self.controller.share_group_api, 'create_share_group_snapshot',
145 mock.Mock(return_value=fake_snap))
146 req, req_context = self._get_fake_custom_request_and_context(
147 microversion, experimental)
149 res_dict = self.controller.create(req, body)
151 self.mock_policy_check.assert_called_once_with(
152 req_context, self.resource_name, 'create')
153 mock_create.assert_called_once_with(
154 req_context, share_group_id=fake_id)
155 res_dict['share_group_snapshot'].pop('links')
157 self.assertEqual(expected_snap, res_dict['share_group_snapshot'])
159 def test_create_group_does_not_exist(self):
160 fake_id = uuidutils.generate_uuid()
161 body = {"share_group_snapshot": {"share_group_id": fake_id}}
162 self.mock_object(
163 self.controller.share_group_api, 'create_share_group_snapshot',
164 mock.Mock(side_effect=exception.ShareGroupNotFound(
165 share_group_id=uuidutils.generate_uuid())))
167 self.assertRaises(
168 webob.exc.HTTPBadRequest,
169 self.controller.create, self.request, body)
171 self.mock_policy_check.assert_called_once_with(
172 self.context, self.resource_name, 'create')
174 def test_create_group_does_not_a_uuid(self):
175 self.mock_object(
176 self.controller.share_group_api, 'create_share_group_snapshot',
177 mock.Mock(side_effect=exception.ShareGroupNotFound(
178 share_group_id='not_a_uuid',
179 )))
180 body = {"share_group_snapshot": {"share_group_id": "not_a_uuid"}}
182 self.assertRaises(
183 webob.exc.HTTPBadRequest,
184 self.controller.create, self.request, body)
186 self.mock_policy_check.assert_called_once_with(
187 self.context, self.resource_name, 'create')
189 def test_create_invalid_share_group(self):
190 fake_id = uuidutils.generate_uuid()
191 body = {"share_group_snapshot": {"share_group_id": fake_id}}
192 self.mock_object(
193 self.controller.share_group_api, 'create_share_group_snapshot',
194 mock.Mock(side_effect=exception.InvalidShareGroup(
195 reason='bad_status')))
197 self.assertRaises(
198 webob.exc.HTTPConflict, self.controller.create, self.request, body)
200 self.mock_policy_check.assert_called_once_with(
201 self.context, self.resource_name, 'create')
203 def test_create_with_name(self):
204 fake_name = 'fake_name'
205 fake_snap, expected_snap = self._get_fake_share_group_snapshot(
206 name=fake_name)
207 fake_id = uuidutils.generate_uuid()
208 mock_create = self.mock_object(
209 self.controller.share_group_api, 'create_share_group_snapshot',
210 mock.Mock(return_value=fake_snap))
211 body = {
212 "share_group_snapshot": {
213 "share_group_id": fake_id,
214 "name": fake_name,
215 }
216 }
217 res_dict = self.controller.create(self.request, body)
219 res_dict['share_group_snapshot'].pop('links')
221 mock_create.assert_called_once_with(
222 self.context, share_group_id=fake_id, name=fake_name)
223 self.assertEqual(expected_snap, res_dict['share_group_snapshot'])
224 self.mock_policy_check.assert_called_once_with(
225 self.context, self.resource_name, 'create')
227 def test_create_with_description(self):
228 fake_description = 'fake_description'
229 fake_snap, expected_snap = self._get_fake_share_group_snapshot(
230 description=fake_description)
231 fake_id = uuidutils.generate_uuid()
232 mock_create = self.mock_object(
233 self.controller.share_group_api, 'create_share_group_snapshot',
234 mock.Mock(return_value=fake_snap))
236 body = {
237 "share_group_snapshot": {
238 "share_group_id": fake_id,
239 "description": fake_description,
240 }
241 }
242 res_dict = self.controller.create(self.request, body)
244 res_dict['share_group_snapshot'].pop('links')
246 mock_create.assert_called_once_with(
247 self.context, share_group_id=fake_id, description=fake_description)
248 self.assertEqual(expected_snap, res_dict['share_group_snapshot'])
249 self.mock_policy_check.assert_called_once_with(
250 self.context, self.resource_name, 'create')
252 def test_create_with_name_and_description(self):
253 fake_name = 'fake_name'
254 fake_description = 'fake_description'
255 fake_id = uuidutils.generate_uuid()
256 fake_snap, expected_snap = self._get_fake_share_group_snapshot(
257 description=fake_description, name=fake_name)
258 mock_create = self.mock_object(
259 self.controller.share_group_api, 'create_share_group_snapshot',
260 mock.Mock(return_value=fake_snap))
262 body = {
263 "share_group_snapshot": {
264 "share_group_id": fake_id,
265 "description": fake_description,
266 "name": fake_name,
267 }
268 }
269 res_dict = self.controller.create(self.request, body)
271 res_dict['share_group_snapshot'].pop('links')
273 mock_create.assert_called_once_with(
274 self.context, share_group_id=fake_id, name=fake_name,
275 description=fake_description)
276 self.assertEqual(expected_snap, res_dict['share_group_snapshot'])
277 self.mock_policy_check.assert_called_once_with(
278 self.context, self.resource_name, 'create')
280 @ddt.data({'microversion': '2.31', 'experimental': True},
281 {'microversion': SG_GRADUATION_VERSION, 'experimental': False})
282 @ddt.unpack
283 def test_update_with_name_and_description(self, microversion,
284 experimental):
285 fake_name = 'fake_name'
286 fake_description = 'fake_description'
287 fake_id = uuidutils.generate_uuid()
288 fake_snap, expected_snap = self._get_fake_share_group_snapshot(
289 description=fake_description, name=fake_name)
290 self.mock_object(
291 self.controller.share_group_api, 'get_share_group_snapshot',
292 mock.Mock(return_value=fake_snap))
293 mock_update = self.mock_object(
294 self.controller.share_group_api, 'update_share_group_snapshot',
295 mock.Mock(return_value=fake_snap))
296 req, req_context = self._get_fake_custom_request_and_context(
297 microversion, experimental)
299 body = {
300 "share_group_snapshot": {
301 "description": fake_description,
302 "name": fake_name,
303 }
304 }
305 res_dict = self.controller.update(req, fake_id, body)
307 res_dict['share_group_snapshot'].pop('links')
309 mock_update.assert_called_once_with(
310 req_context, fake_snap,
311 {"name": fake_name, "description": fake_description})
312 self.assertEqual(expected_snap, res_dict['share_group_snapshot'])
313 self.mock_policy_check.assert_called_once_with(
314 req_context, self.resource_name, 'update')
316 def test_update_snapshot_not_found(self):
317 body = {"share_group_snapshot": {}}
318 self.mock_object(
319 self.controller.share_group_api, 'get_share_group_snapshot',
320 mock.Mock(side_effect=exception.NotFound))
322 self.assertRaises(
323 webob.exc.HTTPNotFound,
324 self.controller.update, self.request, 'fake_id', body)
326 self.mock_policy_check.assert_called_once_with(
327 self.context, self.resource_name, 'update')
329 def test_update_invalid_body(self):
330 body = {"not_group_snapshot": {}}
331 self.assertRaises(webob.exc.HTTPBadRequest,
332 self.controller.update,
333 self.request, 'fake_id', body)
334 self.mock_policy_check.assert_called_once_with(
335 self.context, self.resource_name, 'update')
337 def test_update_invalid_body_invalid_field(self):
338 body = {"share_group_snapshot": {"unknown_field": ""}}
339 exc = self.assertRaises(webob.exc.HTTPBadRequest,
340 self.controller.update,
341 self.request, 'fake_id', body)
342 self.assertIn('unknown_field', str(exc))
343 self.mock_policy_check.assert_called_once_with(
344 self.context, self.resource_name, 'update')
346 def test_update_invalid_body_readonly_field(self):
347 body = {"share_group_snapshot": {"created_at": []}}
348 exc = self.assertRaises(webob.exc.HTTPBadRequest,
349 self.controller.update,
350 self.request, 'fake_id', body)
351 self.assertIn('created_at', str(exc))
352 self.mock_policy_check.assert_called_once_with(
353 self.context, self.resource_name, 'update')
355 @ddt.data({'microversion': '2.31', 'experimental': True},
356 {'microversion': SG_GRADUATION_VERSION, 'experimental': False})
357 @ddt.unpack
358 def test_list_index(self, microversion, experimental):
359 fake_snap, expected_snap = self._get_fake_simple_share_group_snapshot()
360 self.mock_object(
361 self.controller.share_group_api, 'get_all_share_group_snapshots',
362 mock.Mock(return_value=[fake_snap]))
363 req, req_context = self._get_fake_custom_request_and_context(
364 microversion, experimental)
366 res_dict = self.controller.index(req)
368 res_dict['share_group_snapshots'][0].pop('links')
370 self.assertEqual([expected_snap], res_dict['share_group_snapshots'])
371 self.mock_policy_check.assert_called_once_with(
372 req_context, self.resource_name, 'get_all')
374 def test_list_index_no_share_groups(self):
375 self.mock_object(
376 self.controller.share_group_api, 'get_all_share_group_snapshots',
377 mock.Mock(return_value=[]))
379 res_dict = self.controller.index(self.request)
381 self.assertEqual([], res_dict['share_group_snapshots'])
382 self.mock_policy_check.assert_called_once_with(
383 self.context, self.resource_name, 'get_all')
385 def test_list_index_with_limit(self):
386 fake_snap, expected_snap = self._get_fake_simple_share_group_snapshot()
387 fake_snap2, expected_snap2 = (
388 self._get_fake_simple_share_group_snapshot(
389 id="fake_id2"))
390 self.mock_object(
391 self.controller.share_group_api, 'get_all_share_group_snapshots',
392 mock.Mock(return_value=[fake_snap, fake_snap2]))
393 req = fakes.HTTPRequest.blank('/share-group-snapshots?limit=1',
394 version=self.api_version,
395 experimental=True)
396 req_context = req.environ['manila.context']
398 res_dict = self.controller.index(req)
400 res_dict['share_group_snapshots'][0].pop('links')
402 self.assertEqual(1, len(res_dict['share_group_snapshots']))
403 self.assertEqual([expected_snap], res_dict['share_group_snapshots'])
404 self.mock_policy_check.assert_called_once_with(
405 req_context, self.resource_name, 'get_all')
407 def test_list_index_with_limit_and_offset(self):
408 fake_snap, expected_snap = self._get_fake_simple_share_group_snapshot()
409 fake_snap2, expected_snap2 = (
410 self._get_fake_simple_share_group_snapshot(id="fake_id2"))
411 self.mock_object(
412 self.controller.share_group_api, 'get_all_share_group_snapshots',
413 mock.Mock(return_value=[fake_snap, fake_snap2]))
414 req = fakes.HTTPRequest.blank(
415 '/share-group-snapshots?limit=1&offset=1',
416 version=self.api_version, experimental=True)
417 req_context = req.environ['manila.context']
419 res_dict = self.controller.index(req)
421 res_dict['share_group_snapshots'][0].pop('links')
423 self.assertEqual(1, len(res_dict['share_group_snapshots']))
424 self.assertEqual([expected_snap2], res_dict['share_group_snapshots'])
425 self.mock_policy_check.assert_called_once_with(
426 req_context, self.resource_name, 'get_all')
428 @ddt.data({'microversion': '2.31', 'experimental': True},
429 {'microversion': SG_GRADUATION_VERSION, 'experimental': False})
430 @ddt.unpack
431 def test_list_detail(self, microversion, experimental):
432 fake_snap, expected_snap = self._get_fake_share_group_snapshot()
433 self.mock_object(
434 self.controller.share_group_api, 'get_all_share_group_snapshots',
435 mock.Mock(return_value=[fake_snap]))
436 req, context = self._get_fake_custom_request_and_context(
437 microversion, experimental)
439 res_dict = self.controller.detail(req)
441 res_dict['share_group_snapshots'][0].pop('links')
443 self.assertEqual(1, len(res_dict['share_group_snapshots']))
444 self.assertEqual(expected_snap, res_dict['share_group_snapshots'][0])
445 self.mock_policy_check.assert_called_once_with(
446 context, self.resource_name, 'get_all')
448 def test_list_detail_no_share_groups(self):
449 self.mock_object(
450 self.controller.share_group_api, 'get_all_share_group_snapshots',
451 mock.Mock(return_value=[]))
452 res_dict = self.controller.detail(self.request)
453 self.assertEqual([], res_dict['share_group_snapshots'])
454 self.mock_policy_check.assert_called_once_with(
455 self.context, self.resource_name, 'get_all')
457 def test_list_detail_with_limit(self):
458 fake_snap, expected_snap = self._get_fake_share_group_snapshot()
459 fake_snap2, expected_snap2 = self._get_fake_share_group_snapshot(
460 id="fake_id2")
461 self.mock_object(
462 self.controller.share_group_api, 'get_all_share_group_snapshots',
463 mock.Mock(return_value=[fake_snap, fake_snap2]))
464 req = fakes.HTTPRequest.blank('/share-group-snapshots?limit=1',
465 version=self.api_version,
466 experimental=True)
467 req_context = req.environ['manila.context']
469 res_dict = self.controller.detail(req)
471 res_dict['share_group_snapshots'][0].pop('links')
473 self.assertEqual(1, len(res_dict['share_group_snapshots']))
474 self.assertEqual([expected_snap], res_dict['share_group_snapshots'])
475 self.mock_policy_check.assert_called_once_with(
476 req_context, self.resource_name, 'get_all')
478 def test_list_detail_with_limit_and_offset(self):
479 fake_snap, expected_snap = self._get_fake_share_group_snapshot()
480 fake_snap2, expected_snap2 = self._get_fake_share_group_snapshot(
481 id="fake_id2")
482 self.mock_object(
483 self.controller.share_group_api, 'get_all_share_group_snapshots',
484 mock.Mock(return_value=[fake_snap, fake_snap2]))
485 req = fakes.HTTPRequest.blank(
486 '/share-group-snapshots?limit=1&offset=1',
487 version=self.api_version, experimental=True)
488 req_context = req.environ['manila.context']
490 res_dict = self.controller.detail(req)
492 res_dict['share_group_snapshots'][0].pop('links')
494 self.assertEqual(1, len(res_dict['share_group_snapshots']))
495 self.assertEqual([expected_snap2], res_dict['share_group_snapshots'])
496 self.mock_policy_check.assert_called_once_with(
497 req_context, self.resource_name, 'get_all')
499 @ddt.data({'microversion': '2.31', 'experimental': True},
500 {'microversion': SG_GRADUATION_VERSION, 'experimental': False})
501 @ddt.unpack
502 def test_delete(self, microversion, experimental):
503 fake_snap, expected_snap = self._get_fake_share_group_snapshot()
504 self.mock_object(
505 self.controller.share_group_api, 'get_share_group_snapshot',
506 mock.Mock(return_value=fake_snap))
507 self.mock_object(
508 self.controller.share_group_api, 'delete_share_group_snapshot')
509 req, req_context = self._get_fake_custom_request_and_context(
510 microversion, experimental)
512 res = self.controller.delete(req, fake_snap['id'])
514 self.assertEqual(202, res.status_code)
515 self.mock_policy_check.assert_called_once_with(
516 req_context, self.resource_name, 'delete')
518 def test_delete_not_found(self):
519 fake_snap, expected_snap = self._get_fake_share_group_snapshot()
520 self.mock_object(
521 self.controller.share_group_api, 'get_share_group_snapshot',
522 mock.Mock(side_effect=exception.NotFound))
524 self.assertRaises(
525 webob.exc.HTTPNotFound,
526 self.controller.delete, self.request, fake_snap['id'])
528 self.mock_policy_check.assert_called_once_with(
529 self.context, self.resource_name, 'delete')
531 def test_delete_in_conflicting_status(self):
532 fake_snap, expected_snap = self._get_fake_share_group_snapshot()
533 self.mock_object(
534 self.controller.share_group_api, 'get_share_group_snapshot',
535 mock.Mock(return_value=fake_snap))
536 self.mock_object(
537 self.controller.share_group_api, 'delete_share_group_snapshot',
538 mock.Mock(side_effect=exception.InvalidShareGroupSnapshot(
539 reason='blah')))
541 self.assertRaises(
542 webob.exc.HTTPConflict,
543 self.controller.delete, self.request, fake_snap['id'])
545 self.mock_policy_check.assert_called_once_with(
546 self.context, self.resource_name, 'delete')
548 @ddt.data({'microversion': '2.31', 'experimental': True},
549 {'microversion': SG_GRADUATION_VERSION, 'experimental': False})
550 @ddt.unpack
551 def test_show(self, microversion, experimental):
552 fake_snap, expected_snap = self._get_fake_share_group_snapshot()
553 self.mock_object(
554 self.controller.share_group_api, 'get_share_group_snapshot',
555 mock.Mock(return_value=fake_snap))
556 req, req_context = self._get_fake_custom_request_and_context(
557 microversion, experimental)
559 res_dict = self.controller.show(req, fake_snap['id'])
561 res_dict['share_group_snapshot'].pop('links')
563 self.assertEqual(expected_snap, res_dict['share_group_snapshot'])
564 self.mock_policy_check.assert_called_once_with(
565 req_context, self.resource_name, 'get')
567 def test_show_share_group_not_found(self):
568 fake_snap, expected_snap = self._get_fake_share_group_snapshot()
569 self.mock_object(
570 self.controller.share_group_api, 'get_share_group_snapshot',
571 mock.Mock(side_effect=exception.NotFound))
573 self.assertRaises(
574 webob.exc.HTTPNotFound,
575 self.controller.show, self.request, fake_snap['id'])
577 self.mock_policy_check.assert_called_once_with(
578 self.context, self.resource_name, 'get')
580 def _get_context(self, role):
581 return getattr(self, '%s_context' % role)
583 def _setup_share_group_snapshot_data(self, share_group_snapshot=None,
584 version='2.31'):
585 if share_group_snapshot is None: 585 ↛ 589line 585 didn't jump to line 589 because the condition on line 585 was always true
586 share_group_snapshot = db_utils.create_share_group_snapshot(
587 'fake_id', status=constants.STATUS_AVAILABLE)
589 path = ('/v2/fake/share-group-snapshots/%s/action' %
590 share_group_snapshot['id'])
591 req = fakes.HTTPRequest.blank(path, script_name=path, version=version)
592 req.headers[wsgi.API_VERSION_REQUEST_HEADER] = version
593 return share_group_snapshot, req
595 @ddt.data(*fakes.fixture_force_delete_with_different_roles)
596 @ddt.unpack
597 def test_share_group_snapshot_force_delete_with_different_roles(
598 self, role, resp_code, version):
599 group_snap, req = self._setup_share_group_snapshot_data()
600 ctxt = self._get_context(role)
601 req.method = 'POST'
602 req.headers['content-type'] = 'application/json'
603 action_name = 'force_delete'
604 body = {action_name: {'status': constants.STATUS_ERROR}}
605 req.body = jsonutils.dumps(body).encode("utf-8")
606 req.headers['X-Openstack-Manila-Api-Version'] = self.api_version
607 req.headers['X-Openstack-Manila-Api-Experimental'] = True
608 req.environ['manila.context'] = ctxt
610 with mock.patch.object(
611 policy, 'check_policy', fakes.mock_fake_admin_check):
612 resp = req.get_response(fakes.app())
614 # Validate response
615 self.assertEqual(resp_code, resp.status_int)
617 @ddt.data({'microversion': '2.31', 'experimental': True},
618 {'microversion': SG_GRADUATION_VERSION, 'experimental': False})
619 @ddt.unpack
620 def test__force_delete_call(self, microversion, experimental):
621 self.mock_object(self.controller, '_force_delete')
622 req, _junk = self._get_fake_custom_request_and_context(
623 microversion, experimental)
624 sg_id = 'fake'
625 body = {'force_delete': {}}
627 self.controller.share_group_snapshot_force_delete(req, sg_id, body)
628 self.controller._force_delete.assert_called_once_with(req, sg_id, body)
630 @ddt.data(*fakes.fixture_reset_status_with_different_roles)
631 @ddt.unpack
632 def test_share_group_snapshot_reset_status_with_different_roles(
633 self, role, valid_code, valid_status, version):
634 ctxt = self._get_context(role)
635 group_snap, req = self._setup_share_group_snapshot_data()
636 action_name = 'reset_status'
637 body = {action_name: {'status': constants.STATUS_ERROR}}
638 req.method = 'POST'
639 req.headers['content-type'] = 'application/json'
640 req.body = jsonutils.dumps(body).encode("utf-8")
641 req.headers['X-Openstack-Manila-Api-Version'] = self.api_version
642 req.headers['X-Openstack-Manila-Api-Experimental'] = True
643 req.environ['manila.context'] = ctxt
645 with mock.patch.object(
646 policy, 'check_policy', fakes.mock_fake_admin_check):
647 resp = req.get_response(fakes.app())
649 # Validate response code and model status
650 self.assertEqual(valid_code, resp.status_int)
652 actual_model = db.share_group_snapshot_get(ctxt, group_snap['id'])
653 self.assertEqual(valid_status, actual_model['status'])
655 @ddt.data({'microversion': '2.31', 'experimental': True},
656 {'microversion': SG_GRADUATION_VERSION, 'experimental': False})
657 @ddt.unpack
658 def test__reset_status_call(self, microversion, experimental):
659 self.mock_object(self.controller, '_reset_status')
660 req, _junk = self._get_fake_custom_request_and_context(
661 microversion, experimental)
662 sg_id = 'fake'
663 body = {'reset_status': {'status': constants.STATUS_ERROR}}
665 self.controller.share_group_snapshot_reset_status(req, sg_id, body)
666 self.controller._reset_status.assert_called_once_with(req, sg_id, body)