"Fossies" - the Fresh Open Source Software Archive

Member "cinder-13.0.7/cinder/tests/unit/api/v2/test_types.py" (4 Oct 2019, 20708 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_types.py": 14.0.2_vs_15.0.0.

    1 # Copyright 2011 OpenStack Foundation
    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 import uuid
   17 
   18 import mock
   19 from oslo_utils import timeutils
   20 import six
   21 import webob
   22 
   23 from cinder.api.v2 import types
   24 from cinder.api.v2.views import types as views_types
   25 from cinder import context
   26 from cinder import exception
   27 from cinder.policies import volume_type as type_policy
   28 from cinder import test
   29 from cinder.tests.unit.api import fakes
   30 from cinder.tests.unit import fake_constants as fake
   31 from cinder.volume import volume_types
   32 
   33 
   34 def fake_volume_type(id):
   35     specs = {
   36         "key1": "value1",
   37         "key2": "value2",
   38         "key3": "value3",
   39         "key4": "value4",
   40         "key5": "value5"
   41     }
   42     return dict(
   43         id=id,
   44         name='vol_type_%s' % six.text_type(id),
   45         description='vol_type_desc_%s' % six.text_type(id),
   46         extra_specs=specs,
   47     )
   48 
   49 
   50 def return_volume_types_get_all_types(context, filters=None, marker=None,
   51                                       limit=None, sort_keys=None,
   52                                       sort_dirs=None, offset=None,
   53                                       list_result=False):
   54     result = dict(vol_type_1=fake_volume_type(1),
   55                   vol_type_2=fake_volume_type(2),
   56                   vol_type_3=fake_volume_type(3)
   57                   )
   58     if list_result:
   59         return list(result.values())
   60     return result
   61 
   62 
   63 def return_empty_volume_types_get_all_types(context, filters=None, marker=None,
   64                                             limit=None, sort_keys=None,
   65                                             sort_dirs=None, offset=None,
   66                                             list_result=False):
   67     if list_result:
   68         return []
   69     return {}
   70 
   71 
   72 def return_volume_types_get_volume_type(context, id):
   73     if id == fake.WILL_NOT_BE_FOUND_ID:
   74         raise exception.VolumeTypeNotFound(volume_type_id=id)
   75     return fake_volume_type(id)
   76 
   77 
   78 def return_volume_types_get_default():
   79     return fake_volume_type(1)
   80 
   81 
   82 class VolumeTypesApiTest(test.TestCase):
   83 
   84     def _create_volume_type(self, volume_type_name, extra_specs=None,
   85                             is_public=True, projects=None):
   86         return volume_types.create(self.ctxt, volume_type_name, extra_specs,
   87                                    is_public, projects).get('id')
   88 
   89     def setUp(self):
   90         super(VolumeTypesApiTest, self).setUp()
   91         self.controller = types.VolumeTypesController()
   92         self.ctxt = context.RequestContext(user_id=fake.USER_ID,
   93                                            project_id=fake.PROJECT_ID,
   94                                            is_admin=True)
   95         self.mock_authorize = self.patch(
   96             'cinder.context.RequestContext.authorize')
   97         self.type_id1 = self._create_volume_type('volume_type1',
   98                                                  {'key1': 'value1'})
   99         self.type_id2 = self._create_volume_type('volume_type2',
  100                                                  {'key2': 'value2'})
  101         self.type_id3 = self._create_volume_type('volume_type3',
  102                                                  {'key3': 'value3'}, False,
  103                                                  [fake.PROJECT_ID])
  104 
  105     def test_volume_types_index(self):
  106         self.mock_object(volume_types, 'get_all_types',
  107                          return_volume_types_get_all_types)
  108 
  109         req = fakes.HTTPRequest.blank('/v2/%s/types' % fake.PROJECT_ID,
  110                                       use_admin_context=True)
  111         res_dict = self.controller.index(req)
  112 
  113         self.assertEqual(3, len(res_dict['volume_types']))
  114 
  115         expected_names = ['vol_type_1', 'vol_type_2', 'vol_type_3']
  116         actual_names = map(lambda e: e['name'], res_dict['volume_types'])
  117         self.assertEqual(set(expected_names), set(actual_names))
  118         for entry in res_dict['volume_types']:
  119             self.assertEqual('value1', entry['extra_specs']['key1'])
  120         self.mock_authorize.assert_any_call(type_policy.GET_ALL_POLICY)
  121 
  122     def test_volume_types_index_no_data(self):
  123         self.mock_object(volume_types, 'get_all_types',
  124                          return_empty_volume_types_get_all_types)
  125 
  126         req = fakes.HTTPRequest.blank('/v2/%s/types' % fake.PROJECT_ID)
  127         res_dict = self.controller.index(req)
  128 
  129         self.assertEqual(0, len(res_dict['volume_types']))
  130 
  131     def test_volume_types_index_with_limit(self):
  132         req = fakes.HTTPRequest.blank('/v2/%s/types?limit=1' % fake.PROJECT_ID)
  133         req.environ['cinder.context'] = self.ctxt
  134         res = self.controller.index(req)
  135 
  136         self.assertEqual(1, len(res['volume_types']))
  137         self.assertEqual(self.type_id3, res['volume_types'][0]['id'])
  138 
  139         expect_next_link = ('http://localhost/v2/%s/types?limit=1'
  140                             '&marker=%s' %
  141                             (fake.PROJECT_ID, res['volume_types'][0]['id']))
  142         self.assertEqual(expect_next_link, res['volume_type_links'][0]['href'])
  143 
  144     def test_volume_types_index_with_offset(self):
  145         req = fakes.HTTPRequest.blank(
  146             '/v2/%s/types?offset=1' % fake.PROJECT_ID)
  147         req.environ['cinder.context'] = self.ctxt
  148         res = self.controller.index(req)
  149 
  150         self.assertEqual(2, len(res['volume_types']))
  151 
  152     def test_volume_types_index_with_offset_out_of_range(self):
  153         url = '/v2/%s/types?offset=424366766556787' % fake.PROJECT_ID
  154         req = fakes.HTTPRequest.blank(url)
  155         self.assertRaises(webob.exc.HTTPBadRequest,
  156                           self.controller.index, req)
  157 
  158     def test_volume_types_index_with_limit_and_offset(self):
  159         req = fakes.HTTPRequest.blank(
  160             '/v2/%s/types?limit=2&offset=1' % fake.PROJECT_ID)
  161         req.environ['cinder.context'] = self.ctxt
  162         res = self.controller.index(req)
  163 
  164         self.assertEqual(2, len(res['volume_types']))
  165         self.assertEqual(self.type_id2, res['volume_types'][0]['id'])
  166         self.assertEqual(self.type_id1, res['volume_types'][1]['id'])
  167 
  168     def test_volume_types_index_with_limit_and_marker(self):
  169         req = fakes.HTTPRequest.blank('/v2/%s/types?limit=1'
  170                                       '&marker=%s' %
  171                                       (fake.PROJECT_ID,
  172                                        self.type_id2))
  173         req.environ['cinder.context'] = self.ctxt
  174         res = self.controller.index(req)
  175 
  176         self.assertEqual(1, len(res['volume_types']))
  177         self.assertEqual(self.type_id1, res['volume_types'][0]['id'])
  178 
  179     def test_volume_types_index_with_valid_filter(self):
  180         req = fakes.HTTPRequest.blank(
  181             '/v2/%s/types?is_public=True' % fake.PROJECT_ID)
  182         req.environ['cinder.context'] = self.ctxt
  183         res = self.controller.index(req)
  184 
  185         self.assertEqual(3, len(res['volume_types']))
  186         self.assertEqual(self.type_id3, res['volume_types'][0]['id'])
  187         self.assertEqual(self.type_id2, res['volume_types'][1]['id'])
  188         self.assertEqual(self.type_id1, res['volume_types'][2]['id'])
  189 
  190     def test_volume_types_index_with_invalid_filter(self):
  191         req = fakes.HTTPRequest.blank(
  192             '/v2/%s/types?id=%s' % (fake.PROJECT_ID, self.type_id1))
  193         req.environ['cinder.context'] = context.RequestContext(
  194             user_id=fake.USER_ID, project_id=fake.PROJECT_ID, is_admin=False)
  195         res = self.controller.index(req)
  196 
  197         self.assertEqual(3, len(res['volume_types']))
  198 
  199     def test_volume_types_index_with_sort_keys(self):
  200         req = fakes.HTTPRequest.blank('/v2/%s/types?sort=id' % fake.PROJECT_ID)
  201         req.environ['cinder.context'] = self.ctxt
  202         res = self.controller.index(req)
  203         expect_result = [self.type_id1, self.type_id2, self.type_id3]
  204         expect_result.sort(reverse=True)
  205 
  206         self.assertEqual(3, len(res['volume_types']))
  207         self.assertEqual(expect_result[0], res['volume_types'][0]['id'])
  208         self.assertEqual(expect_result[1], res['volume_types'][1]['id'])
  209         self.assertEqual(expect_result[2], res['volume_types'][2]['id'])
  210 
  211     def test_volume_types_index_with_sort_and_limit(self):
  212         req = fakes.HTTPRequest.blank(
  213             '/v2/%s/types?sort=id&limit=2' % fake.PROJECT_ID)
  214         req.environ['cinder.context'] = self.ctxt
  215         res = self.controller.index(req)
  216         expect_result = [self.type_id1, self.type_id2, self.type_id3]
  217         expect_result.sort(reverse=True)
  218 
  219         self.assertEqual(2, len(res['volume_types']))
  220         self.assertEqual(expect_result[0], res['volume_types'][0]['id'])
  221         self.assertEqual(expect_result[1], res['volume_types'][1]['id'])
  222 
  223     def test_volume_types_index_with_sort_keys_and_sort_dirs(self):
  224         req = fakes.HTTPRequest.blank(
  225             '/v2/%s/types?sort=id:asc' % fake.PROJECT_ID)
  226         req.environ['cinder.context'] = self.ctxt
  227         res = self.controller.index(req)
  228         expect_result = [self.type_id1, self.type_id2, self.type_id3]
  229         expect_result.sort()
  230 
  231         self.assertEqual(3, len(res['volume_types']))
  232         self.assertEqual(expect_result[0], res['volume_types'][0]['id'])
  233         self.assertEqual(expect_result[1], res['volume_types'][1]['id'])
  234         self.assertEqual(expect_result[2], res['volume_types'][2]['id'])
  235 
  236     def test_volume_types_show(self):
  237         self.mock_object(volume_types, 'get_volume_type',
  238                          return_volume_types_get_volume_type)
  239 
  240         type_id = str(uuid.uuid4())
  241         req = fakes.HTTPRequest.blank('/v2/%s/types/' % fake.PROJECT_ID
  242                                       + type_id)
  243         res_dict = self.controller.show(req, type_id)
  244 
  245         self.assertEqual(1, len(res_dict))
  246         self.assertEqual(type_id, res_dict['volume_type']['id'])
  247         type_name = 'vol_type_' + type_id
  248         self.assertEqual(type_name, res_dict['volume_type']['name'])
  249         self.mock_authorize.assert_any_call(
  250             type_policy.GET_POLICY, target_obj=mock.ANY)
  251 
  252     def test_volume_types_show_not_found(self):
  253         self.mock_object(volume_types, 'get_volume_type',
  254                          return_volume_types_get_volume_type)
  255 
  256         req = fakes.HTTPRequest.blank('/v2/%s/types/%s' %
  257                                       (fake.PROJECT_ID,
  258                                        fake.WILL_NOT_BE_FOUND_ID))
  259         self.assertRaises(exception.VolumeTypeNotFound, self.controller.show,
  260                           req, fake.WILL_NOT_BE_FOUND_ID)
  261 
  262     def test_get_default(self):
  263         self.mock_object(volume_types, 'get_default_volume_type',
  264                          return_volume_types_get_default)
  265         req = fakes.HTTPRequest.blank('/v2/%s/types/default' % fake.PROJECT_ID)
  266         req.method = 'GET'
  267         res_dict = self.controller.show(req, 'default')
  268         self.assertEqual(1, len(res_dict))
  269         self.assertEqual('vol_type_1', res_dict['volume_type']['name'])
  270         self.assertEqual('vol_type_desc_1',
  271                          res_dict['volume_type']['description'])
  272 
  273     def test_get_default_not_found(self):
  274         self.mock_object(volume_types, 'get_default_volume_type',
  275                          return_value={})
  276         req = fakes.HTTPRequest.blank('/v2/%s/types/default' % fake.PROJECT_ID)
  277         req.method = 'GET'
  278 
  279         self.assertRaises(exception.VolumeTypeNotFound,
  280                           self.controller.show, req, 'default')
  281 
  282     def test_view_builder_show(self):
  283         view_builder = views_types.ViewBuilder()
  284         self.mock_authorize.return_value = False
  285         now = timeutils.utcnow().isoformat()
  286         raw_volume_type = dict(
  287             name='new_type',
  288             description='new_type_desc',
  289             qos_specs_id='new_id',
  290             is_public=True,
  291             deleted=False,
  292             created_at=now,
  293             updated_at=now,
  294             extra_specs={},
  295             deleted_at=None,
  296             id=42,
  297         )
  298 
  299         request = fakes.HTTPRequest.blank("/v2")
  300         output = view_builder.show(request, raw_volume_type)
  301 
  302         self.assertIn('volume_type', output)
  303         expected_volume_type = dict(
  304             name='new_type',
  305             description='new_type_desc',
  306             is_public=True,
  307             id=42,
  308         )
  309         self.assertDictEqual(expected_volume_type, output['volume_type'])
  310 
  311     def test_view_builder_show_admin(self):
  312         view_builder = views_types.ViewBuilder()
  313         self.mock_authorize.return_value = True
  314         now = timeutils.utcnow().isoformat()
  315         raw_volume_type = dict(
  316             name='new_type',
  317             description='new_type_desc',
  318             qos_specs_id='new_id',
  319             is_public=True,
  320             deleted=False,
  321             created_at=now,
  322             updated_at=now,
  323             extra_specs={},
  324             deleted_at=None,
  325             id=42,
  326         )
  327 
  328         request = fakes.HTTPRequest.blank("/v2", use_admin_context=True)
  329         output = view_builder.show(request, raw_volume_type)
  330 
  331         self.assertIn('volume_type', output)
  332         expected_volume_type = dict(
  333             name='new_type',
  334             description='new_type_desc',
  335             qos_specs_id='new_id',
  336             is_public=True,
  337             extra_specs={},
  338             id=42,
  339         )
  340         self.assertDictEqual(expected_volume_type, output['volume_type'])
  341 
  342     def test_view_builder_show_qos_specs_id_policy(self):
  343         with mock.patch('cinder.context.RequestContext.authorize',
  344                         side_effect=[False, True]):
  345             view_builder = views_types.ViewBuilder()
  346             now = timeutils.utcnow().isoformat()
  347             raw_volume_type = dict(
  348                 name='new_type',
  349                 description='new_type_desc',
  350                 qos_specs_id='new_id',
  351                 is_public=True,
  352                 deleted=False,
  353                 created_at=now,
  354                 updated_at=now,
  355                 extra_specs={},
  356                 deleted_at=None,
  357                 id=42,
  358             )
  359 
  360             request = fakes.HTTPRequest.blank("/v2")
  361             output = view_builder.show(request, raw_volume_type)
  362 
  363             self.assertIn('volume_type', output)
  364             expected_volume_type = dict(
  365                 name='new_type',
  366                 description='new_type_desc',
  367                 qos_specs_id='new_id',
  368                 is_public=True,
  369                 id=42,
  370             )
  371             self.assertDictEqual(expected_volume_type, output['volume_type'])
  372 
  373     def test_view_builder_show_extra_specs_policy(self):
  374         with mock.patch('cinder.context.RequestContext.authorize',
  375                         side_effect=[True, False]):
  376             view_builder = views_types.ViewBuilder()
  377             now = timeutils.utcnow().isoformat()
  378             raw_volume_type = dict(
  379                 name='new_type',
  380                 description='new_type_desc',
  381                 qos_specs_id='new_id',
  382                 is_public=True,
  383                 deleted=False,
  384                 created_at=now,
  385                 updated_at=now,
  386                 extra_specs={},
  387                 deleted_at=None,
  388                 id=42,
  389             )
  390 
  391             request = fakes.HTTPRequest.blank("/v2")
  392             output = view_builder.show(request, raw_volume_type)
  393 
  394             self.assertIn('volume_type', output)
  395             expected_volume_type = dict(
  396                 name='new_type',
  397                 description='new_type_desc',
  398                 extra_specs={},
  399                 is_public=True,
  400                 id=42,
  401             )
  402             self.assertDictEqual(expected_volume_type, output['volume_type'])
  403 
  404         with mock.patch('cinder.context.RequestContext.authorize',
  405                         side_effect=[False, False]):
  406             view_builder = views_types.ViewBuilder()
  407             now = timeutils.utcnow().isoformat()
  408             raw_volume_type = dict(
  409                 name='new_type',
  410                 description='new_type_desc',
  411                 qos_specs_id='new_id',
  412                 is_public=True,
  413                 deleted=False,
  414                 created_at=now,
  415                 updated_at=now,
  416                 extra_specs={},
  417                 deleted_at=None,
  418                 id=42,
  419             )
  420 
  421             request = fakes.HTTPRequest.blank("/v2")
  422             output = view_builder.show(request, raw_volume_type)
  423 
  424             self.assertIn('volume_type', output)
  425             expected_volume_type = dict(
  426                 name='new_type',
  427                 description='new_type_desc',
  428                 is_public=True,
  429                 id=42,
  430             )
  431             self.assertDictEqual(expected_volume_type, output['volume_type'])
  432 
  433     def test_view_builder_show_pass_all_policy(self):
  434         with mock.patch('cinder.context.RequestContext.authorize',
  435                         side_effect=[True, True]):
  436             view_builder = views_types.ViewBuilder()
  437             now = timeutils.utcnow().isoformat()
  438             raw_volume_type = dict(
  439                 name='new_type',
  440                 description='new_type_desc',
  441                 qos_specs_id='new_id',
  442                 is_public=True,
  443                 deleted=False,
  444                 created_at=now,
  445                 updated_at=now,
  446                 extra_specs={},
  447                 deleted_at=None,
  448                 id=42,
  449             )
  450 
  451             request = fakes.HTTPRequest.blank("/v2")
  452             output = view_builder.show(request, raw_volume_type)
  453 
  454             self.assertIn('volume_type', output)
  455             expected_volume_type = dict(
  456                 name='new_type',
  457                 description='new_type_desc',
  458                 qos_specs_id='new_id',
  459                 extra_specs={},
  460                 is_public=True,
  461                 id=42,
  462             )
  463             self.assertDictEqual(expected_volume_type, output['volume_type'])
  464 
  465     def test_view_builder_list(self):
  466         view_builder = views_types.ViewBuilder()
  467         self.mock_authorize.return_value = False
  468         now = timeutils.utcnow().isoformat()
  469         raw_volume_types = []
  470         for i in range(0, 10):
  471             raw_volume_types.append(
  472                 dict(
  473                     name='new_type',
  474                     description='new_type_desc',
  475                     qos_specs_id='new_id',
  476                     is_public=True,
  477                     deleted=False,
  478                     created_at=now,
  479                     updated_at=now,
  480                     extra_specs={},
  481                     deleted_at=None,
  482                     id=42 + i
  483                 )
  484             )
  485 
  486         request = fakes.HTTPRequest.blank("/v2")
  487         output = view_builder.index(request, raw_volume_types)
  488 
  489         self.assertIn('volume_types', output)
  490         for i in range(0, 10):
  491             expected_volume_type = dict(
  492                 name='new_type',
  493                 description='new_type_desc',
  494                 is_public=True,
  495                 id=42 + i
  496             )
  497             self.assertDictEqual(expected_volume_type,
  498                                  output['volume_types'][i])
  499 
  500     def test_view_builder_list_admin(self):
  501         view_builder = views_types.ViewBuilder()
  502 
  503         now = timeutils.utcnow().isoformat()
  504         raw_volume_types = []
  505         for i in range(0, 10):
  506             raw_volume_types.append(
  507                 dict(
  508                     name='new_type',
  509                     description='new_type_desc',
  510                     qos_specs_id='new_id',
  511                     is_public=True,
  512                     deleted=False,
  513                     created_at=now,
  514                     updated_at=now,
  515                     extra_specs={},
  516                     deleted_at=None,
  517                     id=42 + i
  518                 )
  519             )
  520 
  521         request = fakes.HTTPRequest.blank("/v2", use_admin_context=True)
  522         output = view_builder.index(request, raw_volume_types)
  523 
  524         self.assertIn('volume_types', output)
  525         for i in range(0, 10):
  526             expected_volume_type = dict(
  527                 name='new_type',
  528                 description='new_type_desc',
  529                 qos_specs_id='new_id',
  530                 is_public=True,
  531                 extra_specs={},
  532                 id=42 + i
  533             )
  534             self.assertDictEqual(expected_volume_type,
  535                                  output['volume_types'][i])