"Fossies" - the Fresh Open Source Software Archive

Member "cinder-13.0.7/cinder/tests/unit/api/v2/test_volumes.py" (4 Oct 2019, 79278 Bytes) of package /linux/misc/openstack/cinder-13.0.7.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. See also the last Fossies "Diffs" side-by-side code changes report for "test_volumes.py": 14.0.2_vs_15.0.0.

    1 # Copyright 2013 Josh Durgin
    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 import datetime
   18 import iso8601
   19 
   20 import ddt
   21 import mock
   22 from oslo_config import cfg
   23 import six
   24 from six.moves import http_client
   25 from six.moves import range
   26 from six.moves import urllib
   27 import webob
   28 
   29 from cinder.api import common
   30 from cinder.api import extensions
   31 from cinder.api.v2 import volumes
   32 from cinder import context
   33 from cinder import db
   34 from cinder import exception
   35 from cinder import group as groupAPI
   36 from cinder import objects
   37 from cinder.objects import fields
   38 from cinder import test
   39 from cinder.tests.unit.api import fakes
   40 from cinder.tests.unit.api.v2 import fakes as v2_fakes
   41 from cinder.tests.unit import fake_constants as fake
   42 from cinder.tests.unit import fake_volume
   43 from cinder.tests.unit.image import fake as fake_image
   44 from cinder.tests.unit import utils
   45 from cinder.volume import api as volume_api
   46 
   47 CONF = cfg.CONF
   48 
   49 NS = '{http://docs.openstack.org/api/openstack-block-storage/2.0/content}'
   50 
   51 DEFAULT_AZ = "zone1:host1"
   52 
   53 
   54 @ddt.ddt
   55 class VolumeApiTest(test.TestCase):
   56     def setUp(self):
   57         super(VolumeApiTest, self).setUp()
   58         self.ext_mgr = extensions.ExtensionManager()
   59         self.ext_mgr.extensions = {}
   60         fake_image.mock_image_service(self)
   61         self.controller = volumes.VolumeController(self.ext_mgr)
   62         self.maxDiff = None
   63         self.ctxt = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
   64 
   65     @mock.patch(
   66         'cinder.api.openstack.wsgi.Controller.validate_name_and_description')
   67     def test_volume_create(self, mock_validate):
   68         self.mock_object(volume_api.API, 'get', v2_fakes.fake_volume_get)
   69         self.mock_object(volume_api.API, "create",
   70                          v2_fakes.fake_volume_api_create)
   71         self.mock_object(db.sqlalchemy.api, '_volume_type_get_full',
   72                          v2_fakes.fake_volume_type_get)
   73 
   74         vol = self._vol_in_request_body()
   75         body = {"volume": vol}
   76         req = fakes.HTTPRequest.blank('/v2/volumes')
   77         res_dict = self.controller.create(req, body=body)
   78         ex = self._expected_vol_from_controller()
   79         self.assertEqual(ex, res_dict)
   80         self.assertTrue(mock_validate.called)
   81 
   82     @mock.patch.object(db, 'volume_get_all', v2_fakes.fake_volume_get_all)
   83     @mock.patch.object(db, 'service_get_all',
   84                        return_value=v2_fakes.fake_service_get_all_by_topic(
   85                            None, None),
   86                        autospec=True)
   87     @mock.patch(
   88         'cinder.api.openstack.wsgi.Controller.validate_name_and_description')
   89     def test_volume_create_with_type(self, mock_validate, mock_service_get):
   90         vol_type = db.volume_type_create(
   91             context.get_admin_context(),
   92             dict(name=CONF.default_volume_type, extra_specs={})
   93         )
   94 
   95         db_vol_type = db.volume_type_get(context.get_admin_context(),
   96                                          vol_type.id)
   97 
   98         vol = self._vol_in_request_body(volume_type="FakeTypeName")
   99         body = {"volume": vol}
  100         req = fakes.HTTPRequest.blank('/v2/volumes')
  101         # Raise 404 when type name isn't valid
  102         self.assertRaises(exception.VolumeTypeNotFoundByName,
  103                           self.controller.create, req, body=body)
  104 
  105         # Use correct volume type name
  106         vol = self._vol_in_request_body(volume_type=CONF.default_volume_type)
  107         body.update(dict(volume=vol))
  108         res_dict = self.controller.create(req, body=body)
  109         volume_id = res_dict['volume']['id']
  110         self.assertEqual(1, len(res_dict))
  111 
  112         # Use correct volume type id
  113         vol = self._vol_in_request_body(volume_type=db_vol_type['id'])
  114         body.update(dict(volume=vol))
  115         res_dict = self.controller.create(req, body=body)
  116         volume_id = res_dict['volume']['id']
  117         self.assertEqual(1, len(res_dict))
  118 
  119         vol_db = v2_fakes.create_fake_volume(volume_id,
  120                                              volume_type={'name': vol_type})
  121         vol_obj = fake_volume.fake_volume_obj(context.get_admin_context(),
  122                                               **vol_db)
  123         self.mock_object(volume_api.API, 'get_all',
  124                          return_value=objects.VolumeList(objects=[vol_obj]))
  125         # NOTE(geguileo): This is required because common get_by_id method in
  126         # cinder.db.sqlalchemy.api caches the real get method.
  127         db.sqlalchemy.api._GET_METHODS = {}
  128         self.mock_object(db.sqlalchemy.api, '_volume_type_get_full',
  129                          v2_fakes.fake_volume_type_get)
  130         req = fakes.HTTPRequest.blank('/v2/volumes/detail')
  131         res_dict = self.controller.detail(req)
  132         self.assertTrue(mock_validate.called)
  133 
  134     @classmethod
  135     def _vol_in_request_body(cls,
  136                              size=v2_fakes.DEFAULT_VOL_SIZE,
  137                              name=v2_fakes.DEFAULT_VOL_NAME,
  138                              description=v2_fakes.DEFAULT_VOL_DESCRIPTION,
  139                              availability_zone=DEFAULT_AZ,
  140                              snapshot_id=None,
  141                              source_volid=None,
  142                              consistencygroup_id=None,
  143                              volume_type=None,
  144                              image_ref=None,
  145                              image_id=None,
  146                              multiattach=False):
  147         vol = {"size": size,
  148                "name": name,
  149                "description": description,
  150                "availability_zone": availability_zone,
  151                "snapshot_id": snapshot_id,
  152                "source_volid": source_volid,
  153                "consistencygroup_id": consistencygroup_id,
  154                "volume_type": volume_type,
  155                "multiattach": multiattach,
  156                }
  157 
  158         if image_id is not None:
  159             vol['image_id'] = image_id
  160         elif image_ref is not None:
  161             vol['imageRef'] = image_ref
  162 
  163         return vol
  164 
  165     def _expected_vol_from_controller(
  166             self,
  167             size=v2_fakes.DEFAULT_VOL_SIZE,
  168             availability_zone=DEFAULT_AZ,
  169             description=v2_fakes.DEFAULT_VOL_DESCRIPTION,
  170             name=v2_fakes.DEFAULT_VOL_NAME,
  171             consistencygroup_id=None,
  172             source_volid=None,
  173             snapshot_id=None,
  174             metadata=None,
  175             attachments=None,
  176             volume_type=v2_fakes.DEFAULT_VOL_TYPE,
  177             status=v2_fakes.DEFAULT_VOL_STATUS,
  178             with_migration_status=False,
  179             multiattach=False):
  180         metadata = metadata or {}
  181         attachments = attachments or []
  182         volume = {'volume':
  183                   {'attachments': attachments,
  184                    'availability_zone': availability_zone,
  185                    'bootable': 'false',
  186                    'consistencygroup_id': consistencygroup_id,
  187                    'created_at': datetime.datetime(
  188                        1900, 1, 1, 1, 1, 1, tzinfo=iso8601.UTC),
  189                    'updated_at': datetime.datetime(
  190                        1900, 1, 1, 1, 1, 1, tzinfo=iso8601.UTC),
  191                    'description': description,
  192                    'id': v2_fakes.DEFAULT_VOL_ID,
  193                    'links':
  194                    [{'href': 'http://localhost/v2/%s/volumes/%s' % (
  195                              fake.PROJECT_ID, fake.VOLUME_ID),
  196                      'rel': 'self'},
  197                     {'href': 'http://localhost/%s/volumes/%s' % (
  198                              fake.PROJECT_ID, fake.VOLUME_ID),
  199                      'rel': 'bookmark'}],
  200                    'metadata': metadata,
  201                    'name': name,
  202                    'replication_status': 'disabled',
  203                    'multiattach': multiattach,
  204                    'size': size,
  205                    'snapshot_id': snapshot_id,
  206                    'source_volid': source_volid,
  207                    'status': status,
  208                    'user_id': fake.USER_ID,
  209                    'volume_type': volume_type,
  210                    'encrypted': False}}
  211 
  212         if with_migration_status:
  213             volume['volume']['migration_status'] = None
  214 
  215         return volume
  216 
  217     def _expected_volume_api_create_kwargs(self, snapshot=None,
  218                                            availability_zone=DEFAULT_AZ,
  219                                            source_volume=None):
  220         return {'metadata': None,
  221                 'snapshot': snapshot,
  222                 'source_volume': source_volume,
  223                 'group': None,
  224                 'consistencygroup': None,
  225                 'availability_zone': availability_zone,
  226                 'scheduler_hints': None,
  227                 'multiattach': False,
  228                 }
  229 
  230     @mock.patch.object(db.sqlalchemy.api, '_volume_type_get_full',
  231                        autospec=True)
  232     @mock.patch.object(volume_api.API, 'get_snapshot', autospec=True)
  233     @mock.patch.object(volume_api.API, 'create', autospec=True)
  234     def test_volume_creation_from_snapshot(self, create, get_snapshot,
  235                                            volume_type_get):
  236         create.side_effect = v2_fakes.fake_volume_api_create
  237         get_snapshot.side_effect = v2_fakes.fake_snapshot_get
  238         volume_type_get.side_effect = v2_fakes.fake_volume_type_get
  239 
  240         snapshot_id = fake.SNAPSHOT_ID
  241         vol = self._vol_in_request_body(snapshot_id=snapshot_id)
  242         body = {"volume": vol}
  243         req = fakes.HTTPRequest.blank('/v2/volumes')
  244         res_dict = self.controller.create(req, body=body)
  245 
  246         ex = self._expected_vol_from_controller(snapshot_id=snapshot_id)
  247         self.assertEqual(ex, res_dict)
  248 
  249         context = req.environ['cinder.context']
  250         get_snapshot.assert_called_once_with(self.controller.volume_api,
  251                                              context, snapshot_id)
  252 
  253         kwargs = self._expected_volume_api_create_kwargs(
  254             v2_fakes.fake_snapshot(snapshot_id))
  255         create.assert_called_once_with(self.controller.volume_api, context,
  256                                        vol['size'], v2_fakes.DEFAULT_VOL_NAME,
  257                                        v2_fakes.DEFAULT_VOL_DESCRIPTION,
  258                                        **kwargs)
  259 
  260     @mock.patch.object(volume_api.API, 'get_snapshot', autospec=True)
  261     def test_volume_creation_fails_with_invalid_snapshot(self, get_snapshot):
  262 
  263         get_snapshot.side_effect = v2_fakes.fake_snapshot_get
  264 
  265         snapshot_id = fake.WILL_NOT_BE_FOUND_ID
  266         vol = self._vol_in_request_body(snapshot_id=snapshot_id)
  267         body = {"volume": vol}
  268         req = fakes.HTTPRequest.blank('/v2/volumes')
  269         # Raise 404 when snapshot cannot be found.
  270         self.assertRaises(exception.SnapshotNotFound, self.controller.create,
  271                           req, body=body)
  272         context = req.environ['cinder.context']
  273         get_snapshot.assert_called_once_with(self.controller.volume_api,
  274                                              context, snapshot_id)
  275 
  276     @ddt.data({'s': 'ea895e29-8485-4930-bbb8-c5616a309c0e'},
  277               ['ea895e29-8485-4930-bbb8-c5616a309c0e'],
  278               42)
  279     def test_volume_creation_fails_with_invalid_snapshot_type(self, value):
  280         snapshot_id = value
  281         vol = self._vol_in_request_body(snapshot_id=snapshot_id)
  282         body = {"volume": vol}
  283         req = fakes.HTTPRequest.blank('/v2/volumes')
  284         # Raise 400 when snapshot has not uuid type.
  285         self.assertRaises(exception.ValidationError, self.controller.create,
  286                           req, body=body)
  287 
  288     @mock.patch.object(db.sqlalchemy.api, '_volume_type_get_full',
  289                        autospec=True)
  290     @mock.patch.object(volume_api.API, 'get_volume', autospec=True)
  291     @mock.patch.object(volume_api.API, 'create', autospec=True)
  292     def test_volume_creation_from_source_volume(self, create, get_volume,
  293                                                 volume_type_get):
  294         get_volume.side_effect = v2_fakes.fake_volume_api_get
  295         create.side_effect = v2_fakes.fake_volume_api_create
  296         volume_type_get.side_effect = v2_fakes.fake_volume_type_get
  297 
  298         source_volid = '2f49aa3a-6aae-488d-8b99-a43271605af6'
  299         vol = self._vol_in_request_body(source_volid=source_volid)
  300         body = {"volume": vol}
  301         req = fakes.HTTPRequest.blank('/v2/volumes')
  302         res_dict = self.controller.create(req, body=body)
  303 
  304         ex = self._expected_vol_from_controller(source_volid=source_volid)
  305         self.assertEqual(ex, res_dict)
  306 
  307         context = req.environ['cinder.context']
  308         get_volume.assert_called_once_with(self.controller.volume_api,
  309                                            context, source_volid)
  310 
  311         db_vol = v2_fakes.create_fake_volume(source_volid)
  312         vol_obj = fake_volume.fake_volume_obj(context, **db_vol)
  313         kwargs = self._expected_volume_api_create_kwargs(
  314             source_volume=vol_obj)
  315         create.assert_called_once_with(self.controller.volume_api, context,
  316                                        vol['size'], v2_fakes.DEFAULT_VOL_NAME,
  317                                        v2_fakes.DEFAULT_VOL_DESCRIPTION,
  318                                        **kwargs)
  319 
  320     @mock.patch.object(volume_api.API, 'get_volume', autospec=True)
  321     def test_volume_creation_fails_with_invalid_source_volume(self,
  322                                                               get_volume):
  323 
  324         get_volume.side_effect = v2_fakes.fake_volume_get_notfound
  325 
  326         source_volid = fake.VOLUME_ID
  327         vol = self._vol_in_request_body(source_volid=source_volid)
  328         body = {"volume": vol}
  329         req = fakes.HTTPRequest.blank('/v2/volumes')
  330         # Raise 404 when source volume cannot be found.
  331         self.assertRaises(exception.VolumeNotFound, self.controller.create,
  332                           req, body=body)
  333 
  334         context = req.environ['cinder.context']
  335         get_volume.assert_called_once_with(self.controller.volume_api,
  336                                            context, source_volid)
  337 
  338     @ddt.data({'source_volid': 1},
  339               {'source_volid': []},
  340               {'consistencygroup_id': 1},
  341               {'consistencygroup_id': []})
  342     def test_volume_creation_fails_with_invalid_uuids(self, updated_uuids):
  343         vol = self._vol_in_request_body()
  344         vol.update(updated_uuids)
  345         body = {"volume": vol}
  346         req = fakes.HTTPRequest.blank('/v2/volumes')
  347         # Raise 400 for resource requested with invalid uuids.
  348         self.assertRaises(exception.ValidationError, self.controller.create,
  349                           req, body=body)
  350 
  351     @mock.patch.object(groupAPI.API, 'get', autospec=True)
  352     def test_volume_creation_fails_with_invalid_consistency_group(self,
  353                                                                   get_cg):
  354 
  355         get_cg.side_effect = v2_fakes.fake_consistencygroup_get_notfound
  356 
  357         consistencygroup_id = '4f49aa3a-6aae-488d-8b99-a43271605af6'
  358         vol = self._vol_in_request_body(
  359             consistencygroup_id=consistencygroup_id)
  360         body = {"volume": vol}
  361         req = fakes.HTTPRequest.blank('/v2/volumes')
  362         # Raise 404 when consistency group is not found.
  363         self.assertRaises(exception.GroupNotFound,
  364                           self.controller.create, req, body=body)
  365 
  366         context = req.environ['cinder.context']
  367         get_cg.assert_called_once_with(self.controller.group_api,
  368                                        context, consistencygroup_id)
  369 
  370     def test_volume_creation_fails_with_bad_size(self):
  371         vol = self._vol_in_request_body(size="")
  372         body = {"volume": vol}
  373         req = fakes.HTTPRequest.blank('/v2/volumes')
  374         self.assertRaises(exception.ValidationError,
  375                           self.controller.create,
  376                           req,
  377                           body=body)
  378 
  379     def test_volume_creation_fails_with_bad_availability_zone(self):
  380         vol = self._vol_in_request_body(availability_zone="zonen:hostn")
  381         body = {"volume": vol}
  382         req = fakes.HTTPRequest.blank('/v2/volumes')
  383         self.assertRaises(exception.InvalidAvailabilityZone,
  384                           self.controller.create,
  385                           req, body=body)
  386 
  387     @mock.patch(
  388         'cinder.api.openstack.wsgi.Controller.validate_name_and_description')
  389     def test_volume_create_with_image_ref(self, mock_validate):
  390         self.mock_object(volume_api.API, "create",
  391                          v2_fakes.fake_volume_api_create)
  392         self.mock_object(db.sqlalchemy.api, '_volume_type_get_full',
  393                          v2_fakes.fake_volume_type_get)
  394 
  395         self.ext_mgr.extensions = {'os-image-create': 'fake'}
  396         vol = self._vol_in_request_body(
  397             availability_zone="nova",
  398             image_ref="c905cedb-7281-47e4-8a62-f26bc5fc4c77")
  399         ex = self._expected_vol_from_controller(availability_zone="nova")
  400         body = {"volume": vol}
  401         req = fakes.HTTPRequest.blank('/v2/volumes')
  402         res_dict = self.controller.create(req, body=body)
  403         self.assertEqual(ex, res_dict)
  404         self.assertTrue(mock_validate.called)
  405 
  406     def test_volume_create_with_image_ref_is_integer(self):
  407         self.mock_object(volume_api.API, "create", v2_fakes.fake_volume_create)
  408         self.ext_mgr.extensions = {'os-image-create': 'fake'}
  409         vol = self._vol_in_request_body(availability_zone="cinder",
  410                                         image_ref=1234)
  411         body = {"volume": vol}
  412         req = fakes.HTTPRequest.blank('/v2/volumes')
  413         self.assertRaises(exception.ValidationError,
  414                           self.controller.create,
  415                           req,
  416                           body=body)
  417 
  418     def test_volume_create_with_image_ref_not_uuid_format(self):
  419         self.mock_object(volume_api.API, "create", v2_fakes.fake_volume_create)
  420         self.mock_object(fake_image._FakeImageService,
  421                          "detail",
  422                          v2_fakes.fake_image_service_detail)
  423         self.ext_mgr.extensions = {'os-image-create': 'fake'}
  424         vol = self._vol_in_request_body(availability_zone="cinder",
  425                                         image_ref="12345")
  426         body = {"volume": vol}
  427         req = fakes.HTTPRequest.blank('/v2/volumes')
  428         self.assertRaises(webob.exc.HTTPBadRequest,
  429                           self.controller.create,
  430                           req,
  431                           body=body)
  432 
  433     def test_volume_create_with_image_ref_with_empty_string(self):
  434         self.mock_object(volume_api.API, "create", v2_fakes.fake_volume_create)
  435         self.mock_object(fake_image._FakeImageService,
  436                          "detail",
  437                          v2_fakes.fake_image_service_detail)
  438         self.ext_mgr.extensions = {'os-image-create': 'fake'}
  439         vol = self._vol_in_request_body(availability_zone="cinder",
  440                                         image_ref="")
  441         body = {"volume": vol}
  442         req = fakes.HTTPRequest.blank('/v2/volumes')
  443         self.assertRaises(webob.exc.HTTPBadRequest,
  444                           self.controller.create,
  445                           req,
  446                           body=body)
  447 
  448     @mock.patch(
  449         'cinder.api.openstack.wsgi.Controller.validate_name_and_description')
  450     def test_volume_create_with_image_id(self, mock_validate):
  451         self.mock_object(volume_api.API, "create",
  452                          v2_fakes.fake_volume_api_create)
  453         self.mock_object(db.sqlalchemy.api, '_volume_type_get_full',
  454                          v2_fakes.fake_volume_type_get)
  455 
  456         self.ext_mgr.extensions = {'os-image-create': 'fake'}
  457         vol = self._vol_in_request_body(
  458             availability_zone="nova",
  459             image_id="c905cedb-7281-47e4-8a62-f26bc5fc4c77")
  460         ex = self._expected_vol_from_controller(availability_zone="nova")
  461         body = {"volume": vol}
  462         req = fakes.HTTPRequest.blank('/v2/volumes')
  463         res_dict = self.controller.create(req, body=body)
  464         self.assertEqual(ex, res_dict)
  465         self.assertTrue(mock_validate.called)
  466 
  467     def test_volume_create_with_image_id_is_integer(self):
  468         self.mock_object(volume_api.API, "create", v2_fakes.fake_volume_create)
  469         self.ext_mgr.extensions = {'os-image-create': 'fake'}
  470         vol = self._vol_in_request_body(availability_zone="cinder",
  471                                         image_id=1234)
  472         body = {"volume": vol}
  473         req = fakes.HTTPRequest.blank('/v2/volumes')
  474         self.assertRaises(exception.ValidationError,
  475                           self.controller.create,
  476                           req,
  477                           body=body)
  478 
  479     def test_volume_create_with_image_id_not_uuid_format(self):
  480         self.mock_object(volume_api.API, "create", v2_fakes.fake_volume_create)
  481         self.mock_object(fake_image._FakeImageService,
  482                          "detail",
  483                          v2_fakes.fake_image_service_detail)
  484         self.ext_mgr.extensions = {'os-image-create': 'fake'}
  485         vol = self._vol_in_request_body(availability_zone="cinder",
  486                                         image_id="12345")
  487         body = {"volume": vol}
  488         req = fakes.HTTPRequest.blank('/v2/volumes')
  489         self.assertRaises(webob.exc.HTTPBadRequest,
  490                           self.controller.create,
  491                           req,
  492                           body=body)
  493 
  494     def test_volume_create_with_image_id_with_empty_string(self):
  495         self.mock_object(volume_api.API, "create", v2_fakes.fake_volume_create)
  496         self.mock_object(fake_image._FakeImageService,
  497                          "detail",
  498                          v2_fakes.fake_image_service_detail)
  499         self.ext_mgr.extensions = {'os-image-create': 'fake'}
  500         vol = self._vol_in_request_body(availability_zone="cinder",
  501                                         image_id="")
  502         body = {"volume": vol}
  503         req = fakes.HTTPRequest.blank('/v2/volumes')
  504         self.assertRaises(webob.exc.HTTPBadRequest,
  505                           self.controller.create,
  506                           req,
  507                           body=body)
  508 
  509     @mock.patch(
  510         'cinder.api.openstack.wsgi.Controller.validate_name_and_description')
  511     def test_volume_create_with_image_name(self, mock_validate):
  512         self.mock_object(volume_api.API, "create",
  513                          v2_fakes.fake_volume_api_create)
  514         self.mock_object(db.sqlalchemy.api, '_volume_type_get_full',
  515                          v2_fakes.fake_volume_type_get)
  516         self.mock_object(fake_image._FakeImageService,
  517                          "detail",
  518                          v2_fakes.fake_image_service_detail)
  519 
  520         test_id = "Fedora-x86_64-20-20140618-sda"
  521         self.ext_mgr.extensions = {'os-image-create': 'fake'}
  522         vol = self._vol_in_request_body(availability_zone="nova",
  523                                         image_ref=test_id)
  524         ex = self._expected_vol_from_controller(availability_zone="nova")
  525         body = {"volume": vol}
  526         req = fakes.HTTPRequest.blank('/v2/volumes')
  527         res_dict = self.controller.create(req, body=body)
  528         self.assertEqual(ex, res_dict)
  529 
  530     def test_volume_create_with_image_name_has_multiple(self):
  531         self.mock_object(db, 'volume_get', v2_fakes.fake_volume_get_db)
  532         self.mock_object(volume_api.API, "create", v2_fakes.fake_volume_create)
  533         self.mock_object(fake_image._FakeImageService,
  534                          "detail",
  535                          v2_fakes.fake_image_service_detail)
  536 
  537         test_id = "multi"
  538         self.ext_mgr.extensions = {'os-image-create': 'fake'}
  539         vol = self._vol_in_request_body(availability_zone="nova",
  540                                         image_ref=test_id)
  541         body = {"volume": vol}
  542         req = fakes.HTTPRequest.blank('/v2/volumes')
  543         self.assertRaises(webob.exc.HTTPConflict,
  544                           self.controller.create,
  545                           req,
  546                           body=body)
  547 
  548     def test_volume_create_with_image_name_no_match(self):
  549         self.mock_object(db, 'volume_get', v2_fakes.fake_volume_get_db)
  550         self.mock_object(volume_api.API, "create", v2_fakes.fake_volume_create)
  551         self.mock_object(fake_image._FakeImageService,
  552                          "detail",
  553                          v2_fakes.fake_image_service_detail)
  554 
  555         test_id = "MissingName"
  556         self.ext_mgr.extensions = {'os-image-create': 'fake'}
  557         vol = self._vol_in_request_body(availability_zone="nova",
  558                                         image_ref=test_id)
  559         body = {"volume": vol}
  560         req = fakes.HTTPRequest.blank('/v2/volumes')
  561         self.assertRaises(webob.exc.HTTPBadRequest,
  562                           self.controller.create,
  563                           req,
  564                           body=body)
  565 
  566     def test_volume_create_with_invalid_multiattach(self):
  567         vol = self._vol_in_request_body(multiattach="InvalidBool")
  568         body = {"volume": vol}
  569         req = fakes.HTTPRequest.blank('/v2/volumes')
  570 
  571         self.assertRaises(exception.ValidationError,
  572                           self.controller.create,
  573                           req,
  574                           body=body)
  575 
  576     @mock.patch.object(volume_api.API, 'create', autospec=True)
  577     @mock.patch.object(volume_api.API, 'get', autospec=True)
  578     @mock.patch.object(db.sqlalchemy.api, '_volume_type_get_full',
  579                        autospec=True)
  580     def test_volume_create_with_valid_multiattach(self,
  581                                                   volume_type_get,
  582                                                   get, create):
  583         create.side_effect = v2_fakes.fake_volume_api_create
  584         get.side_effect = v2_fakes.fake_volume_get
  585         volume_type_get.side_effect = v2_fakes.fake_volume_type_get
  586 
  587         vol = self._vol_in_request_body(multiattach=True)
  588         body = {"volume": vol}
  589 
  590         ex = self._expected_vol_from_controller(multiattach=True)
  591 
  592         req = fakes.HTTPRequest.blank('/v2/volumes')
  593         res_dict = self.controller.create(req, body=body)
  594 
  595         self.assertEqual(ex, res_dict)
  596 
  597     @ddt.data({'a' * 256: 'a'},
  598               {'a': 'a' * 256},
  599               {'': 'a'},
  600               {'a': None})
  601     def test_volume_create_with_invalid_metadata(self, value):
  602         vol = self._vol_in_request_body()
  603         vol['metadata'] = value
  604         body = {"volume": vol}
  605         req = fakes.HTTPRequest.blank('/v2/volumes')
  606 
  607         self.assertRaises(exception.ValidationError,
  608                           self.controller.create,
  609                           req,
  610                           body=body)
  611 
  612     @ddt.data({"name": "Updated Test Name",
  613                "description": "Updated Test Description"},
  614               {"name": "      test name   ",
  615                "description": "    test description   "})
  616     def test_volume_update(self, body):
  617         self.mock_object(volume_api.API, 'get', v2_fakes.fake_volume_api_get)
  618         self.mock_object(volume_api.API, "update", v2_fakes.fake_volume_update)
  619         self.mock_object(db.sqlalchemy.api, '_volume_type_get_full',
  620                          v2_fakes.fake_volume_type_get)
  621         updates = {
  622             "name": body['name'],
  623             "description": body['description']
  624         }
  625         body = {"volume": updates}
  626         req = fakes.HTTPRequest.blank('/v2/volumes/%s' % fake.VOLUME_ID)
  627         self.assertEqual(0, len(self.notifier.notifications))
  628         name = updates["name"].strip()
  629         description = updates["description"].strip()
  630         expected = self._expected_vol_from_controller(
  631             availability_zone=v2_fakes.DEFAULT_AZ, name=name,
  632             description=description,
  633             metadata={'attached_mode': 'rw', 'readonly': 'False'})
  634         res_dict = self.controller.update(req, fake.VOLUME_ID, body=body)
  635         self.assertEqual(expected, res_dict)
  636         self.assertEqual(2, len(self.notifier.notifications))
  637 
  638     @mock.patch(
  639         'cinder.api.openstack.wsgi.Controller.validate_name_and_description')
  640     def test_volume_update_deprecation(self, mock_validate):
  641         self.mock_object(volume_api.API, 'get', v2_fakes.fake_volume_api_get)
  642         self.mock_object(volume_api.API, "update", v2_fakes.fake_volume_update)
  643         self.mock_object(db.sqlalchemy.api, '_volume_type_get_full',
  644                          v2_fakes.fake_volume_type_get)
  645 
  646         updates = {
  647             "display_name": "Updated Test Name",
  648             "display_description": "Updated Test Description",
  649         }
  650         body = {"volume": updates}
  651         req = fakes.HTTPRequest.blank('/v2/volumes/%s' % fake.VOLUME_ID)
  652         self.assertEqual(0, len(self.notifier.notifications))
  653         res_dict = self.controller.update(req, fake.VOLUME_ID, body=body)
  654         expected = self._expected_vol_from_controller(
  655             availability_zone=v2_fakes.DEFAULT_AZ, name="Updated Test Name",
  656             description="Updated Test Description",
  657             metadata={'attached_mode': 'rw', 'readonly': 'False'})
  658         self.assertEqual(expected, res_dict)
  659         self.assertEqual(2, len(self.notifier.notifications))
  660         self.assertTrue(mock_validate.called)
  661 
  662     @mock.patch(
  663         'cinder.api.openstack.wsgi.Controller.validate_name_and_description')
  664     def test_volume_update_deprecation_key_priority(self, mock_validate):
  665         """Test current update keys have priority over deprecated keys."""
  666         self.mock_object(volume_api.API, 'get', v2_fakes.fake_volume_api_get)
  667         self.mock_object(volume_api.API, "update", v2_fakes.fake_volume_update)
  668         self.mock_object(db.sqlalchemy.api, '_volume_type_get_full',
  669                          v2_fakes.fake_volume_type_get)
  670 
  671         updates = {
  672             "name": "New Name",
  673             "description": "New Description",
  674             "display_name": "Not Shown Name",
  675             "display_description": "Not Shown Description",
  676         }
  677         body = {"volume": updates}
  678         req = fakes.HTTPRequest.blank('/v2/volumes/%s' % fake.VOLUME_ID)
  679         self.assertEqual(0, len(self.notifier.notifications))
  680         res_dict = self.controller.update(req, fake.VOLUME_ID, body=body)
  681         expected = self._expected_vol_from_controller(
  682             availability_zone=v2_fakes.DEFAULT_AZ,
  683             name="New Name", description="New Description",
  684             metadata={'attached_mode': 'rw', 'readonly': 'False'})
  685         self.assertEqual(expected, res_dict)
  686         self.assertEqual(2, len(self.notifier.notifications))
  687         self.assertTrue(mock_validate.called)
  688 
  689     @mock.patch(
  690         'cinder.api.openstack.wsgi.Controller.validate_name_and_description')
  691     def test_volume_update_metadata(self, mock_validate):
  692         self.mock_object(volume_api.API, 'get', v2_fakes.fake_volume_api_get)
  693         self.mock_object(volume_api.API, "update", v2_fakes.fake_volume_update)
  694         self.mock_object(db.sqlalchemy.api, '_volume_type_get_full',
  695                          v2_fakes.fake_volume_type_get)
  696 
  697         updates = {
  698             "metadata": {"qos_max_iops": '2000'}
  699         }
  700         body = {"volume": updates}
  701         req = fakes.HTTPRequest.blank('/v2/volumes/%s' % fake.VOLUME_ID)
  702         self.assertEqual(0, len(self.notifier.notifications))
  703         res_dict = self.controller.update(req, fake.VOLUME_ID, body=body)
  704         expected = self._expected_vol_from_controller(
  705             availability_zone=v2_fakes.DEFAULT_AZ,
  706             metadata={'attached_mode': 'rw', 'readonly': 'False',
  707                       'qos_max_iops': '2000'})
  708         self.assertEqual(expected, res_dict)
  709         self.assertEqual(2, len(self.notifier.notifications))
  710         self.assertTrue(mock_validate.called)
  711 
  712     @mock.patch(
  713         'cinder.api.openstack.wsgi.Controller.validate_name_and_description')
  714     def test_volume_update_with_admin_metadata(self, mock_validate):
  715         self.mock_object(volume_api.API, "update", v2_fakes.fake_volume_update)
  716 
  717         volume = v2_fakes.create_fake_volume(fake.VOLUME_ID)
  718         del volume['name']
  719         del volume['volume_type']
  720         del volume['volume_type_id']
  721         volume['metadata'] = {'key': 'value'}
  722         db.volume_create(context.get_admin_context(), volume)
  723 
  724         db.volume_admin_metadata_update(context.get_admin_context(),
  725                                         fake.VOLUME_ID,
  726                                         {"readonly": "True",
  727                                          "invisible_key": "invisible_value"},
  728                                         False)
  729         values = {'volume_id': fake.VOLUME_ID, }
  730         attachment = db.volume_attach(context.get_admin_context(), values)
  731         db.volume_attached(context.get_admin_context(),
  732                            attachment['id'], fake.INSTANCE_ID, None, '/')
  733         attach_tmp = db.volume_attachment_get(context.get_admin_context(),
  734                                               attachment['id'])
  735         volume_tmp = db.volume_get(context.get_admin_context(), fake.VOLUME_ID)
  736         updates = {
  737             "name": "Updated Test Name",
  738         }
  739         body = {"volume": updates}
  740         req = fakes.HTTPRequest.blank('/v2/volumes/%s' % fake.VOLUME_ID)
  741         self.assertEqual(0, len(self.notifier.notifications))
  742         admin_ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
  743         req.environ['cinder.context'] = admin_ctx
  744         res_dict = self.controller.update(req, fake.VOLUME_ID, body=body)
  745         expected = self._expected_vol_from_controller(
  746             availability_zone=v2_fakes.DEFAULT_AZ, volume_type=None,
  747             status='in-use', name='Updated Test Name',
  748             attachments=[{'id': fake.VOLUME_ID,
  749                           'attachment_id': attachment['id'],
  750                           'volume_id': v2_fakes.DEFAULT_VOL_ID,
  751                           'server_id': fake.INSTANCE_ID,
  752                           'host_name': None,
  753                           'device': '/',
  754                           'attached_at': attach_tmp['attach_time'].replace(
  755                               tzinfo=iso8601.UTC),
  756                           }],
  757             metadata={'key': 'value', 'readonly': 'True'},
  758             with_migration_status=True)
  759         expected['volume']['updated_at'] = volume_tmp['updated_at'].replace(
  760             tzinfo=iso8601.UTC)
  761         self.assertEqual(expected, res_dict)
  762         self.assertEqual(2, len(self.notifier.notifications))
  763         self.assertTrue(mock_validate.called)
  764 
  765     @ddt.data({'a' * 256: 'a'},
  766               {'a': 'a' * 256},
  767               {'': 'a'},
  768               {'a': None})
  769     @mock.patch.object(volume_api.API, 'get',
  770                        side_effect=v2_fakes.fake_volume_api_get, autospec=True)
  771     def test_volume_update_with_invalid_metadata(self, value, get):
  772         updates = {
  773             "metadata": value
  774         }
  775         body = {"volume": updates}
  776         req = fakes.HTTPRequest.blank('/v2/volumes/%s' % fake.VOLUME_ID)
  777 
  778         self.assertRaises(exception.ValidationError,
  779                           self.controller.update,
  780                           req, fake.VOLUME_ID, body=body)
  781 
  782     def test_update_empty_body(self):
  783         body = {}
  784         req = fakes.HTTPRequest.blank('/v2/volumes/%s' % fake.VOLUME_ID)
  785         self.assertRaises(exception.ValidationError,
  786                           self.controller.update,
  787                           req, fake.VOLUME_ID, body=body)
  788 
  789     def test_update_invalid_body(self):
  790         body = {
  791             'name': 'missing top level volume key'
  792         }
  793         req = fakes.HTTPRequest.blank('/v2/volumes/%s' % fake.VOLUME_ID)
  794         self.assertRaises(exception.ValidationError,
  795                           self.controller.update,
  796                           req, fake.VOLUME_ID, body=body)
  797 
  798     @ddt.data({'name': 'a' * 256},
  799               {'description': 'a' * 256},
  800               {'display_name': 'a' * 256},
  801               {'display_description': 'a' * 256})
  802     def test_update_exceeds_length_name_description(self, vol):
  803         req = fakes.HTTPRequest.blank('/v2/volumes/%s' % fake.VOLUME_ID)
  804         body = {'volume': vol}
  805         self.assertRaises(exception.InvalidInput,
  806                           self.controller.update,
  807                           req, fake.VOLUME_ID, body=body)
  808 
  809     def test_update_not_found(self):
  810         self.mock_object(volume_api.API, "get",
  811                          v2_fakes.fake_volume_get_notfound)
  812         updates = {
  813             "name": "Updated Test Name",
  814         }
  815         body = {"volume": updates}
  816         req = fakes.HTTPRequest.blank('/v2/volumes/%s' % fake.VOLUME_ID)
  817         self.assertRaises(exception.VolumeNotFound,
  818                           self.controller.update,
  819                           req, fake.VOLUME_ID, body=body)
  820 
  821     def test_volume_list_summary(self):
  822         self.mock_object(volume_api.API, 'get_all',
  823                          v2_fakes.fake_volume_api_get_all_by_project)
  824         self.mock_object(db.sqlalchemy.api, '_volume_type_get_full',
  825                          v2_fakes.fake_volume_type_get)
  826 
  827         req = fakes.HTTPRequest.blank('/v2/volumes')
  828         res_dict = self.controller.index(req)
  829         expected = {
  830             'volumes': [
  831                 {
  832                     'name': v2_fakes.DEFAULT_VOL_NAME,
  833                     'id': fake.VOLUME_ID,
  834                     'links': [
  835                         {
  836                             'href': 'http://localhost/v2/%s/volumes/%s' % (
  837                                     fake.PROJECT_ID, fake.VOLUME_ID),
  838                             'rel': 'self'
  839                         },
  840                         {
  841                             'href': 'http://localhost/%s/volumes/%s' % (
  842                                     fake.PROJECT_ID, fake.VOLUME_ID),
  843                             'rel': 'bookmark'
  844                         }
  845                     ],
  846                 }
  847             ]
  848         }
  849         self.assertEqual(expected, res_dict)
  850         # Finally test that we cached the returned volumes
  851         self.assertEqual(1, len(req.cached_resource()))
  852 
  853     def test_volume_list_detail(self):
  854         self.mock_object(volume_api.API, 'get_all',
  855                          v2_fakes.fake_volume_api_get_all_by_project)
  856         self.mock_object(db.sqlalchemy.api, '_volume_type_get_full',
  857                          v2_fakes.fake_volume_type_get)
  858 
  859         req = fakes.HTTPRequest.blank('/v2/volumes/detail')
  860         res_dict = self.controller.detail(req)
  861         exp_vol = self._expected_vol_from_controller(
  862             availability_zone=v2_fakes.DEFAULT_AZ,
  863             metadata={'attached_mode': 'rw', 'readonly': 'False'})
  864         expected = {'volumes': [exp_vol['volume']]}
  865         self.assertEqual(expected, res_dict)
  866         # Finally test that we cached the returned volumes
  867         self.assertEqual(1, len(req.cached_resource()))
  868 
  869     def test_volume_list_detail_with_admin_metadata(self):
  870         volume = v2_fakes.create_fake_volume(fake.VOLUME_ID)
  871         del volume['name']
  872         del volume['volume_type']
  873         del volume['volume_type_id']
  874         volume['metadata'] = {'key': 'value'}
  875         db.volume_create(context.get_admin_context(), volume)
  876         db.volume_admin_metadata_update(context.get_admin_context(),
  877                                         fake.VOLUME_ID,
  878                                         {"readonly": "True",
  879                                          "invisible_key": "invisible_value"},
  880                                         False)
  881         values = {'volume_id': fake.VOLUME_ID, }
  882         attachment = db.volume_attach(context.get_admin_context(), values)
  883         db.volume_attached(context.get_admin_context(),
  884                            attachment['id'], fake.INSTANCE_ID, None, '/')
  885         attach_tmp = db.volume_attachment_get(context.get_admin_context(),
  886                                               attachment['id'])
  887         volume_tmp = db.volume_get(context.get_admin_context(), fake.VOLUME_ID)
  888 
  889         req = fakes.HTTPRequest.blank('/v2/volumes/detail')
  890         admin_ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
  891         req.environ['cinder.context'] = admin_ctx
  892         res_dict = self.controller.detail(req)
  893         exp_vol = self._expected_vol_from_controller(
  894             availability_zone=v2_fakes.DEFAULT_AZ,
  895             status="in-use", volume_type=None,
  896             attachments=[{'attachment_id': attachment['id'],
  897                           'device': '/',
  898                           'server_id': fake.INSTANCE_ID,
  899                           'host_name': None,
  900                           'id': fake.VOLUME_ID,
  901                           'volume_id': v2_fakes.DEFAULT_VOL_ID,
  902                           'attached_at': attach_tmp['attach_time'].replace(
  903                               tzinfo=iso8601.UTC),
  904                           }],
  905             metadata={'key': 'value', 'readonly': 'True'},
  906             with_migration_status=True)
  907         exp_vol['volume']['updated_at'] = volume_tmp['updated_at'].replace(
  908             tzinfo=iso8601.UTC)
  909         expected = {'volumes': [exp_vol['volume']]}
  910         self.assertEqual(expected, res_dict)
  911 
  912     def test_volume_index_with_marker(self):
  913         def fake_volume_get_all_by_project(context, project_id, marker, limit,
  914                                            sort_keys=None, sort_dirs=None,
  915                                            filters=None,
  916                                            viewable_admin_meta=False,
  917                                            offset=0):
  918             return [
  919                 v2_fakes.create_fake_volume(fake.VOLUME_ID,
  920                                             display_name='vol1'),
  921                 v2_fakes.create_fake_volume(fake.VOLUME2_ID,
  922                                             display_name='vol2'),
  923             ]
  924         self.mock_object(db, 'volume_get_all_by_project',
  925                          fake_volume_get_all_by_project)
  926         self.mock_object(volume_api.API, 'get', v2_fakes.fake_volume_get)
  927 
  928         req = fakes.HTTPRequest.blank('/v2/volumes?marker=1')
  929         res_dict = self.controller.index(req)
  930         volumes = res_dict['volumes']
  931         self.assertEqual(2, len(volumes))
  932         self.assertEqual(fake.VOLUME_ID, volumes[0]['id'])
  933         self.assertEqual(fake.VOLUME2_ID, volumes[1]['id'])
  934 
  935     def test_volume_index_limit(self):
  936         self.mock_object(db, 'volume_get_all_by_project',
  937                          v2_fakes.fake_volume_get_all_by_project)
  938         self.mock_object(volume_api.API, 'get', v2_fakes.fake_volume_get)
  939 
  940         req = fakes.HTTPRequest.blank('/v2/volumes'
  941                                       '?limit=1&name=foo'
  942                                       '&sort=id1:asc')
  943         res_dict = self.controller.index(req)
  944         volumes = res_dict['volumes']
  945         self.assertEqual(1, len(volumes))
  946 
  947         # Ensure that the next link is correctly formatted, it should
  948         # contain the same limit, filter, and sort information as the
  949         # original request as well as a marker; this ensures that the
  950         # caller can simply use the "next" link and that they do not
  951         # need to manually insert the limit and sort information.
  952         links = res_dict['volumes_links']
  953         self.assertEqual('next', links[0]['rel'])
  954         href_parts = urllib.parse.urlparse(links[0]['href'])
  955         self.assertEqual('/v2/%s/volumes' % fake.PROJECT_ID, href_parts.path)
  956         params = urllib.parse.parse_qs(href_parts.query)
  957         self.assertEqual(str(volumes[0]['id']), params['marker'][0])
  958         self.assertEqual('1', params['limit'][0])
  959         self.assertEqual('foo', params['name'][0])
  960         self.assertEqual('id1:asc', params['sort'][0])
  961 
  962     def test_volume_index_limit_negative(self):
  963         req = fakes.HTTPRequest.blank('/v2/volumes?limit=-1')
  964         self.assertRaises(webob.exc.HTTPBadRequest,
  965                           self.controller.index,
  966                           req)
  967 
  968     def test_volume_index_limit_non_int(self):
  969         req = fakes.HTTPRequest.blank('/v2/volumes?limit=a')
  970         self.assertRaises(webob.exc.HTTPBadRequest,
  971                           self.controller.index,
  972                           req)
  973 
  974     def test_volume_index_limit_marker(self):
  975         self.mock_object(db, 'volume_get_all_by_project',
  976                          v2_fakes.fake_volume_get_all_by_project)
  977         self.mock_object(volume_api.API, 'get', v2_fakes.fake_volume_get)
  978 
  979         req = fakes.HTTPRequest.blank('/v2/volumes?marker=1&limit=1')
  980         res_dict = self.controller.index(req)
  981         volumes = res_dict['volumes']
  982         self.assertEqual(1, len(volumes))
  983         self.assertEqual(fake.VOLUME_ID, volumes[0]['id'])
  984 
  985     def _create_db_volumes(self, num_volumes):
  986         volumes = [utils.create_volume(self.ctxt, display_name='vol%s' % i)
  987                    for i in range(num_volumes)]
  988         for vol in volumes:
  989             self.addCleanup(db.volume_destroy, self.ctxt, vol.id)
  990         volumes.reverse()
  991         return volumes
  992 
  993     def test_volume_index_limit_offset(self):
  994         created_volumes = self._create_db_volumes(2)
  995         req = fakes.HTTPRequest.blank('/v2/volumes?limit=2&offset=1')
  996         res_dict = self.controller.index(req)
  997         volumes = res_dict['volumes']
  998         self.assertEqual(1, len(volumes))
  999         self.assertEqual(created_volumes[1].id, volumes[0]['id'])
 1000 
 1001         req = fakes.HTTPRequest.blank('/v2/volumes?limit=-1&offset=1')
 1002         self.assertRaises(webob.exc.HTTPBadRequest,
 1003                           self.controller.index,
 1004                           req)
 1005 
 1006         req = fakes.HTTPRequest.blank('/v2/volumes?limit=a&offset=1')
 1007         self.assertRaises(webob.exc.HTTPBadRequest,
 1008                           self.controller.index,
 1009                           req)
 1010 
 1011         # Test that we get an exception HTTPBadRequest(400) with an offset
 1012         # greater than the maximum offset value.
 1013         url = '/v2/volumes?limit=2&offset=43543564546567575'
 1014         req = fakes.HTTPRequest.blank(url)
 1015         self.assertRaises(webob.exc.HTTPBadRequest,
 1016                           self.controller.index,
 1017                           req)
 1018 
 1019     def test_volume_detail_with_marker(self):
 1020         def fake_volume_get_all_by_project(context, project_id, marker, limit,
 1021                                            sort_keys=None, sort_dirs=None,
 1022                                            filters=None,
 1023                                            viewable_admin_meta=False,
 1024                                            offset=0):
 1025             return [
 1026                 v2_fakes.create_fake_volume(fake.VOLUME_ID,
 1027                                             display_name='vol1'),
 1028                 v2_fakes.create_fake_volume(fake.VOLUME2_ID,
 1029                                             display_name='vol2'),
 1030             ]
 1031         self.mock_object(db, 'volume_get_all_by_project',
 1032                          fake_volume_get_all_by_project)
 1033         self.mock_object(db.sqlalchemy.api, '_volume_type_get_full',
 1034                          v2_fakes.fake_volume_type_get)
 1035 
 1036         req = fakes.HTTPRequest.blank('/v2/volumes/detail?marker=1')
 1037         res_dict = self.controller.detail(req)
 1038         volumes = res_dict['volumes']
 1039         self.assertEqual(2, len(volumes))
 1040         self.assertEqual(fake.VOLUME_ID, volumes[0]['id'])
 1041         self.assertEqual(fake.VOLUME2_ID, volumes[1]['id'])
 1042 
 1043     def test_volume_detail_limit(self):
 1044         self.mock_object(db, 'volume_get_all_by_project',
 1045                          v2_fakes.fake_volume_get_all_by_project)
 1046         self.mock_object(db.sqlalchemy.api, '_volume_type_get_full',
 1047                          v2_fakes.fake_volume_type_get)
 1048         req = fakes.HTTPRequest.blank('/v2/volumes/detail?limit=1')
 1049         res_dict = self.controller.detail(req)
 1050         volumes = res_dict['volumes']
 1051         self.assertEqual(1, len(volumes))
 1052 
 1053         # Ensure that the next link is correctly formatted
 1054         links = res_dict['volumes_links']
 1055         self.assertEqual('next', links[0]['rel'])
 1056         href_parts = urllib.parse.urlparse(links[0]['href'])
 1057         self.assertEqual('/v2/%s/volumes/detail' % fake.PROJECT_ID,
 1058                          href_parts.path)
 1059         params = urllib.parse.parse_qs(href_parts.query)
 1060         self.assertIn('marker', params)
 1061         self.assertEqual('1', params['limit'][0])
 1062 
 1063     def test_volume_detail_limit_negative(self):
 1064         req = fakes.HTTPRequest.blank('/v2/volumes/detail?limit=-1')
 1065         self.assertRaises(webob.exc.HTTPBadRequest,
 1066                           self.controller.detail,
 1067                           req)
 1068 
 1069     def test_volume_detail_limit_non_int(self):
 1070         req = fakes.HTTPRequest.blank('/v2/volumes/detail?limit=a')
 1071         self.assertRaises(webob.exc.HTTPBadRequest,
 1072                           self.controller.detail,
 1073                           req)
 1074 
 1075     def test_volume_detail_limit_marker(self):
 1076         self.mock_object(db, 'volume_get_all_by_project',
 1077                          v2_fakes.fake_volume_get_all_by_project)
 1078         self.mock_object(db.sqlalchemy.api, '_volume_type_get_full',
 1079                          v2_fakes.fake_volume_type_get)
 1080 
 1081         req = fakes.HTTPRequest.blank('/v2/volumes/detail?marker=1&limit=1')
 1082         res_dict = self.controller.detail(req)
 1083         volumes = res_dict['volumes']
 1084         self.assertEqual(1, len(volumes))
 1085         self.assertEqual(fake.VOLUME_ID, volumes[0]['id'])
 1086 
 1087     def test_volume_detail_limit_offset(self):
 1088         created_volumes = self._create_db_volumes(2)
 1089         req = fakes.HTTPRequest.blank('/v2/volumes/detail?limit=2&offset=1')
 1090         res_dict = self.controller.detail(req)
 1091         volumes = res_dict['volumes']
 1092         self.assertEqual(1, len(volumes))
 1093         self.assertEqual(created_volumes[1].id, volumes[0]['id'])
 1094 
 1095         req = fakes.HTTPRequest.blank('/v2/volumes/detail?limit=2&offset=1',
 1096                                       use_admin_context=True)
 1097         res_dict = self.controller.detail(req)
 1098         volumes = res_dict['volumes']
 1099         self.assertEqual(1, len(volumes))
 1100         self.assertEqual(created_volumes[1].id, volumes[0]['id'])
 1101 
 1102         req = fakes.HTTPRequest.blank('/v2/volumes/detail?limit=-1&offset=1')
 1103         self.assertRaises(webob.exc.HTTPBadRequest,
 1104                           self.controller.detail,
 1105                           req)
 1106 
 1107         req = fakes.HTTPRequest.blank('/v2/volumes/detail?limit=a&offset=1')
 1108         self.assertRaises(webob.exc.HTTPBadRequest,
 1109                           self.controller.detail,
 1110                           req)
 1111 
 1112         url = '/v2/volumes/detail?limit=2&offset=4536546546546467'
 1113         req = fakes.HTTPRequest.blank(url)
 1114         self.assertRaises(webob.exc.HTTPBadRequest,
 1115                           self.controller.detail,
 1116                           req)
 1117 
 1118     def test_volume_with_limit_zero(self):
 1119         def fake_volume_get_all(context, marker, limit, **kwargs):
 1120             return []
 1121         self.mock_object(db, 'volume_get_all', fake_volume_get_all)
 1122         req = fakes.HTTPRequest.blank('/v2/volumes?limit=0')
 1123         res_dict = self.controller.index(req)
 1124         expected = {'volumes': []}
 1125         self.assertEqual(expected, res_dict)
 1126 
 1127     def _validate_next_link(self, detailed, item_count, osapi_max_limit, limit,
 1128                             should_link_exist):
 1129         keys_fns = (('volumes', self.controller.index),
 1130                     ('volumes/detail', self.controller.detail))
 1131         key, fn = keys_fns[detailed]
 1132 
 1133         req_string = '/v2/%s?all_tenants=1' % key
 1134         if limit:
 1135             req_string += '&limit=%s' % limit
 1136         req = fakes.HTTPRequest.blank(req_string, use_admin_context=True)
 1137 
 1138         link_return = [{"rel": "next", "href": "fake_link"}]
 1139         self.flags(osapi_max_limit=osapi_max_limit)
 1140 
 1141         def get_pagination_params(params, max_limit=CONF.osapi_max_limit,
 1142                                   original_call=common.get_pagination_params):
 1143             return original_call(params, max_limit)
 1144 
 1145         def _get_limit_param(params, max_limit=CONF.osapi_max_limit,
 1146                              original_call=common._get_limit_param):
 1147             return original_call(params, max_limit)
 1148 
 1149         with mock.patch.object(common, 'get_pagination_params',
 1150                                get_pagination_params), \
 1151                 mock.patch.object(common, '_get_limit_param',
 1152                                   _get_limit_param), \
 1153                 mock.patch.object(common.ViewBuilder, '_generate_next_link',
 1154                                   return_value=link_return):
 1155             res_dict = fn(req)
 1156             self.assertEqual(item_count, len(res_dict['volumes']))
 1157             self.assertEqual(should_link_exist, 'volumes_links' in res_dict)
 1158 
 1159     def test_volume_default_limit(self):
 1160         self._create_db_volumes(3)
 1161 
 1162         # Verify both the index and detail queries
 1163         for detailed in (True, False):
 1164             # Number of volumes less than max, do not include
 1165             self._validate_next_link(detailed, item_count=3, osapi_max_limit=4,
 1166                                      limit=None, should_link_exist=False)
 1167 
 1168             # Number of volumes equals the max, next link will be included
 1169             self._validate_next_link(detailed, item_count=3, osapi_max_limit=3,
 1170                                      limit=None, should_link_exist=True)
 1171 
 1172             # Number of volumes more than the max, include next link
 1173             self._validate_next_link(detailed, item_count=2, osapi_max_limit=2,
 1174                                      limit=None, should_link_exist=True)
 1175 
 1176             # Limit lower than max but doesn't limit, no next link
 1177             self._validate_next_link(detailed, item_count=3, osapi_max_limit=5,
 1178                                      limit=4, should_link_exist=False)
 1179 
 1180             # Limit lower than max and limits, we have next link
 1181             self._validate_next_link(detailed, item_count=2, osapi_max_limit=4,
 1182                                      limit=2, should_link_exist=True)
 1183 
 1184             # Limit higher than max and max limits, we have next link
 1185             self._validate_next_link(detailed, item_count=2, osapi_max_limit=2,
 1186                                      limit=4, should_link_exist=True)
 1187 
 1188             # Limit higher than max but none of them limiting, no next link
 1189             self._validate_next_link(detailed, item_count=3, osapi_max_limit=4,
 1190                                      limit=5, should_link_exist=False)
 1191 
 1192     def test_volume_list_default_filters(self):
 1193         """Tests that the default filters from volume.api.API.get_all are set.
 1194 
 1195         1. 'no_migration_status'=True for non-admins and get_all_by_project is
 1196         invoked.
 1197         2. 'no_migration_status' is not included for admins.
 1198         3. When 'all_tenants' is not specified, then it is removed and
 1199         get_all_by_project is invoked for admins.
 1200         3. When 'all_tenants' is specified, then it is removed and get_all
 1201         is invoked for admins.
 1202         """
 1203         # Non-admin, project function should be called with no_migration_status
 1204         def fake_volume_get_all_by_project(context, project_id, marker, limit,
 1205                                            sort_keys=None, sort_dirs=None,
 1206                                            filters=None,
 1207                                            viewable_admin_meta=False,
 1208                                            offset=0):
 1209             self.assertTrue(filters['no_migration_targets'])
 1210             self.assertNotIn('all_tenants', filters)
 1211             return [v2_fakes.create_fake_volume(fake.VOLUME_ID,
 1212                                                 display_name='vol1')]
 1213 
 1214         def fake_volume_get_all(context, marker, limit,
 1215                                 sort_keys=None, sort_dirs=None,
 1216                                 filters=None,
 1217                                 viewable_admin_meta=False, offset=0):
 1218             return []
 1219         self.mock_object(db, 'volume_get_all_by_project',
 1220                          fake_volume_get_all_by_project)
 1221         self.mock_object(db, 'volume_get_all', fake_volume_get_all)
 1222 
 1223         # all_tenants does not matter for non-admin
 1224         for params in ['', '?all_tenants=1']:
 1225             req = fakes.HTTPRequest.blank('/v2/volumes%s' % params)
 1226             resp = self.controller.index(req)
 1227             self.assertEqual(1, len(resp['volumes']))
 1228             self.assertEqual('vol1', resp['volumes'][0]['name'])
 1229 
 1230         # Admin, all_tenants is not set, project function should be called
 1231         # without no_migration_status
 1232         def fake_volume_get_all_by_project2(context, project_id, marker, limit,
 1233                                             sort_keys=None, sort_dirs=None,
 1234                                             filters=None,
 1235                                             viewable_admin_meta=False,
 1236                                             offset=0):
 1237             self.assertNotIn('no_migration_targets', filters)
 1238             return [v2_fakes.create_fake_volume(fake.VOLUME_ID,
 1239                                                 display_name='vol2')]
 1240 
 1241         def fake_volume_get_all2(context, marker, limit,
 1242                                  sort_keys=None, sort_dirs=None,
 1243                                  filters=None,
 1244                                  viewable_admin_meta=False, offset=0):
 1245             return []
 1246         self.mock_object(db, 'volume_get_all_by_project',
 1247                          fake_volume_get_all_by_project2)
 1248         self.mock_object(db, 'volume_get_all', fake_volume_get_all2)
 1249 
 1250         req = fakes.HTTPRequest.blank('/v2/volumes', use_admin_context=True)
 1251         resp = self.controller.index(req)
 1252         self.assertEqual(1, len(resp['volumes']))
 1253         self.assertEqual('vol2', resp['volumes'][0]['name'])
 1254 
 1255         # Admin, all_tenants is set, get_all function should be called
 1256         # without no_migration_status
 1257         def fake_volume_get_all_by_project3(context, project_id, marker, limit,
 1258                                             sort_keys=None, sort_dirs=None,
 1259                                             filters=None,
 1260                                             viewable_admin_meta=False,
 1261                                             offset=0):
 1262             return []
 1263 
 1264         def fake_volume_get_all3(context, marker, limit,
 1265                                  sort_keys=None, sort_dirs=None,
 1266                                  filters=None,
 1267                                  viewable_admin_meta=False, offset=0):
 1268             self.assertNotIn('no_migration_targets', filters)
 1269             self.assertNotIn('all_tenants', filters)
 1270             return [v2_fakes.create_fake_volume(fake.VOLUME3_ID,
 1271                                                 display_name='vol3')]
 1272         self.mock_object(db, 'volume_get_all_by_project',
 1273                          fake_volume_get_all_by_project3)
 1274         self.mock_object(db, 'volume_get_all', fake_volume_get_all3)
 1275 
 1276         req = fakes.HTTPRequest.blank('/v2/volumes?all_tenants=1',
 1277                                       use_admin_context=True)
 1278         resp = self.controller.index(req)
 1279         self.assertEqual(1, len(resp['volumes']))
 1280         self.assertEqual('vol3', resp['volumes'][0]['name'])
 1281 
 1282     def test_volume_show(self):
 1283         self.mock_object(volume_api.API, 'get', v2_fakes.fake_volume_api_get)
 1284         self.mock_object(db.sqlalchemy.api, '_volume_type_get_full',
 1285                          v2_fakes.fake_volume_type_get)
 1286 
 1287         req = fakes.HTTPRequest.blank('/v2/volumes/%s' % fake.VOLUME_ID)
 1288         res_dict = self.controller.show(req, fake.VOLUME_ID)
 1289         expected = self._expected_vol_from_controller(
 1290             availability_zone=v2_fakes.DEFAULT_AZ,
 1291             metadata={'attached_mode': 'rw', 'readonly': 'False'})
 1292         self.assertEqual(expected, res_dict)
 1293         # Finally test that we cached the returned volume
 1294         self.assertIsNotNone(req.cached_resource_by_id(fake.VOLUME_ID))
 1295 
 1296     def test_volume_show_no_attachments(self):
 1297         def fake_volume_get(self, context, volume_id, **kwargs):
 1298             vol = v2_fakes.create_fake_volume(
 1299                 volume_id, attach_status=
 1300                 fields.VolumeAttachStatus.DETACHED)
 1301             return fake_volume.fake_volume_obj(context, **vol)
 1302 
 1303         def fake_volume_admin_metadata_get(context, volume_id, **kwargs):
 1304             return v2_fakes.fake_volume_admin_metadata_get(
 1305                 context, volume_id, attach_status=
 1306                 fields.VolumeAttachStatus.DETACHED)
 1307 
 1308         self.mock_object(volume_api.API, 'get', fake_volume_get)
 1309         self.mock_object(db, 'volume_admin_metadata_get',
 1310                          fake_volume_admin_metadata_get)
 1311         self.mock_object(db.sqlalchemy.api, '_volume_type_get_full',
 1312                          v2_fakes.fake_volume_type_get)
 1313 
 1314         req = fakes.HTTPRequest.blank('/v2/volumes/%s' % fake.VOLUME_ID)
 1315         res_dict = self.controller.show(req, fake.VOLUME_ID)
 1316         expected = self._expected_vol_from_controller(
 1317             availability_zone=v2_fakes.DEFAULT_AZ,
 1318             metadata={'readonly': 'False'})
 1319 
 1320         self.assertEqual(expected, res_dict)
 1321 
 1322     def test_volume_show_no_volume(self):
 1323         self.mock_object(volume_api.API, "get",
 1324                          v2_fakes.fake_volume_get_notfound)
 1325 
 1326         req = fakes.HTTPRequest.blank('/v2/volumes/%s' % fake.VOLUME_ID)
 1327         self.assertRaises(exception.VolumeNotFound, self.controller.show,
 1328                           req, 1)
 1329         # Finally test that nothing was cached
 1330         self.assertIsNone(req.cached_resource_by_id(fake.VOLUME_ID))
 1331 
 1332     def test_volume_show_with_admin_metadata(self):
 1333         volume = v2_fakes.create_fake_volume(fake.VOLUME_ID)
 1334         del volume['name']
 1335         del volume['volume_type']
 1336         del volume['volume_type_id']
 1337         volume['metadata'] = {'key': 'value'}
 1338         db.volume_create(context.get_admin_context(), volume)
 1339         db.volume_admin_metadata_update(context.get_admin_context(),
 1340                                         fake.VOLUME_ID,
 1341                                         {"readonly": "True",
 1342                                          "invisible_key": "invisible_value"},
 1343                                         False)
 1344         values = {'volume_id': fake.VOLUME_ID, }
 1345         attachment = db.volume_attach(context.get_admin_context(), values)
 1346         db.volume_attached(context.get_admin_context(),
 1347                            attachment['id'], fake.INSTANCE_ID, None, '/')
 1348         attach_tmp = db.volume_attachment_get(context.get_admin_context(),
 1349                                               attachment['id'])
 1350         volume_tmp = db.volume_get(context.get_admin_context(), fake.VOLUME_ID)
 1351         req = fakes.HTTPRequest.blank('/v2/volumes/%s' % fake.VOLUME_ID)
 1352         admin_ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
 1353         req.environ['cinder.context'] = admin_ctx
 1354         res_dict = self.controller.show(req, fake.VOLUME_ID)
 1355         expected = self._expected_vol_from_controller(
 1356             availability_zone=v2_fakes.DEFAULT_AZ,
 1357             volume_type=None, status='in-use',
 1358             attachments=[{'id': fake.VOLUME_ID,
 1359                           'attachment_id': attachment['id'],
 1360                           'volume_id': v2_fakes.DEFAULT_VOL_ID,
 1361                           'server_id': fake.INSTANCE_ID,
 1362                           'host_name': None,
 1363                           'device': '/',
 1364                           'attached_at': attach_tmp['attach_time'].replace(
 1365                               tzinfo=iso8601.UTC),
 1366                           }],
 1367             metadata={'key': 'value', 'readonly': 'True'},
 1368             with_migration_status=True)
 1369         expected['volume']['updated_at'] = volume_tmp['updated_at'].replace(
 1370             tzinfo=iso8601.UTC)
 1371         self.assertEqual(expected, res_dict)
 1372 
 1373     def test_volume_show_with_encrypted_volume(self):
 1374         def fake_volume_get(self, context, volume_id, **kwargs):
 1375             vol = v2_fakes.create_fake_volume(volume_id,
 1376                                               encryption_key_id=fake.KEY_ID)
 1377             return fake_volume.fake_volume_obj(context, **vol)
 1378 
 1379         self.mock_object(volume_api.API, 'get', fake_volume_get)
 1380         self.mock_object(db.sqlalchemy.api, '_volume_type_get_full',
 1381                          v2_fakes.fake_volume_type_get)
 1382 
 1383         req = fakes.HTTPRequest.blank('/v2/volumes/%s' % fake.VOLUME_ID)
 1384         res_dict = self.controller.show(req, fake.VOLUME_ID)
 1385         self.assertTrue(res_dict['volume']['encrypted'])
 1386 
 1387     def test_volume_show_with_unencrypted_volume(self):
 1388         self.mock_object(volume_api.API, 'get', v2_fakes.fake_volume_api_get)
 1389         self.mock_object(db.sqlalchemy.api, '_volume_type_get_full',
 1390                          v2_fakes.fake_volume_type_get)
 1391 
 1392         req = fakes.HTTPRequest.blank('/v2/volumes/%s' % fake.VOLUME_ID)
 1393         res_dict = self.controller.show(req, fake.VOLUME_ID)
 1394         self.assertEqual(False, res_dict['volume']['encrypted'])
 1395 
 1396     def test_volume_show_with_error_managing_deleting(self):
 1397         def fake_volume_get(self, context, volume_id, **kwargs):
 1398             vol = v2_fakes.create_fake_volume(volume_id,
 1399                                               status='error_managing_deleting')
 1400             return fake_volume.fake_volume_obj(context, **vol)
 1401 
 1402         self.mock_object(volume_api.API, 'get', fake_volume_get)
 1403         self.mock_object(db.sqlalchemy.api, '_volume_type_get_full',
 1404                          v2_fakes.fake_volume_type_get)
 1405 
 1406         req = fakes.HTTPRequest.blank('/v2/volumes/%s' % fake.VOLUME_ID)
 1407         res_dict = self.controller.show(req, fake.VOLUME_ID)
 1408         self.assertEqual('deleting', res_dict['volume']['status'])
 1409 
 1410     @mock.patch.object(volume_api.API, 'delete', v2_fakes.fake_volume_delete)
 1411     @mock.patch.object(volume_api.API, 'get', v2_fakes.fake_volume_get)
 1412     def test_volume_delete(self):
 1413         req = fakes.HTTPRequest.blank('/v2/volumes/%s' % fake.VOLUME_ID)
 1414         resp = self.controller.delete(req, fake.VOLUME_ID)
 1415         self.assertEqual(http_client.ACCEPTED, resp.status_int)
 1416 
 1417     def test_volume_delete_attached(self):
 1418         def fake_volume_attached(self, context, volume,
 1419                                  force=False, cascade=False):
 1420             raise exception.VolumeAttached(volume_id=volume['id'])
 1421         self.mock_object(volume_api.API, "delete", fake_volume_attached)
 1422         self.mock_object(volume_api.API, 'get', v2_fakes.fake_volume_get)
 1423 
 1424         req = fakes.HTTPRequest.blank('/v2/volumes/%s' % fake.VOLUME_ID)
 1425         exp = self.assertRaises(exception.VolumeAttached,
 1426                                 self.controller.delete,
 1427                                 req, 1)
 1428         expect_msg = "Volume 1 is still attached, detach volume first."
 1429         self.assertEqual(expect_msg, six.text_type(exp))
 1430 
 1431     def test_volume_delete_no_volume(self):
 1432         self.mock_object(volume_api.API, "get",
 1433                          v2_fakes.fake_volume_get_notfound)
 1434 
 1435         req = fakes.HTTPRequest.blank('/v2/volumes/%s' % fake.VOLUME_ID)
 1436         self.assertRaises(exception.VolumeNotFound, self.controller.delete,
 1437                           req, 1)
 1438 
 1439     def test_admin_list_volumes_limited_to_project(self):
 1440         self.mock_object(db, 'volume_get_all_by_project',
 1441                          v2_fakes.fake_volume_get_all_by_project)
 1442 
 1443         req = fakes.HTTPRequest.blank('/v2/%s/volumes' % fake.PROJECT_ID,
 1444                                       use_admin_context=True)
 1445         res = self.controller.index(req)
 1446 
 1447         self.assertIn('volumes', res)
 1448         self.assertEqual(1, len(res['volumes']))
 1449 
 1450     @mock.patch.object(db, 'volume_get_all', v2_fakes.fake_volume_get_all)
 1451     @mock.patch.object(db, 'volume_get_all_by_project',
 1452                        v2_fakes.fake_volume_get_all_by_project)
 1453     def test_admin_list_volumes_all_tenants(self):
 1454         req = fakes.HTTPRequest.blank(
 1455             '/v2/%s/volumes?all_tenants=1' % fake.PROJECT_ID,
 1456             use_admin_context=True)
 1457         res = self.controller.index(req)
 1458         self.assertIn('volumes', res)
 1459         self.assertEqual(3, len(res['volumes']))
 1460 
 1461     @mock.patch.object(db, 'volume_get_all', v2_fakes.fake_volume_get_all)
 1462     @mock.patch.object(db, 'volume_get_all_by_project',
 1463                        v2_fakes.fake_volume_get_all_by_project)
 1464     @mock.patch.object(volume_api.API, 'get', v2_fakes.fake_volume_get)
 1465     def test_all_tenants_non_admin_gets_all_tenants(self):
 1466         req = fakes.HTTPRequest.blank(
 1467             '/v2/%s/volumes?all_tenants=1' % fake.PROJECT_ID)
 1468         res = self.controller.index(req)
 1469         self.assertIn('volumes', res)
 1470         self.assertEqual(1, len(res['volumes']))
 1471 
 1472     @mock.patch.object(db, 'volume_get_all_by_project',
 1473                        v2_fakes.fake_volume_get_all_by_project)
 1474     @mock.patch.object(volume_api.API, 'get', v2_fakes.fake_volume_get)
 1475     def test_non_admin_get_by_project(self):
 1476         req = fakes.HTTPRequest.blank('/v2/%s/volumes' % fake.PROJECT_ID)
 1477         res = self.controller.index(req)
 1478         self.assertIn('volumes', res)
 1479         self.assertEqual(1, len(res['volumes']))
 1480 
 1481     def _create_volume_bad_request(self, body):
 1482         req = fakes.HTTPRequest.blank('/v2/%s/volumes' % fake.PROJECT_ID)
 1483         req.method = 'POST'
 1484 
 1485         self.assertRaises(exception.ValidationError,
 1486                           self.controller.create, req, body=body)
 1487 
 1488     def test_create_no_body(self):
 1489         self._create_volume_bad_request(body=None)
 1490 
 1491     def test_create_missing_volume(self):
 1492         body = {'foo': {'a': 'b'}}
 1493         self._create_volume_bad_request(body=body)
 1494 
 1495     def test_create_malformed_entity(self):
 1496         body = {'volume': 'string'}
 1497         self._create_volume_bad_request(body=body)
 1498 
 1499     def _test_get_volumes_by_name(self, get_all, display_name):
 1500         req = mock.MagicMock()
 1501         context = mock.Mock()
 1502         req.environ = {'cinder.context': context}
 1503         req.params = {'display_name': display_name}
 1504         self.controller._view_builder.detail_list = mock.Mock()
 1505         self.controller._get_volumes(req, True)
 1506         get_all.assert_called_once_with(
 1507             context, None, CONF.osapi_max_limit,
 1508             sort_keys=['created_at'], sort_dirs=['desc'],
 1509             filters={'display_name': display_name},
 1510             viewable_admin_meta=True, offset=0)
 1511 
 1512     @mock.patch('cinder.volume.api.API.get_all')
 1513     def test_get_volumes_filter_with_string(self, get_all):
 1514         """Test to get a volume with an alpha-numeric display name."""
 1515         self._test_get_volumes_by_name(get_all, 'Volume-573108026')
 1516 
 1517     @mock.patch('cinder.volume.api.API.get_all')
 1518     def test_get_volumes_filter_with_double_quoted_string(self, get_all):
 1519         """Test to get a volume with a double-quoted display name."""
 1520         self._test_get_volumes_by_name(get_all, '"Volume-573108026"')
 1521 
 1522     @mock.patch('cinder.volume.api.API.get_all')
 1523     def test_get_volumes_filter_with_single_quoted_string(self, get_all):
 1524         """Test to get a volume with a single-quoted display name."""
 1525         self._test_get_volumes_by_name(get_all, "'Volume-573108026'")
 1526 
 1527     @mock.patch('cinder.volume.api.API.get_all')
 1528     def test_get_volumes_filter_with_quote_in_between_string(self, get_all):
 1529         """Test to get a volume with a quote in between the display name."""
 1530         self._test_get_volumes_by_name(get_all, 'Volu"me-573108026')
 1531 
 1532     @mock.patch('cinder.volume.api.API.get_all')
 1533     def test_get_volumes_filter_with_mixed_quoted_string(self, get_all):
 1534         """Test to get a volume with a mix of single and double quotes. """
 1535         # The display name starts with a single quote and ends with a
 1536         # double quote
 1537         self._test_get_volumes_by_name(get_all, '\'Volume-573108026"')
 1538 
 1539     @mock.patch('cinder.volume.api.API.get_all')
 1540     def test_get_volumes_filter_with_true(self, get_all):
 1541         req = mock.MagicMock()
 1542         context = mock.Mock()
 1543         req.environ = {'cinder.context': context}
 1544         req.params = {'display_name': 'Volume-573108026', 'bootable': 1}
 1545         self.controller._view_builder.detail_list = mock.Mock()
 1546         self.controller._get_volumes(req, True)
 1547         get_all.assert_called_once_with(
 1548             context, None, CONF.osapi_max_limit,
 1549             sort_keys=['created_at'], sort_dirs=['desc'],
 1550             filters={'display_name': 'Volume-573108026', 'bootable': True},
 1551             viewable_admin_meta=True, offset=0)
 1552 
 1553     @mock.patch('cinder.volume.api.API.get_all')
 1554     def test_get_volumes_filter_with_false(self, get_all):
 1555         req = mock.MagicMock()
 1556         context = mock.Mock()
 1557         req.environ = {'cinder.context': context}
 1558         req.params = {'display_name': 'Volume-573108026', 'bootable': 0}
 1559         self.controller._view_builder.detail_list = mock.Mock()
 1560         self.controller._get_volumes(req, True)
 1561         get_all.assert_called_once_with(
 1562             context, None, CONF.osapi_max_limit,
 1563             sort_keys=['created_at'], sort_dirs=['desc'],
 1564             filters={'display_name': 'Volume-573108026', 'bootable': False},
 1565             viewable_admin_meta=True, offset=0)
 1566 
 1567     @mock.patch('cinder.volume.api.API.get_all')
 1568     def test_get_volumes_filter_with_list(self, get_all):
 1569         req = mock.MagicMock()
 1570         context = mock.Mock()
 1571         req.environ = {'cinder.context': context}
 1572         req.params = {'id': "['%s', '%s', '%s']" % (
 1573             fake.VOLUME_ID, fake.VOLUME2_ID, fake.VOLUME3_ID)}
 1574         self.controller._view_builder.detail_list = mock.Mock()
 1575         self.controller._get_volumes(req, True)
 1576         get_all.assert_called_once_with(
 1577             context, None, CONF.osapi_max_limit,
 1578             sort_keys=['created_at'], sort_dirs=['desc'],
 1579             filters={'id': [fake.VOLUME_ID, fake.VOLUME2_ID, fake.VOLUME3_ID]},
 1580             viewable_admin_meta=True,
 1581             offset=0)
 1582 
 1583     @mock.patch('cinder.volume.api.API.get_all')
 1584     def test_get_volumes_filter_with_expression(self, get_all):
 1585         req = mock.MagicMock()
 1586         context = mock.Mock()
 1587         req.environ = {'cinder.context': context}
 1588         req.params = {'name': "d-"}
 1589         self.controller._view_builder.detail_list = mock.Mock()
 1590         self.controller._get_volumes(req, True)
 1591         get_all.assert_called_once_with(
 1592             context, None, CONF.osapi_max_limit,
 1593             sort_keys=['created_at'], sort_dirs=['desc'],
 1594             filters={'display_name': 'd-'}, viewable_admin_meta=True, offset=0)
 1595 
 1596     @mock.patch('cinder.volume.api.API.get_all')
 1597     def test_get_volumes_filter_with_status(self, get_all):
 1598         req = mock.MagicMock()
 1599         ctxt = context.RequestContext(
 1600             fake.USER_ID, fake.PROJECT_ID, auth_token=True)
 1601         req.environ = {'cinder.context': ctxt}
 1602         req.params = {'status': 'available'}
 1603         self.controller._view_builder.detail_list = mock.Mock()
 1604         self.controller._get_volumes(req, True)
 1605         get_all.assert_called_once_with(
 1606             ctxt, None, CONF.osapi_max_limit,
 1607             sort_keys=['created_at'], sort_dirs=['desc'],
 1608             filters={'status': 'available'}, viewable_admin_meta=True,
 1609             offset=0)
 1610 
 1611     @mock.patch('cinder.volume.api.API.get_all')
 1612     def test_get_volumes_filter_with_metadata(self, get_all):
 1613         req = mock.MagicMock()
 1614         ctxt = context.RequestContext(
 1615             fake.USER_ID, fake.PROJECT_ID, auth_token=True)
 1616         req.environ = {'cinder.context': ctxt}
 1617         req.params = {'metadata': "{'fake_key': 'fake_value'}"}
 1618         self.controller._view_builder.detail_list = mock.Mock()
 1619         self.controller._get_volumes(req, True)
 1620         get_all.assert_called_once_with(
 1621             ctxt, None, CONF.osapi_max_limit,
 1622             sort_keys=['created_at'], sort_dirs=['desc'],
 1623             filters={'metadata': {'fake_key': 'fake_value'}},
 1624             viewable_admin_meta=True, offset=0)
 1625 
 1626     @mock.patch('cinder.volume.api.API.get_all')
 1627     def test_get_volumes_filter_with_availability_zone(self, get_all):
 1628         req = mock.MagicMock()
 1629         ctxt = context.RequestContext(
 1630             fake.USER_ID, fake.PROJECT_ID, auth_token=True)
 1631         req.environ = {'cinder.context': ctxt}
 1632         req.params = {'availability_zone': 'nova'}
 1633         self.controller._view_builder.detail_list = mock.Mock()
 1634         self.controller._get_volumes(req, True)
 1635         get_all.assert_called_once_with(
 1636             ctxt, None, CONF.osapi_max_limit,
 1637             sort_keys=['created_at'], sort_dirs=['desc'],
 1638             filters={'availability_zone': 'nova'}, viewable_admin_meta=True,
 1639             offset=0)
 1640 
 1641     @mock.patch('cinder.volume.api.API.get_all')
 1642     def test_get_volumes_filter_with_bootable(self, get_all):
 1643         req = mock.MagicMock()
 1644         ctxt = context.RequestContext(
 1645             fake.USER_ID, fake.PROJECT_ID, auth_token=True)
 1646         req.environ = {'cinder.context': ctxt}
 1647         req.params = {'bootable': 1}
 1648         self.controller._view_builder.detail_list = mock.Mock()
 1649         self.controller._get_volumes(req, True)
 1650         get_all.assert_called_once_with(
 1651             ctxt, None, CONF.osapi_max_limit,
 1652             sort_keys=['created_at'], sort_dirs=['desc'],
 1653             filters={'bootable': True}, viewable_admin_meta=True,
 1654             offset=0)
 1655 
 1656     @mock.patch('cinder.volume.api.API.get_all')
 1657     def test_get_volumes_filter_with_invalid_filter(self, get_all):
 1658         req = mock.MagicMock()
 1659         ctxt = context.RequestContext(
 1660             fake.USER_ID, fake.PROJECT_ID, auth_token=True)
 1661         req.environ = {'cinder.context': ctxt}
 1662         req.params = {'invalid_filter': 'invalid',
 1663                       'availability_zone': 'nova'}
 1664         self.controller._view_builder.detail_list = mock.Mock()
 1665         self.controller._get_volumes(req, True)
 1666         get_all.assert_called_once_with(
 1667             ctxt, None, CONF.osapi_max_limit,
 1668             sort_keys=['created_at'], sort_dirs=['desc'],
 1669             filters={'availability_zone': 'nova'}, viewable_admin_meta=True,
 1670             offset=0)
 1671 
 1672     @mock.patch('cinder.volume.api.API.get_all')
 1673     def test_get_volumes_sort_by_name(self, get_all):
 1674         """Name in client means display_name in database."""
 1675 
 1676         req = mock.MagicMock()
 1677         ctxt = context.RequestContext(
 1678             fake.USER_ID, fake.PROJECT_ID, auth_token=True)
 1679         req.environ = {'cinder.context': ctxt}
 1680         req.params = {'sort': 'name'}
 1681         self.controller._view_builder.detail_list = mock.Mock()
 1682         self.controller._get_volumes(req, True)
 1683         get_all.assert_called_once_with(
 1684             ctxt, None, CONF.osapi_max_limit,
 1685             sort_dirs=['desc'], viewable_admin_meta=True,
 1686             sort_keys=['display_name'], filters={}, offset=0)
 1687 
 1688     def test_get_volume_filter_options_using_config(self):
 1689         filter_list = ['name', 'status', 'metadata', 'bootable',
 1690                        'availability_zone']
 1691         self.override_config('query_volume_filters', filter_list)
 1692         self.assertEqual(filter_list,
 1693                          self.controller._get_volume_filter_options())