"Fossies" - the Fresh Open Source Software Archive

Member "cinder-14.0.2/cinder/tests/unit/api/v3/test_group_snapshots.py" (4 Oct 2019, 29822 Bytes) of package /linux/misc/openstack/cinder-14.0.2.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file.

    1 # Copyright (C) 2016 EMC Corporation.
    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.
   15 
   16 """
   17 Tests for group_snapshot code.
   18 """
   19 
   20 import ddt
   21 import mock
   22 from oslo_policy import policy as oslo_policy
   23 from six.moves import http_client
   24 import webob
   25 
   26 from cinder.api import microversions as mv
   27 from cinder.api.v3 import group_snapshots as v3_group_snapshots
   28 from cinder import context
   29 from cinder import db
   30 from cinder import exception
   31 from cinder.group import api as group_api
   32 from cinder import objects
   33 from cinder.objects import fields
   34 from cinder.policies import base as base_policy
   35 from cinder.policies import group_snapshots as group_snapshots_policy
   36 from cinder import policy
   37 from cinder import test
   38 from cinder.tests.unit.api import fakes
   39 from cinder.tests.unit import fake_constants as fake
   40 from cinder.tests.unit import utils
   41 import cinder.volume
   42 
   43 
   44 @ddt.ddt
   45 class GroupSnapshotsAPITestCase(test.TestCase):
   46     """Test Case for group_snapshots API."""
   47 
   48     def setUp(self):
   49         super(GroupSnapshotsAPITestCase, self).setUp()
   50         self.controller = v3_group_snapshots.GroupSnapshotsController()
   51         self.volume_api = cinder.volume.API()
   52         self.context = context.get_admin_context()
   53         self.context.project_id = fake.PROJECT_ID
   54         self.context.user_id = fake.USER_ID
   55         self.user_ctxt = context.RequestContext(
   56             fake.USER_ID, fake.PROJECT_ID, auth_token=True)
   57         self.group = utils.create_group(self.context,
   58                                         group_type_id=fake.GROUP_TYPE_ID,
   59                                         volume_type_ids=[fake.VOLUME_TYPE_ID])
   60         self.volume = utils.create_volume(self.context,
   61                                           group_id=self.group.id,
   62                                           volume_type_id=fake.VOLUME_TYPE_ID)
   63         self.g_snapshots_array = [
   64             utils.create_group_snapshot(
   65                 self.context,
   66                 group_id=self.group.id,
   67                 group_type_id=self.group.group_type_id) for _ in range(3)]
   68         self.addCleanup(self._cleanup)
   69 
   70     def _cleanup(self):
   71         for snapshot in self.g_snapshots_array:
   72             snapshot.destroy()
   73         self.volume.destroy()
   74         self.group.destroy()
   75 
   76     def test_show_group_snapshot(self):
   77         group_snapshot = utils.create_group_snapshot(
   78             self.context, group_id=self.group.id)
   79         req = fakes.HTTPRequest.blank('/v3/%s/group_snapshots/%s' %
   80                                       (fake.PROJECT_ID, group_snapshot.id),
   81                                       version=mv.GROUP_SNAPSHOTS)
   82         res_dict = self.controller.show(req, group_snapshot.id)
   83 
   84         self.assertEqual(1, len(res_dict))
   85         self.assertEqual('this is a test group snapshot',
   86                          res_dict['group_snapshot']['description'])
   87         self.assertEqual('test_group_snapshot',
   88                          res_dict['group_snapshot']['name'])
   89         self.assertEqual(fields.GroupSnapshotStatus.CREATING,
   90                          res_dict['group_snapshot']['status'])
   91 
   92         group_snapshot.destroy()
   93 
   94     @ddt.data(True, False)
   95     def test_list_group_snapshots_with_limit(self, is_detail):
   96 
   97         url = '/v3/%s/group_snapshots?limit=1' % fake.PROJECT_ID
   98         if is_detail:
   99             url = '/v3/%s/group_snapshots/detail?limit=1' % fake.PROJECT_ID
  100         req = fakes.HTTPRequest.blank(url,
  101                                       version=mv.GROUP_SNAPSHOT_PAGINATION)
  102         if is_detail:
  103             res_dict = self.controller.detail(req)
  104         else:
  105             res_dict = self.controller.index(req)
  106 
  107         self.assertEqual(2, len(res_dict))
  108         self.assertEqual(1, len(res_dict['group_snapshots']))
  109         self.assertEqual(self.g_snapshots_array[2].id,
  110                          res_dict['group_snapshots'][0]['id'])
  111         next_link = (
  112             'http://localhost/v3/%s/group_snapshots?limit='
  113             '1&marker=%s' %
  114             (fake.PROJECT_ID, res_dict['group_snapshots'][0]['id']))
  115         self.assertEqual(next_link,
  116                          res_dict['group_snapshot_links'][0]['href'])
  117         if is_detail:
  118             self.assertIn('description', res_dict['group_snapshots'][0].keys())
  119         else:
  120             self.assertNotIn('description',
  121                              res_dict['group_snapshots'][0].keys())
  122 
  123     @ddt.data(True, False)
  124     def test_list_group_snapshot_with_offset(self, is_detail):
  125         url = '/v3/%s/group_snapshots?offset=1' % fake.PROJECT_ID
  126         if is_detail:
  127             url = '/v3/%s/group_snapshots/detail?offset=1' % fake.PROJECT_ID
  128         req = fakes.HTTPRequest.blank(url,
  129                                       version=mv.GROUP_SNAPSHOT_PAGINATION)
  130         if is_detail:
  131             res_dict = self.controller.detail(req)
  132         else:
  133             res_dict = self.controller.index(req)
  134         self.assertEqual(1, len(res_dict))
  135         self.assertEqual(2, len(res_dict['group_snapshots']))
  136         self.assertEqual(self.g_snapshots_array[1].id,
  137                          res_dict['group_snapshots'][0]['id'])
  138         self.assertEqual(self.g_snapshots_array[0].id,
  139                          res_dict['group_snapshots'][1]['id'])
  140         if is_detail:
  141             self.assertIn('description', res_dict['group_snapshots'][0].keys())
  142         else:
  143             self.assertNotIn('description',
  144                              res_dict['group_snapshots'][0].keys())
  145 
  146     @ddt.data(True, False)
  147     def test_list_group_snapshot_with_offset_out_of_range(self, is_detail):
  148         url = ('/v3/%s/group_snapshots?offset=234523423455454' %
  149                fake.PROJECT_ID)
  150         if is_detail:
  151             url = ('/v3/%s/group_snapshots/detail?offset=234523423455454' %
  152                    fake.PROJECT_ID)
  153         req = fakes.HTTPRequest.blank(url,
  154                                       version=mv.GROUP_SNAPSHOT_PAGINATION)
  155         if is_detail:
  156             self.assertRaises(webob.exc.HTTPBadRequest, self.controller.detail,
  157                               req)
  158         else:
  159             self.assertRaises(webob.exc.HTTPBadRequest, self.controller.index,
  160                               req)
  161 
  162     @ddt.data(False, True)
  163     def test_list_group_snapshot_with_limit_and_offset(self, is_detail):
  164         group_snapshot = utils.create_group_snapshot(
  165             self.context,
  166             group_id=self.group.id,
  167             group_type_id=self.group.group_type_id)
  168         url = '/v3/%s/group_snapshots?limit=2&offset=1' % fake.PROJECT_ID
  169         if is_detail:
  170             url = ('/v3/%s/group_snapshots/detail?limit=2&offset=1' %
  171                    fake.PROJECT_ID)
  172         req = fakes.HTTPRequest.blank(url,
  173                                       version=mv.GROUP_SNAPSHOT_PAGINATION)
  174         if is_detail:
  175             res_dict = self.controller.detail(req)
  176         else:
  177             res_dict = self.controller.index(req)
  178 
  179         self.assertEqual(2, len(res_dict))
  180         self.assertEqual(2, len(res_dict['group_snapshots']))
  181         self.assertEqual(self.g_snapshots_array[2].id,
  182                          res_dict['group_snapshots'][0]['id'])
  183         self.assertEqual(self.g_snapshots_array[1].id,
  184                          res_dict['group_snapshots'][1]['id'])
  185         self.assertIsNotNone(res_dict['group_snapshot_links'][0]['href'])
  186         if is_detail:
  187             self.assertIn('description', res_dict['group_snapshots'][0].keys())
  188         else:
  189             self.assertNotIn('description',
  190                              res_dict['group_snapshots'][0].keys())
  191         group_snapshot.destroy()
  192 
  193     @ddt.data(mv.get_prior_version(mv.RESOURCE_FILTER),
  194               mv.RESOURCE_FILTER,
  195               mv.LIKE_FILTER)
  196     @mock.patch('cinder.api.common.reject_invalid_filters')
  197     def test_group_snapshot_list_with_general_filter(self,
  198                                                      version, mock_update):
  199         url = '/v3/%s/group_snapshots' % fake.PROJECT_ID
  200         req = fakes.HTTPRequest.blank(url,
  201                                       version=version,
  202                                       use_admin_context=False)
  203         self.controller.index(req)
  204 
  205         if version != mv.get_prior_version(mv.RESOURCE_FILTER):
  206             support_like = True if version == mv.LIKE_FILTER else False
  207             mock_update.assert_called_once_with(req.environ['cinder.context'],
  208                                                 mock.ANY, 'group_snapshot',
  209                                                 support_like)
  210 
  211     @ddt.data(False, True)
  212     def test_list_group_snapshot_with_filter(self, is_detail):
  213         url = ('/v3/%s/group_snapshots?'
  214                'all_tenants=True&id=%s') % (fake.PROJECT_ID,
  215                                             self.g_snapshots_array[0].id)
  216         if is_detail:
  217             url = ('/v3/%s/group_snapshots/detail?'
  218                    'all_tenants=True&id=%s') % (fake.PROJECT_ID,
  219                                                 self.g_snapshots_array[0].id)
  220         req = fakes.HTTPRequest.blank(url,
  221                                       version=mv.GROUP_SNAPSHOT_PAGINATION,
  222                                       use_admin_context=True)
  223         if is_detail:
  224             res_dict = self.controller.detail(req)
  225         else:
  226             res_dict = self.controller.index(req)
  227 
  228         self.assertEqual(1, len(res_dict))
  229         self.assertEqual(1, len(res_dict['group_snapshots']))
  230         self.assertEqual(self.g_snapshots_array[0].id,
  231                          res_dict['group_snapshots'][0]['id'])
  232         if is_detail:
  233             self.assertIn('description', res_dict['group_snapshots'][0].keys())
  234         else:
  235             self.assertNotIn('description',
  236                              res_dict['group_snapshots'][0].keys())
  237 
  238     @ddt.data({'is_detail': True, 'version': mv.GROUP_SNAPSHOTS},
  239               {'is_detail': False, 'version': mv.GROUP_SNAPSHOTS},
  240               {'is_detail': True, 'version': mv.POOL_FILTER},
  241               {'is_detail': False, 'version': mv.POOL_FILTER},)
  242     @ddt.unpack
  243     def test_list_group_snapshot_with_filter_previous_version(self, is_detail,
  244                                                               version):
  245         url = ('/v3/%s/group_snapshots?'
  246                'all_tenants=True&id=%s') % (fake.PROJECT_ID,
  247                                             self.g_snapshots_array[0].id)
  248         if is_detail:
  249             url = ('/v3/%s/group_snapshots/detail?'
  250                    'all_tenants=True&id=%s') % (fake.PROJECT_ID,
  251                                                 self.g_snapshots_array[0].id)
  252         req = fakes.HTTPRequest.blank(url, version=version,
  253                                       use_admin_context=True)
  254 
  255         if is_detail:
  256             res_dict = self.controller.detail(req)
  257         else:
  258             res_dict = self.controller.index(req)
  259 
  260         self.assertEqual(1, len(res_dict))
  261         self.assertEqual(3, len(res_dict['group_snapshots']))
  262 
  263     @ddt.data(False, True)
  264     def test_list_group_snapshot_with_sort(self, is_detail):
  265         url = '/v3/%s/group_snapshots?sort=id:asc' % fake.PROJECT_ID
  266         if is_detail:
  267             url = ('/v3/%s/group_snapshots/detail?sort=id:asc' %
  268                    fake.PROJECT_ID)
  269         req = fakes.HTTPRequest.blank(url,
  270                                       version=mv.GROUP_SNAPSHOT_PAGINATION)
  271         expect_result = [snapshot.id for snapshot in self.g_snapshots_array]
  272         expect_result.sort()
  273         if is_detail:
  274             res_dict = self.controller.detail(req)
  275         else:
  276             res_dict = self.controller.index(req)
  277         self.assertEqual(1, len(res_dict))
  278         self.assertEqual(3, len(res_dict['group_snapshots']))
  279         self.assertEqual(expect_result[0],
  280                          res_dict['group_snapshots'][0]['id'])
  281         self.assertEqual(expect_result[1],
  282                          res_dict['group_snapshots'][1]['id'])
  283         self.assertEqual(expect_result[2],
  284                          res_dict['group_snapshots'][2]['id'])
  285         if is_detail:
  286             self.assertIn('description', res_dict['group_snapshots'][0].keys())
  287         else:
  288             self.assertNotIn('description',
  289                              res_dict['group_snapshots'][0].keys())
  290 
  291     def test_show_group_snapshot_with_group_snapshot_not_found(self):
  292         req = fakes.HTTPRequest.blank('/v3/%s/group_snapshots/%s' %
  293                                       (fake.PROJECT_ID,
  294                                        fake.WILL_NOT_BE_FOUND_ID),
  295                                       version=mv.GROUP_SNAPSHOTS)
  296         self.assertRaises(exception.GroupSnapshotNotFound,
  297                           self.controller.show,
  298                           req, fake.WILL_NOT_BE_FOUND_ID)
  299 
  300     def test_show_group_snapshot_with_project_id(self):
  301         group_snapshot = utils.create_group_snapshot(
  302             self.context, group_id=self.group.id)
  303         req = fakes.HTTPRequest.blank(
  304             '/v3/%s/group_snapshots/%s' % (fake.PROJECT_ID,
  305                                            group_snapshot.id),
  306             version=mv.GROUP_GROUPSNAPSHOT_PROJECT_ID,
  307             use_admin_context=True)
  308         res_dict = self.controller.show(req, group_snapshot.id)
  309 
  310         self.assertEqual(1, len(res_dict))
  311         self.assertEqual('test_group_snapshot',
  312                          res_dict['group_snapshot']['name'])
  313         self.assertEqual(fake.PROJECT_ID,
  314                          res_dict['group_snapshot']['project_id'])
  315 
  316         group_snapshot.destroy()
  317 
  318     def test_show_group_snapshot_without_project_id(self):
  319         group_snapshot = utils.create_group_snapshot(
  320             self.context, group_id=self.group.id)
  321         # using mv.TRANSFER_WITH_HISTORY (3.57) to test the
  322         # project_id field is not in response before mv 3.58
  323         req = fakes.HTTPRequest.blank(
  324             '/v3/%s/group_snapshots/%s' % (fake.PROJECT_ID,
  325                                            group_snapshot.id),
  326             version=mv.TRANSFER_WITH_HISTORY,
  327             use_admin_context=True)
  328         res_dict = self.controller.show(req, group_snapshot.id)
  329 
  330         self.assertEqual(1, len(res_dict))
  331         self.assertEqual('test_group_snapshot',
  332                          res_dict['group_snapshot']['name'])
  333         self.assertNotIn('project_id', res_dict['group_snapshot'])
  334 
  335         group_snapshot.destroy()
  336 
  337     @ddt.data(True, False)
  338     def test_list_group_snapshots_json(self, is_detail):
  339         if is_detail:
  340             request_url = '/v3/%s/group_snapshots/detail'
  341         else:
  342             request_url = '/v3/%s/group_snapshots'
  343         req = fakes.HTTPRequest.blank(request_url % fake.PROJECT_ID,
  344                                       version=mv.GROUP_SNAPSHOTS)
  345         if is_detail:
  346             res_dict = self.controller.detail(req)
  347         else:
  348             res_dict = self.controller.index(req)
  349 
  350         self.assertEqual(1, len(res_dict))
  351         self.assertEqual(3, len(res_dict['group_snapshots']))
  352         for index, snapshot in enumerate(self.g_snapshots_array):
  353             self.assertEqual(snapshot.id,
  354                              res_dict['group_snapshots'][2 - index]['id'])
  355             self.assertIsNotNone(
  356                 res_dict['group_snapshots'][2 - index]['name'])
  357             if is_detail:
  358                 self.assertIn('description',
  359                               res_dict['group_snapshots'][2 - index].keys())
  360             else:
  361                 self.assertNotIn('description',
  362                                  res_dict['group_snapshots'][2 - index].keys())
  363 
  364     @ddt.data(True, False)
  365     def test_list_group_snapshots_with_project_id(self, is_detail):
  366         if is_detail:
  367             request_url = '/v3/%s/group_snapshots/detail'
  368         else:
  369             request_url = '/v3/%s/group_snapshots'
  370         req = fakes.HTTPRequest.blank(
  371             request_url % fake.PROJECT_ID,
  372             version=mv.GROUP_GROUPSNAPSHOT_PROJECT_ID,
  373             use_admin_context=True)
  374         if is_detail:
  375             res_dict = self.controller.detail(req)
  376         else:
  377             res_dict = self.controller.index(req)
  378 
  379         self.assertEqual(1, len(res_dict))
  380         self.assertEqual(3, len(res_dict['group_snapshots']))
  381         for group in res_dict['group_snapshots']:
  382             if is_detail:
  383                 self.assertIsNotNone(group['project_id'])
  384             else:
  385                 self.assertNotIn('project_id', group)
  386 
  387     @mock.patch('cinder.db.volume_type_get')
  388     @mock.patch('cinder.quota.VolumeTypeQuotaEngine.reserve')
  389     def test_create_group_snapshot_json(self, mock_quota, mock_vol_type):
  390         body = {"group_snapshot": {"name": "group_snapshot1",
  391                                    "description":
  392                                    "Group Snapshot 1",
  393                                    "group_id": self.group.id}}
  394         req = fakes.HTTPRequest.blank('/v3/%s/group_snapshots' %
  395                                       fake.PROJECT_ID,
  396                                       version=mv.GROUP_SNAPSHOTS)
  397         res_dict = self.controller.create(req, body=body)
  398 
  399         self.assertEqual(1, len(res_dict))
  400         self.assertIn('id', res_dict['group_snapshot'])
  401         group_snapshot = objects.GroupSnapshot.get_by_id(
  402             context.get_admin_context(), res_dict['group_snapshot']['id'])
  403         group_snapshot.destroy()
  404 
  405     @mock.patch('cinder.db.volume_type_get')
  406     def test_create_group_snapshot_when_volume_in_error_status(
  407             self, mock_vol_type):
  408         group = utils.create_group(
  409             self.context,
  410             group_type_id=fake.GROUP_TYPE_ID,
  411             volume_type_ids=[fake.VOLUME_TYPE_ID],)
  412         volume_id = utils.create_volume(
  413             self.context,
  414             status='error',
  415             group_id=group.id,
  416             volume_type_id=fake.VOLUME_TYPE_ID)['id']
  417         body = {"group_snapshot": {"name": "group_snapshot1",
  418                                    "description":
  419                                    "Group Snapshot 1",
  420                                    "group_id": group.id}}
  421         req = fakes.HTTPRequest.blank('/v3/%s/group_snapshots' %
  422                                       fake.PROJECT_ID,
  423                                       version=mv.GROUP_SNAPSHOTS)
  424         self.assertRaises(webob.exc.HTTPBadRequest, self.controller.create,
  425                           req, body=body)
  426 
  427         group.destroy()
  428         db.volume_destroy(context.get_admin_context(),
  429                           volume_id)
  430 
  431     def test_create_group_snapshot_with_no_body(self):
  432         # omit body from the request
  433         req = fakes.HTTPRequest.blank('/v3/%s/group_snapshots' %
  434                                       fake.PROJECT_ID,
  435                                       version=mv.GROUP_SNAPSHOTS)
  436         self.assertRaises(exception.ValidationError, self.controller.create,
  437                           req, body=None)
  438 
  439     def test_create_group_snapshot_with_empty_body(self):
  440         # empty body in the request
  441         req = fakes.HTTPRequest.blank('/v3/%s/group_snapshots' %
  442                                       fake.PROJECT_ID,
  443                                       version=mv.GROUP_SNAPSHOTS)
  444         body = {"group_snapshot": {}}
  445         self.assertRaises(exception.ValidationError, self.controller.create,
  446                           req, body=body)
  447 
  448     @mock.patch.object(group_api.API, 'create_group_snapshot',
  449                        side_effect=exception.InvalidGroupSnapshot(
  450                            reason='Invalid group snapshot'))
  451     def test_create_with_invalid_group_snapshot(self, mock_create_group_snap):
  452         body = {"group_snapshot": {"name": "group_snapshot1",
  453                                    "description":
  454                                    "Group Snapshot 1",
  455                                    "group_id": self.group.id}}
  456         req = fakes.HTTPRequest.blank('/v3/%s/group_snapshots' %
  457                                       fake.PROJECT_ID,
  458                                       version=mv.GROUP_SNAPSHOTS)
  459         self.assertRaises(webob.exc.HTTPBadRequest, self.controller.create,
  460                           req, body=body)
  461 
  462     @mock.patch.object(group_api.API, 'create_group_snapshot',
  463                        side_effect=exception.GroupSnapshotNotFound(
  464                            group_snapshot_id='invalid_id'))
  465     def test_create_with_group_snapshot_not_found(self, mock_create_grp_snap):
  466         body = {"group_snapshot": {"name": "group_snapshot1",
  467                                    "description":
  468                                    "Group Snapshot 1",
  469                                    "group_id": self.group.id}}
  470         req = fakes.HTTPRequest.blank('/v3/%s/group_snapshots' %
  471                                       fake.PROJECT_ID,
  472                                       version=mv.GROUP_SNAPSHOTS)
  473         self.assertRaises(exception.GroupSnapshotNotFound,
  474                           self.controller.create,
  475                           req, body=body)
  476 
  477     def test_create_group_snapshot_from_empty_group(self):
  478         empty_group = utils.create_group(
  479             self.context,
  480             group_type_id=fake.GROUP_TYPE_ID,
  481             volume_type_ids=[fake.VOLUME_TYPE_ID])
  482         body = {"group_snapshot": {"name": "group_snapshot1",
  483                                    "description":
  484                                    "Group Snapshot 1",
  485                                    "group_id": empty_group.id}}
  486         req = fakes.HTTPRequest.blank('/v3/%s/group_snapshots' %
  487                                       fake.PROJECT_ID,
  488                                       version=mv.GROUP_SNAPSHOTS)
  489 
  490         self.assertRaises(webob.exc.HTTPBadRequest, self.controller.create,
  491                           req, body=body)
  492         empty_group.destroy()
  493 
  494     def test_delete_group_snapshot_available(self):
  495         group_snapshot = utils.create_group_snapshot(
  496             self.context,
  497             group_id=self.group.id,
  498             status=fields.GroupSnapshotStatus.AVAILABLE)
  499         req = fakes.HTTPRequest.blank('/v3/%s/group_snapshots/%s' %
  500                                       (fake.PROJECT_ID, group_snapshot.id),
  501                                       version=mv.GROUP_SNAPSHOTS)
  502         res_dict = self.controller.delete(req, group_snapshot.id)
  503 
  504         group_snapshot = objects.GroupSnapshot.get_by_id(self.context,
  505                                                          group_snapshot.id)
  506         self.assertEqual(http_client.ACCEPTED, res_dict.status_int)
  507         self.assertEqual(fields.GroupSnapshotStatus.DELETING,
  508                          group_snapshot.status)
  509 
  510         group_snapshot.destroy()
  511 
  512     def test_delete_group_snapshot_available_used_as_source(self):
  513         group_snapshot = utils.create_group_snapshot(
  514             self.context,
  515             group_id=self.group.id,
  516             status=fields.GroupSnapshotStatus.AVAILABLE)
  517 
  518         group2 = utils.create_group(
  519             self.context, status='creating',
  520             group_snapshot_id=group_snapshot.id,
  521             group_type_id=fake.GROUP_TYPE_ID,
  522             volume_type_ids=[fake.VOLUME_TYPE_ID],)
  523         req = fakes.HTTPRequest.blank('/v3/%s/group_snapshots/%s' %
  524                                       (fake.PROJECT_ID, group_snapshot.id),
  525                                       version=mv.GROUP_SNAPSHOTS)
  526         self.assertRaises(webob.exc.HTTPBadRequest, self.controller.delete,
  527                           req, group_snapshot.id)
  528 
  529         group_snapshot.destroy()
  530         group2.destroy()
  531 
  532     def test_delete_group_snapshot_with_group_snapshot_NotFound(self):
  533         req = fakes.HTTPRequest.blank('/v3/%s/group_snapshots/%s' %
  534                                       (fake.PROJECT_ID,
  535                                        fake.WILL_NOT_BE_FOUND_ID),
  536                                       version=mv.GROUP_SNAPSHOTS)
  537         self.assertRaises(exception.GroupSnapshotNotFound,
  538                           self.controller.delete,
  539                           req, fake.WILL_NOT_BE_FOUND_ID)
  540 
  541     def test_delete_group_snapshot_with_invalid_group_snapshot(self):
  542         group_snapshot = utils.create_group_snapshot(
  543             self.context,
  544             group_id=self.group.id,
  545             status='invalid')
  546         req = fakes.HTTPRequest.blank('/v3/%s/group_snapshots/%s' %
  547                                       (fake.PROJECT_ID, group_snapshot.id),
  548                                       version=mv.GROUP_SNAPSHOTS)
  549         self.assertRaises(webob.exc.HTTPBadRequest, self.controller.delete,
  550                           req, group_snapshot.id)
  551 
  552         group_snapshot.destroy()
  553 
  554     def test_delete_group_snapshot_policy_not_authorized(self):
  555         group_snapshot = utils.create_group_snapshot(
  556             self.context,
  557             group_id=self.group.id,
  558             status=fields.GroupSnapshotStatus.AVAILABLE)
  559 
  560         req = fakes.HTTPRequest.blank('/v3/%s/group_snapshots/%s/' %
  561                                       (fake.PROJECT_ID, group_snapshot.id),
  562                                       version=mv.GROUP_SNAPSHOTS,
  563                                       use_admin_context=False)
  564 
  565         rules = {
  566             group_snapshots_policy.DELETE_POLICY: base_policy.RULE_ADMIN_API
  567         }
  568         policy.set_rules(oslo_policy.Rules.from_dict(rules))
  569         self.addCleanup(policy.reset)
  570 
  571         self.assertRaises(exception.PolicyNotAuthorized,
  572                           self.controller.delete,
  573                           req, group_snapshot.id)
  574 
  575     @ddt.data((mv.GROUP_TYPE, 'fake_snapshot_001',
  576                fields.GroupSnapshotStatus.AVAILABLE,
  577                exception.VersionNotFoundForAPIMethod),
  578               (mv.get_prior_version(mv.GROUP_SNAPSHOT_RESET_STATUS),
  579                'fake_snapshot_001',
  580                fields.GroupSnapshotStatus.AVAILABLE,
  581                exception.VersionNotFoundForAPIMethod),
  582               (mv.GROUP_SNAPSHOT_RESET_STATUS, 'fake_snapshot_001',
  583                fields.GroupSnapshotStatus.AVAILABLE,
  584                exception.GroupSnapshotNotFound))
  585     @ddt.unpack
  586     def test_reset_group_snapshot_status_illegal(self, version,
  587                                                  group_snapshot_id,
  588                                                  status, exceptions):
  589         req = fakes.HTTPRequest.blank('/v3/%s/group_snapshots/%s/action' %
  590                                       (fake.PROJECT_ID, group_snapshot_id),
  591                                       version=version)
  592         body = {"reset_status": {
  593             "status": status
  594         }}
  595         self.assertRaises(exceptions,
  596                           self.controller.reset_status,
  597                           req, group_snapshot_id, body=body)
  598 
  599     def test_reset_group_snapshot_status_invalid_status(self):
  600         group_snapshot = utils.create_group_snapshot(
  601             self.context,
  602             group_id=self.group.id,
  603             status=fields.GroupSnapshotStatus.CREATING)
  604         req = fakes.HTTPRequest.blank('/v3/%s/group_snapshots/%s/action' %
  605                                       (fake.PROJECT_ID, group_snapshot.id),
  606                                       version=mv.GROUP_SNAPSHOT_RESET_STATUS)
  607         body = {"reset_status": {
  608             "status": "invalid_test_status"
  609         }}
  610         self.assertRaises(exception.InvalidGroupSnapshotStatus,
  611                           self.controller.reset_status,
  612                           req, group_snapshot.id, body=body)
  613         group_snapshot.destroy()
  614 
  615     def test_reset_group_snapshot_status(self):
  616         group_snapshot = utils.create_group_snapshot(
  617             self.context,
  618             group_id=self.group.id,
  619             status=fields.GroupSnapshotStatus.CREATING)
  620         req = fakes.HTTPRequest.blank('/v3/%s/group_snapshots/%s/action' %
  621                                       (fake.PROJECT_ID, group_snapshot.id),
  622                                       version=mv.GROUP_SNAPSHOT_RESET_STATUS)
  623         body = {"reset_status": {
  624             "status": fields.GroupSnapshotStatus.AVAILABLE
  625         }}
  626         response = self.controller.reset_status(req, group_snapshot.id,
  627                                                 body=body)
  628 
  629         g_snapshot = objects.GroupSnapshot.get_by_id(self.context,
  630                                                      group_snapshot.id)
  631         self.assertEqual(http_client.ACCEPTED, response.status_int)
  632         self.assertEqual(fields.GroupSnapshotStatus.AVAILABLE,
  633                          g_snapshot.status)
  634         group_snapshot.destroy()
  635 
  636     @mock.patch('cinder.db.volume_type_get')
  637     @mock.patch('cinder.quota.VolumeTypeQuotaEngine.reserve')
  638     def test_create_group_snapshot_with_null_validate(
  639             self, mock_quota, mock_vol_type):
  640         body = {"group_snapshot": {"name": None,
  641                                    "description": None,
  642                                    "group_id": self.group.id}}
  643         req = fakes.HTTPRequest.blank('/v3/%s/group_snapshots' %
  644                                       self.context.project_id,
  645                                       version=mv.GROUP_SNAPSHOTS)
  646         res_dict = self.controller.create(req, body=body)
  647 
  648         self.assertIn('group_snapshot', res_dict)
  649         self.assertIsNone(res_dict['group_snapshot']['name'])
  650         group_snapshot = objects.GroupSnapshot.get_by_id(
  651             context.get_admin_context(), res_dict['group_snapshot']['id'])
  652         group_snapshot.destroy()